From 7df69b110481064d286e47127c98ec390ed065fe Mon Sep 17 00:00:00 2001 From: Mike Arpaia Date: Wed, 30 Sep 2026 22:22:15 -0600 Subject: [PATCH] Model supplier catalog evidence and explicit material receipts --- scripts/check_install.py | 1 + src/lab/suppliers/__init__.py | 28 +++ src/lab/suppliers/addgene.py | 103 +++++++++++ src/lab/suppliers/types.py | 336 ++++++++++++++++++++++++++++++++++ tests/test_suppliers.py | 103 +++++++++++ 5 files changed, 571 insertions(+) create mode 100644 src/lab/suppliers/__init__.py create mode 100644 src/lab/suppliers/addgene.py create mode 100644 src/lab/suppliers/types.py create mode 100644 tests/test_suppliers.py diff --git a/scripts/check_install.py b/scripts/check_install.py index 6c917e8..b42d76b 100644 --- a/scripts/check_install.py +++ b/scripts/check_install.py @@ -25,6 +25,7 @@ def main() -> None: import_module("lab.samples") import_module("lab.provenance") import_module("lab.inventory") + import_module("lab.suppliers") import_module("lab.labop") package = distribution("lab-compiler") diff --git a/src/lab/suppliers/__init__.py b/src/lab/suppliers/__init__.py new file mode 100644 index 0000000..6d75c31 --- /dev/null +++ b/src/lab/suppliers/__init__.py @@ -0,0 +1,28 @@ +"""Read-only supplier catalogs and explicit purchase/receipt records.""" + +from lab.suppliers.addgene import AddgeneClient, parse_plasmid +from lab.suppliers.types import ( + AcquisitionRequest, + Catalog, + CatalogEntry, + CatalogSequence, + OrderReference, + QuoteRecord, + Receipt, + SequenceSource, + SupplierItem, +) + +__all__ = [ + "AcquisitionRequest", + "AddgeneClient", + "Catalog", + "CatalogEntry", + "CatalogSequence", + "OrderReference", + "QuoteRecord", + "Receipt", + "SequenceSource", + "SupplierItem", + "parse_plasmid", +] diff --git a/src/lab/suppliers/addgene.py b/src/lab/suppliers/addgene.py new file mode 100644 index 0000000..caa21b1 --- /dev/null +++ b/src/lab/suppliers/addgene.py @@ -0,0 +1,103 @@ +"""Explicit, read-only Addgene catalog retrieval. + +Contract: https://docs.developers.addgene.org/docs/schema/ (e285fff01c). +Credentials stay on the client and are never included in snapshots. Catalog +access requires an approved token with the appropriate retrieve scope. +""" + +import json +from dataclasses import dataclass, field +from datetime import UTC, datetime +from email.message import Message +from typing import IO +from urllib.error import HTTPError +from urllib.request import HTTPRedirectHandler, Request, build_opener + +from lab.artifacts import canonical_json +from lab.suppliers.types import CatalogSequence, SequenceSource, SupplierItem + +BASE_URL = "https://api.developers.addgene.org" + + +class _NoRedirect(HTTPRedirectHandler): + # Never forward a credential to a catalog redirect destination. + def redirect_request( + self, req: Request, fp: IO[bytes], code: int, msg: str, headers: Message, newurl: str + ) -> None: + return None + + +def parse_plasmid(payload: str, *, retrieved_at: datetime) -> SupplierItem: + """Parse a saved retrieve response; no network or automatic sequence choice.""" + data = json.loads(payload) + if not isinstance(data, dict) or type(data.get("id")) is not int or data["id"] < 1: + raise ValueError("Expected an Addgene plasmid retrieve response") + plasmid_id = data["id"] + identity = f"https://www.addgene.org/{plasmid_id}/" + sequences = [] + groups = data.get("sequences", {}) + if not isinstance(groups, dict): + raise ValueError("Catalog sequences must be grouped by source and completeness") + for key, source, complete in ( + ("public_user_full_sequences", SequenceSource.DEPOSITOR, True), + ("public_addgene_full_sequences", SequenceSource.ADDGENE, True), + ("public_user_partial_sequences", SequenceSource.DEPOSITOR, False), + ("public_addgene_partial_sequences", SequenceSource.ADDGENE, False), + ): + for row in groups.get(key, []): + if type(row["sequence_id"]) is not int or not isinstance(row["sequence"], str): + raise ValueError("Invalid catalog sequence") + sequences.append( + CatalogSequence( + identity=identity + f"sequence/{row['sequence_id']}", + description=row["sequence_description"], + elements=row["sequence"], + source=source, + complete=complete, + reported_length=row["length"], + genbank_url=row.get("genbank_api_url") or row.get("genbank_url") or None, + ) + ) + return SupplierItem( + identity=identity, + supplier="Addgene", + catalog_id=str(plasmid_id), + name=data["name"], + url=data["url"], + retrieved_at=retrieved_at, + sequences=tuple(sequences), + metadata_json=canonical_json(data), + ) + + +@dataclass(frozen=True, kw_only=True) +class AddgeneClient: + token: str = field(repr=False) + timeout: float = 30 + + def __post_init__(self) -> None: + if not self.token or any(character.isspace() for character in self.token): + raise ValueError("Pass a nonempty Addgene token without whitespace") + if not 0 < self.timeout <= 120: + raise ValueError("Timeout must be between zero and 120 seconds") + + def plasmid(self, plasmid_id: int, *, include_sequences: bool = True) -> SupplierItem: + if type(plasmid_id) is not int or plasmid_id < 1: + raise ValueError("Plasmid ID must be a positive integer") + endpoint = "plasmid-with-sequences" if include_sequences else "plasmid" + request = Request( + f"{BASE_URL}/catalog/{endpoint}/{plasmid_id}/", + headers={"Authorization": f"Token {self.token}", "Accept": "application/json"}, + method="GET", + ) + try: + with build_opener(_NoRedirect()).open(request, timeout=self.timeout) as response: + payload = response.read(16_000_001) + except HTTPError as error: + raise ValueError(f"Addgene catalog returned HTTP {error.code}") from None + if len(payload) > 16_000_000: + raise ValueError("Catalog response exceeds 16 MB") + result = parse_plasmid(payload.decode("utf-8"), retrieved_at=datetime.now(UTC)) + if result.catalog_id != str(plasmid_id): + raise ValueError("Catalog returned a different plasmid ID") + return result diff --git a/src/lab/suppliers/types.py b/src/lab/suppliers/types.py new file mode 100644 index 0000000..0f1ee7e --- /dev/null +++ b/src/lab/suppliers/types.py @@ -0,0 +1,336 @@ +"""Immutable catalog evidence and records of externally placed purchases.""" + +import json +from dataclasses import dataclass +from datetime import datetime +from decimal import Decimal +from enum import StrEnum +from pathlib import Path + +from lab.artifacts import canonical_json, digest, write_bundle +from lab.inventory import CountedStock, MaterialForm, Stock, StockLocation +from lab.provenance import ( + Activity, + Agent, + Association, + Component, + Document, + DocumentSnapshot, + EvidenceState, + Implementation, + Ref, +) +from lab.provenance.types import require_iri +from lab.provenance.vocabulary import LAB + + +def aware(value: datetime) -> None: + if not isinstance(value, datetime) or value.utcoffset() is None: + raise ValueError("Record timestamps must include a timezone") + + +class SequenceSource(StrEnum): + DEPOSITOR = "depositor" + ADDGENE = "addgene" + + +@dataclass(frozen=True, kw_only=True) +class CatalogSequence: + identity: str + description: str + elements: str + source: SequenceSource + complete: bool + reported_length: int | None + genbank_url: str | None = None + + def __post_init__(self) -> None: + require_iri(self.identity) + if not isinstance(self.source, SequenceSource) or type(self.complete) is not bool: + raise TypeError("Sequence source and completeness must be explicit") + if self.reported_length is not None and ( + type(self.reported_length) is not int or self.reported_length < 0 + ): + raise ValueError("Reported length must be nonnegative") + if self.genbank_url is not None: + require_iri(self.genbank_url) + + +@dataclass(frozen=True, kw_only=True) +class SupplierItem: + identity: str + supplier: str + catalog_id: str + name: str + url: str + retrieved_at: datetime + sequences: tuple[CatalogSequence, ...] = () + metadata_json: str = "{}" + + def __post_init__(self) -> None: + require_iri(self.identity) + require_iri(self.url) + aware(self.retrieved_at) + if not self.supplier.strip() or not self.catalog_id.strip(): + raise ValueError("Supplier and catalog ID are required") + if not isinstance(self.sequences, tuple) or not all( + isinstance(item, CatalogSequence) for item in self.sequences + ): + raise TypeError("Catalog sequences must be an immutable tuple") + if len({item.identity for item in self.sequences}) != len(self.sequences): + raise ValueError("Catalog sequence identities must be unique") + metadata = json.loads(self.metadata_json) + if not isinstance(metadata, dict): + raise ValueError("Catalog metadata must be a JSON object") + object.__setattr__(self, "metadata_json", canonical_json(metadata)) + + +@dataclass(frozen=True, kw_only=True) +class CatalogEntry: + """Caller-reviewed mapping; a catalog name is never a design identity. + + The caller explicitly selects a design and the supplied material form. + Sequence candidates remain evidence, not an automatic assertion of identity. + """ + + item: SupplierItem + design: Ref[Component] + form: MaterialForm + + def __post_init__(self) -> None: + if not isinstance(self.item, SupplierItem) or not isinstance(self.design, Ref): + raise TypeError("Catalog entries need an item and a design reference") + if not isinstance(self.form, MaterialForm): + raise TypeError("Catalog material form must be explicit") + + +@dataclass(frozen=True, kw_only=True) +class Catalog: + entries: tuple[CatalogEntry, ...] = () + + def __post_init__(self) -> None: + if not isinstance(self.entries, tuple) or not all( + isinstance(item, CatalogEntry) for item in self.entries + ): + raise TypeError("Catalog entries must be an immutable tuple") + keys = [(entry.item.identity, entry.design, entry.form) for entry in self.entries] + if len(set(keys)) != len(keys): + raise ValueError("Duplicate catalog mapping") + object.__setattr__( + self, + "entries", + tuple( + sorted( + self.entries, + key=lambda entry: ( + entry.item.supplier.casefold() != "addgene", + entry.item.identity, + entry.design.identity, + entry.form.value, + ), + ) + ), + ) + + def candidates(self, design: Ref[Component]) -> tuple[CatalogEntry, ...]: + return tuple(entry for entry in self.entries if entry.design == design) + + @property + def digest(self) -> str: + return digest(self) + + def write(self, path: str | Path) -> Path: + path = Path(path) + write_bundle( + path.parent, {path.name: canonical_json({"format": "lab.catalog.v1", "catalog": self})} + ) + return path + + @classmethod + def read(cls, path: str | Path) -> "Catalog": + data = json.loads(Path(path).read_text(encoding="utf-8")) + if data.get("format") != "lab.catalog.v1": + raise ValueError("Expected lab.catalog.v1") + entries = [] + for row in data["catalog"]["entries"]: + raw = row["item"] + sequences = tuple( + CatalogSequence(**{**sequence, "source": SequenceSource(sequence["source"])}) + for sequence in raw["sequences"] + ) + item = SupplierItem( + **{ + **raw, + "retrieved_at": datetime.fromisoformat(raw["retrieved_at"]), + "sequences": sequences, + } + ) + entries.append( + CatalogEntry( + item=item, design=Ref(row["design"]["identity"]), form=MaterialForm(row["form"]) + ) + ) + return cls(entries=tuple(entries)) + + +@dataclass(frozen=True, kw_only=True) +class AcquisitionRequest: + identity: str + design: Ref[Component] + required_form: MaterialForm + volume_ul: Decimal + candidates: tuple[CatalogEntry, ...] = () + count: int = 0 + + def __post_init__(self) -> None: + require_iri(self.identity) + if ( + not isinstance(self.volume_ul, Decimal) + or not self.volume_ul.is_finite() + or self.volume_ul < 0 + or type(self.count) is not int + or self.count < 0 + or (self.required_form.counted and (self.volume_ul != 0 or self.count <= 0)) + or (not self.required_form.counted and (self.volume_ul <= 0 or self.count != 0)) + ): + raise ValueError("Acquisition requires a positive quantity of the declared form") + if not isinstance(self.candidates, tuple) or any( + not isinstance(item, CatalogEntry) or item.design != self.design + for item in self.candidates + ): + raise ValueError("Acquisition candidates must refer to the requested design") + + +@dataclass(frozen=True, kw_only=True) +class QuoteRecord: + identity: str + acquisition: str + item: str + reference: str + issued_at: datetime + attachment: str | None = None + + def __post_init__(self) -> None: + for value in (self.identity, self.acquisition, self.item): + require_iri(value) + aware(self.issued_at) + if not self.reference.strip(): + raise ValueError("Quote reference is required") + if self.attachment is not None: + require_iri(self.attachment) + + +@dataclass(frozen=True, kw_only=True) +class OrderReference: + identity: str + acquisition: str + item: str + reference: str + placed_at: datetime + quote: str | None = None + tracking_url: str | None = None + + def __post_init__(self) -> None: + for value in (self.identity, self.acquisition, self.item): + require_iri(value) + aware(self.placed_at) + if not self.reference.strip(): + raise ValueError("External order reference is required") + for optional in (self.quote, self.tracking_url): + if optional is not None: + require_iri(optional) + + +@dataclass(frozen=True, kw_only=True) +class Receipt: + """Observation of receipt, not sequence verification or DNA preparation. + + Counts describe supplied packages. A liquid aliquot enters inventory only + through an explicit measured quantity and its own implementation identity. + """ + + identity: str + order: OrderReference + design: Ref[Component] + form: MaterialForm + received_at: datetime + received_by: Ref[Agent] + packages: int + lot: str | None = None + + def __post_init__(self) -> None: + require_iri(self.identity) + aware(self.received_at) + if not isinstance(self.order, OrderReference) or not isinstance(self.form, MaterialForm): + raise TypeError("A receipt needs an order and an explicit material form") + if not isinstance(self.design, Ref) or not isinstance(self.received_by, Ref): + raise TypeError("Receipt design and receiving agent must be references") + if type(self.packages) is not int or self.packages < 1: + raise ValueError("Receipt package count must be positive") + if self.received_at < self.order.placed_at: + raise ValueError("Receipt precedes its order") + + @property + def implementation(self) -> Ref[Implementation]: + return Ref(self.identity + "/material") + + def record(self, document: DocumentSnapshot) -> DocumentSnapshot: + document.resolve(self.design) + document.resolve(self.received_by) + activity = Activity( + identity=self.identity, + types=(LAB + "receipt",), + evidence_state=EvidenceState.RECORDED, + end_time=self.received_at, + association=(Association(agent=self.received_by),), + description=f"Received {self.packages} package(s); order {self.order.reference}; " + f"item {self.order.item}; form {self.form.value}; lot {self.lot or 'unrecorded'}.", + ) + material = Implementation( + identity=self.implementation.identity, + derived_from=(self.design,), + generated_by=(activity.ref,), + evidence_state=EvidenceState.RECORDED, + ) + result = Document.from_snapshot(document) + result.add(activity, material) + return result.freeze() + + def counted_stock( + self, *, identity: str, count: int, location: StockLocation | None = None + ) -> CountedStock: + """Record an explicitly assessed count; package count is not a material count.""" + return CountedStock( + identity=identity, + implementation=self.implementation, + design=self.design, + form=self.form, + count=count, + location=location, + supplier_item=self.order.item, + ) + + def stock( + self, + *, + identity: str, + quantity: object, + location: StockLocation | None = None, + concentration: object = None, + ) -> Stock: + if self.form.counted: + raise ValueError( + "A bacterial stab requires explicit preparation before liquid inventory" + ) + if self.packages != 1: + raise ValueError("Record each package separately before measuring an inventory aliquot") + return Stock( + identity=identity, + implementation=self.implementation, + design=self.design, + form=self.form, + quantity=quantity, + concentration=concentration, + location=location, + supplier_item=self.order.item, + ) diff --git a/tests/test_suppliers.py b/tests/test_suppliers.py new file mode 100644 index 0000000..31d21b0 --- /dev/null +++ b/tests/test_suppliers.py @@ -0,0 +1,103 @@ +import io +import json +from dataclasses import replace +from datetime import UTC, datetime +from unittest.mock import patch + +import pytest + +from lab.inventory import MaterialForm +from lab.provenance import Ref +from lab.suppliers import ( + AddgeneClient, + Catalog, + CatalogEntry, + SequenceSource, + parse_plasmid, +) + +TIME = datetime(2026, 9, 27, tzinfo=UTC) + + +def response(): + # Synthetic neutral bases, with the field names of Addgene's retrieve schema. + return json.dumps( + { + "id": 42, + "name": "Contract fixture", + "url": "https://www.addgene.org/42/", + "sequences": { + "public_user_full_sequences": [ + { + "sequence_id": 17, + "sequence_description": "Depositor assertion", + "sequence": "ACGT", + "length": 4, + "genbank_url": "", + "genbank_api_url": "https://example.org/sequence.gb", + } + ], + "public_addgene_partial_sequences": [ + { + "sequence_id": 18, + "sequence_description": "Partial trace", + "sequence": "AC", + "length": 2, + "genbank_api_url": "", + } + ], + }, + } + ) + + +def catalog(design): + item = parse_plasmid(response(), retrieved_at=TIME) + return Catalog( + entries=(CatalogEntry(item=item, design=design, form=MaterialForm.BACTERIAL_STAB),) + ) + + +def test_catalog_preserves_source_completeness_and_original_metadata(tmp_path): + snapshot = catalog(Ref("https://example.org/catalog/design")) + item = snapshot.entries[0].item + assert item.sequences[0].source is SequenceSource.DEPOSITOR + assert item.sequences[0].complete + assert item.sequences[1].source is SequenceSource.ADDGENE + assert not item.sequences[1].complete + assert json.loads(item.metadata_json) == json.loads(response()) + path = snapshot.write(tmp_path / "catalog.json") + assert Catalog.read(path) == snapshot + + +def test_catalog_http_request_matches_documented_auth_and_endpoint(): + client = AddgeneClient(token="secret-example") + assert "secret-example" not in repr(client) + with patch("lab.suppliers.addgene.build_opener") as opener: + opener.return_value.open.return_value = io.BytesIO(response().encode()) + result = client.plasmid(42) + request = opener.return_value.open.call_args.args[0] + assert request.full_url == ( + "https://api.developers.addgene.org/catalog/plasmid-with-sequences/42/" + ) + assert request.method == "GET" + assert request.headers["Authorization"] == "Token secret-example" + assert result.catalog_id == "42" + with pytest.raises(ValueError, match="positive integer"): + client.plasmid(True) + with patch("lab.suppliers.addgene.build_opener") as opener: + opener.return_value.open.return_value = io.BytesIO(response().encode()) + with pytest.raises(ValueError, match="different plasmid"): + client.plasmid(43) + + +def test_addgene_is_preferred_without_automatic_sequence_selection(): + entry = catalog(Ref("https://example.org/catalog/design")).entries[0] + alternative = replace( + entry, + item=replace( + entry.item, supplier="Example", identity="https://example.org/catalog/supplier" + ), + ) + snapshot = Catalog(entries=(alternative, entry)) + assert snapshot.candidates(entry.design) == (entry, alternative)