Add SynBioHub PartShop client, collection pulling, and Golden Gate digestion/ligation simulation

README.md

@@ -179,6 +179,34 @@ result = full_build(
 
 `BuildCompiler.__init__` should stay lightweight and dependency-injected. Automatic SynBioHub collection indexing belongs in `BuildCompiler.from_synbiohub(...)`.
 
+### SynBioHub repository authentication
+
+`BuildCompiler.from_synbiohub(...)` supports anonymous pulls, auth-token reuse, or login via email/password:
+
+```python
+from buildcompiler.api import BuildCompiler
+
+# Anonymous/public access
+public_compiler = BuildCompiler.from_synbiohub(
+    repository_url="https://synbiohub.org",
+    collections=["https://synbiohub.org/public/igem"],
+)
+
+# Existing auth token
+token_compiler = BuildCompiler.from_synbiohub(
+    repository_url="https://synbiohub.org",
+    auth_token="<token>",
+    collections=["https://synbiohub.org/user/private_collection"],
+)
+
+# Email/password login (token kept in-memory)
+login_compiler = BuildCompiler.from_synbiohub(
+    repository_url="https://synbiohub.org",
+    email="user@example.org",
+    password="<password>",
+)
+```
+
 ## Local development
 
 Recommended local workflow:

notebooks/buildcompiler_synbiohub_tutorial.ipynb

@@ -0,0 +1,105 @@
+{
+  "cells": [
+    {
+      "cell_type": "markdown",
+      "metadata": {},
+      "source": [
+        "# BuildCompiler SynBioHub Tutorial\n",
+        "\n",
+        "This notebook shows how to authenticate to SynBioHub, load private collections into BuildCompiler, and run a full-build workflow using an abstract design URI.\n"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "from getpass import getpass\n",
+        "import sbol2\n",
+        "\n",
+        "from buildcompiler.api import BuildCompiler\n"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "repository_url = \"https://synbiohub.org\"\n",
+        "user = \"user\"\n",
+        "password = getpass(\"SynBioHub password: \")\n",
+        "\n",
+        "abstract_design_uri = \"https://synbiohub.org/user/Gon/abstract_design/standard_GFP/1\"\n",
+        "collections = [\n",
+        "    \"https://synbiohub.org/user/Gon/impl_test/impl_test_collection/1\",\n",
+        "    \"https://synbiohub.org/user/Gon/Enzyme_Implementations/Enzyme_Implementations_collection/1\",\n",
+        "]\n"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "sbol_doc = sbol2.Document()\n",
+        "\n",
+        "compiler = BuildCompiler.from_synbiohub(\n",
+        "    repository_url=repository_url,\n",
+        "    email=user,\n",
+        "    password=password,\n",
+        "    collections=collections,\n",
+        "    sbol_doc=sbol_doc,\n",
+        ")\n",
+        "\n",
+        "print(f\"Loaded SBOL objects: {len(list(sbol_doc))}\")\n",
+        "print(f\"Indexed plasmids: {len(getattr(compiler.inventory, 'plasmids', [])) if compiler.inventory else 'n/a'}\")\n"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "# Pull/resolve the abstract design into the shared document.\n",
+        "if compiler.repository_client is not None:\n",
+        "    compiler.repository_client.pull_identity(abstract_design_uri)\n",
+        "\n",
+        "abstract_design = sbol_doc.find(abstract_design_uri)\n",
+        "if abstract_design is None:\n",
+        "    raise LookupError(f\"Could not resolve abstract design: {abstract_design_uri}\")\n",
+        "\n",
+        "plan = compiler.plan([abstract_design])\n",
+        "result = compiler.execute(plan)\n",
+        "result\n"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {},
+      "source": [
+        "## Notes\n",
+        "\n",
+        "- BuildCompiler reuses a single authenticated `sbol2.PartShop` client for collection pulls and identity fallback pulls.\n",
+        "- Tokens returned by SynBioHub login are kept in memory only; they are not serialized by default.\n",
+        "- Assembly uses Golden Gate digestion/ligation simulation and links reagent usages + generated products through a stage activity in SBOL.\n"
+      ]
+    }
+  ],
+  "metadata": {
+    "kernelspec": {
+      "display_name": "Python 3",
+      "language": "python",
+      "name": "python3"
+    },
+    "language_info": {
+      "name": "python",
+      "version": "3.11"
+    }
+  },
+  "nbformat": 4,
+  "nbformat_minor": 5
+}

src/buildcompiler/api/compiler.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from buildcompiler.planning import FullBuildPlanner
+from buildcompiler.sbol import PartShopRepositoryClient
 
 from .options import BuildOptions
 
@@ -17,24 +18,55 @@ class BuildCompiler:
     planner: Any = None
     executor: Any = None
     adapters: Any = None
+    repository_client: PartShopRepositoryClient | None = None
     options: BuildOptions = field(default_factory=BuildOptions)
 
     @classmethod
     def from_synbiohub(
         cls,
         *,
         collections: list[str] | None = None,
+        repository_url: str | None = None,
         sbh_registry: str | None = None,
         auth_token: str | None = None,
+        email: str | None = None,
+        password: str | None = None,
         sbol_doc: Any = None,
         options: BuildOptions | None = None,
         **kwargs: Any,
     ) -> "BuildCompiler":
-        if collections:
-            raise NotImplementedError(
-                "Automatic SynBioHub collection loading/indexing is not implemented yet. Inject inventory dependencies directly for now."
+        resolved_repository_url = repository_url or sbh_registry
+        if auth_token and password:
+            raise ValueError("auth_token cannot be combined with password")
+
+        requires_repository = bool(
+            collections or resolved_repository_url or auth_token or email or password
+        )
+        if requires_repository and not resolved_repository_url:
+            raise ValueError("repository_url (or sbh_registry) is required")
+
+        document = sbol_doc
+        repository_client = None
+        if resolved_repository_url:
+            import sbol2
+
+            document = document or sbol2.Document()
+            repository_client = PartShopRepositoryClient(
+                repository_url=resolved_repository_url,
+                document=document,
+                auth_token=auth_token,
+                email=email,
+                password=password,
             )
-        return cls(sbol_document=sbol_doc, options=options or BuildOptions(), **kwargs)
+            for collection_uri in collections or []:
+                repository_client.pull_collection(collection_uri)
+
+        return cls(
+            sbol_document=document,
+            repository_client=repository_client,
+            options=options or BuildOptions(),
+            **kwargs,
+        )
 
     def plan(self, abstract_designs: Any, options: BuildOptions | None = None) -> Any:
         effective_options = options or self.options
@@ -60,6 +92,11 @@ def execute(self, plan: Any, options: BuildOptions | None = None) -> Any:
                 sbol_document=self.sbol_document,
                 options=effective_options,
                 adapters=self.adapters,
+                pull_client=(
+                    self.repository_client.pull_identity
+                    if self.repository_client is not None
+                    else None
+                ),
             )
         return executor.execute(plan, options=effective_options)
 
@@ -80,22 +117,31 @@ def full_build(
     adapters: Any = None,
     options: BuildOptions | None = None,
     collections: list[str] | None = None,
+    repository_url: str | None = None,
     sbh_registry: str | None = None,
     auth_token: str | None = None,
+    email: str | None = None,
+    password: str | None = None,
     sbol_doc: Any = None,
     **kwargs: Any,
 ) -> Any:
     compiler_options = options or BuildOptions()
     if (
         collections is not None
         or sbh_registry is not None
+        or repository_url is not None
         or auth_token is not None
+        or email is not None
+        or password is not None
         or sbol_doc is not None
     ):
         compiler = BuildCompiler.from_synbiohub(
             collections=collections,
+            repository_url=repository_url,
             sbh_registry=sbh_registry,
             auth_token=auth_token,
+            email=email,
+            password=password,
             sbol_doc=sbol_doc,
             options=compiler_options,
             inventory=inventory,

src/buildcompiler/execution/executor.py

@@ -61,11 +61,12 @@ def from_dependencies(
         sbol_document: Any,
         options: BuildOptions,
         adapters: Any = None,
+        pull_client: Any = None,
         graph: Any = None,
         logger: Any = None,
         **stage_overrides: Any,
     ) -> "FullBuildExecutor":
-        resolver = SbolResolver(sbol_document)
+        resolver = SbolResolver(sbol_document, pull_client=pull_client)
         return cls(
             context=BuildContext(
                 sbol=resolver,

src/buildcompiler/sbol/__init__.py

@@ -2,6 +2,7 @@
 
 from .assembly import AssemblyJob, AssemblySbolResult, AssemblyService
 from .domestication import DomesticationJob, DomesticationSbolResult, DomesticationService
+from .repository import PartShopRepositoryClient
 from .resolver import PullPolicy, SbolResolver
 
 __all__ = [
@@ -11,6 +12,7 @@
     "DomesticationJob",
     "DomesticationSbolResult",
     "DomesticationService",
+    "PartShopRepositoryClient",
     "PullPolicy",
     "SbolResolver",
 ]

src/buildcompiler/sbol/assembly.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass, field
 
 import sbol2
+from Bio import Restriction
+from pydna.dseqrecord import Dseqrecord
 
 from buildcompiler.domain import (
     BuildStage,
@@ -74,6 +76,13 @@ def run(self, job: AssemblyJob) -> AssemblySbolResult:
         )
         ligase_impl = self._implementation_from_record(job.ligase, job.source_document)
 
+        self._simulate_golden_gate(
+            part_components=[p.plasmid_definition for p in legacy_parts],
+            backbone_component=legacy_backbone.plasmid_definition,
+            restriction_impl=restriction_impl,
+            source_document=job.source_document,
+        )
+
         composite_prefix = job.product_display_id or job.product_identity.split("/")[-1]
         legacy_assembly = Assembly(
             part_plasmids=legacy_parts,
@@ -104,6 +113,63 @@ def run(self, job: AssemblyJob) -> AssemblySbolResult:
             logs=logs,
         )
 
+    def _simulate_golden_gate(
+        self,
+        *,
+        part_components: list[sbol2.ComponentDefinition],
+        backbone_component: sbol2.ComponentDefinition,
+        restriction_impl: sbol2.Implementation,
+        source_document: sbol2.Document,
+    ) -> None:
+        enzyme_definition = source_document.find(restriction_impl.built)
+        enzyme_name = getattr(enzyme_definition, "displayId", None) or getattr(
+            enzyme_definition, "name", None
+        )
+        if not enzyme_name or not hasattr(Restriction, enzyme_name):
+            return
+        enzyme = getattr(Restriction, enzyme_name)
+
+        part_inserts = []
+        for component in part_components:
+            fragments = self._digest_component(component, enzyme, source_document)
+            if len(fragments) != 2:
+                raise ValueError(
+                    f"Part plasmid {component.displayId} digestion with {enzyme_name} produced {len(fragments)} fragments; expected 2"
+                )
+            part_inserts.append(min(fragments, key=len))
+
+        backbone_fragments = self._digest_component(
+            backbone_component, enzyme, source_document
+        )
+        if len(backbone_fragments) != 2:
+            raise ValueError(
+                f"Backbone {backbone_component.displayId} digestion with {enzyme_name} produced {len(backbone_fragments)} fragments; expected 2"
+            )
+        open_backbone = max(backbone_fragments, key=len)
+
+        assembled = open_backbone
+        for part_insert in part_inserts:
+            assembled = assembled + part_insert
+        ligated = assembled.looped()
+        if ligated is None or len(ligated) == 0:
+            raise ValueError("Golden Gate ligation failed: expected one circular product")
+
+    def _digest_component(
+        self,
+        component: sbol2.ComponentDefinition,
+        enzyme: Restriction.RestrictionType,
+        source_document: sbol2.Document,
+    ) -> list[Dseqrecord]:
+        if len(component.sequences) != 1:
+            raise ValueError(
+                f"Component {component.displayId} must have exactly one sequence for digestion simulation"
+            )
+        sequence_obj = source_document.find(component.sequences[0])
+        if not isinstance(sequence_obj, sbol2.Sequence):
+            raise ValueError(f"Missing sequence for component {component.displayId}")
+        ds_record = Dseqrecord(sequence_obj.elements, circular=True)
+        return ds_record.cut([enzyme])
+
     def _record_to_legacy_plasmid(
         self,
         record: IndexedPlasmid,