kbank.db & kbank.io¶

@dataclass(frozen=True, slots=True)
class GlycanRecord:
    r"""Immutable record for a bacterial surface glycan entry across integrated databases.

    Consolidates taxonomic metadata, literature provenance, structural notations, parsed
    [`Glycan`][kbank.glycan.Glycan] graphs, de novo matched [`Locus`][serogen.loci.Locus]
    gene clusters from `serogen`, and resolved [`GeneGlycanAssignment`][kbank.biosynthesis.GeneGlycanAssignment]
    bipartite edges. Uses byte strings for textual fields to ensure zero-copy compatibility with
    Structure-of-Arrays (SoA) containers.

    Attributes:
        source_db: Source database identifier (`b"CSDB"`, `b"ECODAB"`, `b"EK3D"`, `b"EPSDB"`, `b"KEGG"`).
        record_id: Primary identifier within the source database (e.g. `b"70"`, `b"K1"`, `b"G00525"`).
        organism: Canonical bacterial organism name (e.g. `b"Escherichia coli"`).
        strain: Specific bacterial strain designation (e.g. `b"K12"`, `b"ATCC 43816"`).
        serotype: Serotype or antigen designation (e.g. `b"O1"`, `b"K2"`).
        antigen_type: Surface polysaccharide antigen class (`b"CPS"`, `b"O-antigen"`, `b"EPS"`, `b"WTA"`, `b"LPS"`).
        assembly_id: Linked genome assembly or nucleotide accession (e.g. `b"GCF_000005845.2"`, `b"AB811615"`).
        archetype: Biosynthetic locus archetype (`LocusArchetype`) matched or inferred for this glycan.
        locus: Matched [`Locus`][serogen.loci.Locus] biosynthetic gene cluster, or `None` if unpaired.
        alignment_score: Combined geno-glycomic alignment score in `[0.0, 1.0]` from `BiosyntheticAssigner`.
        assignments: Resolved gene-to-glycan component assignments (`GRAPH_ALIGNED` or `RULE_INFERRED`).
        structure_raw: Raw structure notation from the source database.
        structure_linear: Normalized linear sequence notation (CSDB linear code or IUPAC condensed).
        glycan: Parsed [`Glycan`][kbank.glycan.Glycan] core graph instance, or `None` if unparsed.
        is_repeating: Whether the structure represents a repeating polymer unit.
        pmid: PubMed ID integer, or `0` if unknown.
        publication_year: Four-digit year of primary publication, or `0` if unknown.
        doi: Digital Object Identifier byte string.
        glytoucan_id: GlyTouCan accession identifier (e.g. `b"G12345AB"`), or empty byte string.
        notes: Additional curation notes or cross-reference annotations.
    """

    source_db: bytes
    record_id: bytes
    organism: bytes
    strain: bytes = b""
    serotype: bytes = b""
    antigen_type: bytes = b""
    assembly_id: bytes = b""
    archetype: LocusArchetype = LocusArchetype.UNKNOWN
    locus: Locus | None = None
    alignment_score: float = 0.0
    assignments: tuple[GeneGlycanAssignment, ...] = ()
    structure_raw: bytes = b""
    structure_linear: bytes = b""
    glycan: Glycan | None = None
    is_repeating: bool = True
    pmid: int = 0
    publication_year: int = 0
    doi: bytes = b""
    glytoucan_id: bytes = b""
    notes: bytes = b""

    @property
    def compound_key(self) -> bytes:
        r"""Global unique identifier combining source database and primary record ID.

        Returns:
            bytes: Byte string formatted as `b"<source_db>:<record_id>"` (e.g. `b"ECODAB:70"`).
        """
        return b"%s:%s" % (self.source_db, self.record_id)

    @property
    def has_locus(self) -> bool:
        r"""True if this glycan record is paired with a non-empty biosynthetic `Locus`."""
        return self.locus is not None and len(self.locus.nodes) > 0

    @property
    def num_assignments(self) -> int:
        r"""Total number of gene-to-glycan component assignments on this record."""
        return len(self.assignments)

    @property
    def num_aligned_assignments(self) -> int:
        r"""Number of graph-aligned ORF-to-glycan component assignments (`gene_index >= 0`)."""
        return sum(1 for a in self.assignments if a.is_aligned)

    @property
    def num_pan_consensus_assignments(self) -> int:
        r"""Number of assignments resolved by multi-record pan-graph homology or synteny consensus."""
        return sum(1 for a in self.assignments if a.is_pan_consensus)

    @property
    def num_rule_assignments(self) -> int:
        r"""Number of stoichiometric rule-inferred `GeneRole`-to-glycan component assignments."""
        return sum(1 for a in self.assignments if not a.is_aligned)

    def format_assignments(self) -> str:
        r"""Compact human-readable summary string of all gene-to-glycan assignments."""
        return "; ".join(a.format_short() for a in self.assignments)

@dataclass(slots=True, frozen=True)
class GlycanRecords(BatchedContainer[GlycanRecord, "GlycanRecords"]):
    r"""Pure Structure-of-Arrays (SoA) batch container for consolidated bacterial glycan records.

    Inherits from [`BatchedContainer`][serogen.core.collections.BatchedContainer] to provide
    vectorized operations, contiguous flat NumPy arrays, embedded [`Glycans`][kbank.glycan.Glycans],
    [`Loci`][serogen.loci.Loci], and [`GeneGlycanAssignments`][kbank.biosynthesis.GeneGlycanAssignments]
    batch representations, zero-pickle `.npz` serialization, and seamless slice / boolean / integer mask
    indexing. Individual [`GlycanRecord`][kbank.db.models.GlycanRecord] scalar views are materialized on
    demand when indexing a single row `records[i]`.

    Attributes:
        source_dbs: 1D array of source database identifier byte strings (`"S"`).
        record_ids: 1D array of primary record identifiers (`"S"`).
        organisms: 1D array of canonical bacterial organism names (`"S"`).
        strains: 1D array of bacterial strain designations (`"S"`).
        serotypes: 1D array of serotype designations (`"S"`).
        antigen_types: 1D array of surface antigen categories (`"S"`).
        assembly_ids: 1D array of linked genome assembly or nucleotide accessions (`"S"`).
        compound_keys: 1D array of compound keys (`b"<source_db>:<record_id>"`) (`"S"`).
        archetypes: 1D `int8` array of `LocusArchetype` integer IDs (`0..10`).
        alignment_scores: 1D `float32` array of geno-glycomic alignment scores in `[0.0, 1.0]`.
        pmids: 1D `int64` array of PubMed IDs.
        years: 1D `int16` array of publication years.
        is_repeating: 1D boolean array indicating whether the glycan represents a polymer repeat unit.
        glycans: Embedded [`Glycans`][kbank.glycan.Glycans] batch of all parsed glycan graphs.
        loci: Embedded [`Loci`][serogen.loci.Loci] batch of all paired biosynthetic gene clusters.
        assignments: Embedded [`GeneGlycanAssignments`][kbank.biosynthesis.GeneGlycanAssignments] ragged batch of cross-domain gene-to-glycan edges.
        structure_raws: 1D array of raw structure strings from the source database (`"S"`).
        structure_linears: 1D array of normalized linear structure strings (`"S"`).
        dois: 1D array of Digital Object Identifier byte strings (`"S"`).
        glytoucan_ids: 1D array of GlyTouCan accession byte strings (`"S"`).
        notes: 1D array of curation notes and provenance tags (`"S"`).
    """

    source_dbs: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    record_ids: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    organisms: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    strains: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    serotypes: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    antigen_types: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    assembly_ids: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    compound_keys: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    archetypes: np.ndarray[tuple[int], np.dtype[np.int8]]
    alignment_scores: np.ndarray[tuple[int], np.dtype[np.float32]]
    pmids: np.ndarray[tuple[int], np.dtype[np.int64]]
    years: np.ndarray[tuple[int], np.dtype[np.int16]]
    is_repeating: np.ndarray[tuple[int], np.dtype[np.bool_]]
    glycans: Glycans
    loci: Loci
    assignments: GeneGlycanAssignments
    structure_raws: np.ndarray[tuple[int], np.dtype[np.bytes_]] = field(default_factory=lambda: np.empty(0, dtype="S1"))
    structure_linears: np.ndarray[tuple[int], np.dtype[np.bytes_]] = field(
        default_factory=lambda: np.empty(0, dtype="S1")
    )
    dois: np.ndarray[tuple[int], np.dtype[np.bytes_]] = field(default_factory=lambda: np.empty(0, dtype="S1"))
    glytoucan_ids: np.ndarray[tuple[int], np.dtype[np.bytes_]] = field(default_factory=lambda: np.empty(0, dtype="S1"))
    notes: np.ndarray[tuple[int], np.dtype[np.bytes_]] = field(default_factory=lambda: np.empty(0, dtype="S1"))

    def __post_init__(self) -> None:
        n = len(self.source_dbs)
        if n > 0:
            for attr in ("structure_raws", "structure_linears", "dois", "glytoucan_ids", "notes"):
                arr = getattr(self, attr)
                if len(arr) == 0:
                    object.__setattr__(self, attr, np.zeros(n, dtype="S1"))

    def __len__(self) -> int:
        r"""Returns the total number of records in the batch container."""
        return len(self.source_dbs)

    def __iter__(self) -> Iterator[GlycanRecord]:
        r"""Iterates over the container, yielding each materialized `GlycanRecord` scalar view."""
        for i in range(len(self)):
            yield self[i]

    def __contains__(self, item: object) -> bool:
        r"""Checks membership by `GlycanRecord`, compound key, or assembly ID."""
        if isinstance(item, GlycanRecord):
            return bool(np.any(self.compound_keys == item.compound_key))
        if isinstance(item, str):
            item = item.encode("utf-8")
        if isinstance(item, bytes):
            return bool(np.any(self.compound_keys == item) or np.any(self.assembly_ids == item))
        return False

    @overload
    def __getitem__(self, item: int | np.integer[Any]) -> GlycanRecord: ...

    @overload
    def __getitem__(self, item: slice | np.ndarray[tuple[int], Any] | Sequence[Any]) -> Self: ...

    @overload
    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | Sequence[Any],
    ) -> GlycanRecord | Self: ...

    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | Sequence[Any],
    ) -> GlycanRecord | Self:
        r"""Access a materialized `GlycanRecord` by scalar index, or slice/mask the SoA container."""
        if isinstance(item, (int, np.integer)):
            idx = int(item)
            n = len(self)
            if idx < 0:
                idx += n
            if idx < 0 or idx >= n:
                raise IndexError(f"GlycanRecords index {item} out of range for batch of length {n}.")

            g_obj = self.glycans[idx]
            loc_obj = self.loci[idx]
            return GlycanRecord(
                source_db=bytes(self.source_dbs[idx]),
                record_id=bytes(self.record_ids[idx]),
                organism=bytes(self.organisms[idx]),
                strain=bytes(self.strains[idx]),
                serotype=bytes(self.serotypes[idx]),
                antigen_type=bytes(self.antigen_types[idx]),
                assembly_id=bytes(self.assembly_ids[idx]),
                archetype=LocusArchetype(int(self.archetypes[idx])),
                locus=loc_obj if len(loc_obj.nodes) > 0 else None,
                alignment_score=float(self.alignment_scores[idx]),
                assignments=self.assignments[idx],
                structure_raw=bytes(self.structure_raws[idx]),
                structure_linear=bytes(self.structure_linears[idx]),
                glycan=g_obj if len(g_obj.nodes) > 0 else None,
                is_repeating=bool(self.is_repeating[idx]),
                pmid=int(self.pmids[idx]),
                publication_year=int(self.years[idx]),
                doi=bytes(self.dois[idx]),
                glytoucan_id=bytes(self.glytoucan_ids[idx]),
                notes=bytes(self.notes[idx]),
            )

        if isinstance(item, slice):
            return type(self)(
                source_dbs=self.source_dbs[item],
                record_ids=self.record_ids[item],
                organisms=self.organisms[item],
                strains=self.strains[item],
                serotypes=self.serotypes[item],
                antigen_types=self.antigen_types[item],
                assembly_ids=self.assembly_ids[item],
                compound_keys=self.compound_keys[item],
                archetypes=self.archetypes[item],
                alignment_scores=self.alignment_scores[item],
                pmids=self.pmids[item],
                years=self.years[item],
                is_repeating=self.is_repeating[item],
                glycans=self.glycans[item],
                loci=self.loci[item],
                assignments=self.assignments[item],
                structure_raws=self.structure_raws[item],
                structure_linears=self.structure_linears[item],
                dois=self.dois[item],
                glytoucan_ids=self.glytoucan_ids[item],
                notes=self.notes[item],
            )

        arr_item = np.asarray(item)
        if arr_item.dtype == bool:
            if len(arr_item) != len(self):
                raise ValueError(f"Mask length {len(arr_item)} does not match batch length {len(self)}.")
            return type(self)(
                source_dbs=self.source_dbs[arr_item],
                record_ids=self.record_ids[arr_item],
                organisms=self.organisms[arr_item],
                strains=self.strains[arr_item],
                serotypes=self.serotypes[arr_item],
                antigen_types=self.antigen_types[arr_item],
                assembly_ids=self.assembly_ids[arr_item],
                compound_keys=self.compound_keys[arr_item],
                archetypes=self.archetypes[arr_item],
                alignment_scores=self.alignment_scores[arr_item],
                pmids=self.pmids[arr_item],
                years=self.years[arr_item],
                is_repeating=self.is_repeating[arr_item],
                glycans=self.glycans[arr_item],
                loci=self.loci[arr_item],
                assignments=self.assignments[arr_item],
                structure_raws=self.structure_raws[arr_item],
                structure_linears=self.structure_linears[arr_item],
                dois=self.dois[arr_item],
                glytoucan_ids=self.glytoucan_ids[arr_item],
                notes=self.notes[arr_item],
            )

        if np.issubdtype(arr_item.dtype, np.integer) or len(arr_item) == 0:
            indices = self._normalize_indices(arr_item)
            if len(indices) == 0:
                return self.empty()
            return type(self)(
                source_dbs=self.source_dbs[indices],
                record_ids=self.record_ids[indices],
                organisms=self.organisms[indices],
                strains=self.strains[indices],
                serotypes=self.serotypes[indices],
                antigen_types=self.antigen_types[indices],
                assembly_ids=self.assembly_ids[indices],
                compound_keys=self.compound_keys[indices],
                archetypes=self.archetypes[indices],
                alignment_scores=self.alignment_scores[indices],
                pmids=self.pmids[indices],
                years=self.years[indices],
                is_repeating=self.is_repeating[indices],
                glycans=self.glycans[indices],
                loci=self.loci[indices],
                assignments=self.assignments[indices],
                structure_raws=self.structure_raws[indices],
                structure_linears=self.structure_linears[indices],
                dois=self.dois[indices],
                glytoucan_ids=self.glytoucan_ids[indices],
                notes=self.notes[indices],
            )

        raise TypeError(f"Unsupported index type: {type(item)}")

    # Vectorized filtering methods relying directly on __getitem__ masking
    def where_organism(self, organism: bytes | str) -> Self:
        r"""Filters records by exact bacterial organism name."""
        org_b = organism.encode("utf-8") if isinstance(organism, str) else organism
        return self[self.organisms == org_b]

    def where_serotype(self, serotype: bytes | str) -> Self:
        r"""Filters records by serotype designation."""
        sero_b = serotype.encode("utf-8") if isinstance(serotype, str) else serotype
        return self[self.serotypes == sero_b]

    def where_source_db(self, source: bytes | str) -> Self:
        r"""Filters records by source database name (e.g. `b"CSDB"`, `b"ECODAB"`)."""
        src_b = source.encode("utf-8") if isinstance(source, str) else source
        return self[self.source_dbs == src_b]

    def where_assembly_id(self, assembly_id: bytes | str) -> Self:
        r"""Filters records by genome assembly or nucleotide accession."""
        asm_b = assembly_id.encode("utf-8") if isinstance(assembly_id, str) else assembly_id
        return self[self.assembly_ids == asm_b]

    def where_archetype(self, archetype: LocusArchetype | int | str | bytes) -> Self:
        r"""Filters records by `LocusArchetype`."""
        arch = LocusArchetype.from_name(archetype)
        return self[self.archetypes == int(arch)]

    def where_has_locus(self) -> Self:
        r"""Filters records to those paired with a non-empty biosynthetic `Locus`."""
        has_loc = np.diff(self.loci.node_offsets) > 0
        return self[has_loc]

    # Property aliases for convenience
    @property
    def organism(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `organisms`."""
        return self.organisms

    @property
    def serotype(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `serotypes`."""
        return self.serotypes

    @property
    def source_db(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `source_dbs`."""
        return self.source_dbs

    @property
    def assembly_id(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `assembly_ids`."""
        return self.assembly_ids

    @property
    def compound_key(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `compound_keys`."""
        return self.compound_keys

    @property
    def structure_raw(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `structure_raws`."""
        return self.structure_raws

    @property
    def structure_linear(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `structure_linears`."""
        return self.structure_linears

    @property
    def doi(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `dois`."""
        return self.dois

    @property
    def glytoucan_id(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        r"""Alias for `glytoucan_ids`."""
        return self.glytoucan_ids

    @property
    def has_locus(self) -> np.ndarray[tuple[int], np.dtype[np.bool_]]:
        r"""Boolean mask indicating which records have a non-empty paired `Locus`."""
        return np.diff(self.loci.node_offsets) > 0

    @classmethod
    def empty(cls) -> Self:
        r"""Constructs an empty, zero-length `GlycanRecords` container."""
        return cls(
            source_dbs=np.empty(0, dtype="S16"),
            record_ids=np.empty(0, dtype="S32"),
            organisms=np.empty(0, dtype="S64"),
            strains=np.empty(0, dtype="S64"),
            serotypes=np.empty(0, dtype="S32"),
            antigen_types=np.empty(0, dtype="S32"),
            assembly_ids=np.empty(0, dtype="S32"),
            compound_keys=np.empty(0, dtype="S48"),
            archetypes=np.empty(0, dtype=np.int8),
            alignment_scores=np.empty(0, dtype=np.float32),
            pmids=np.empty(0, dtype=np.int64),
            years=np.empty(0, dtype=np.int16),
            is_repeating=np.empty(0, dtype=bool),
            glycans=Glycans.empty(),
            loci=Loci.empty(),
            assignments=GeneGlycanAssignments.empty(),
            structure_raws=np.empty(0, dtype="S1"),
            structure_linears=np.empty(0, dtype="S1"),
            dois=np.empty(0, dtype="S1"),
            glytoucan_ids=np.empty(0, dtype="S1"),
            notes=np.empty(0, dtype="S1"),
        )

    @classmethod
    def concat(cls, batches: Iterable[Self]) -> Self:
        r"""Concatenates multiple `GlycanRecords` batches into a single container."""
        batch_list = [b for b in batches if len(b) > 0]
        if not batch_list:
            return cls.empty()
        if len(batch_list) == 1:
            return batch_list[0]

        return cls(
            source_dbs=np.concatenate([b.source_dbs for b in batch_list]),
            record_ids=np.concatenate([b.record_ids for b in batch_list]),
            organisms=np.concatenate([b.organisms for b in batch_list]),
            strains=np.concatenate([b.strains for b in batch_list]),
            serotypes=np.concatenate([b.serotypes for b in batch_list]),
            antigen_types=np.concatenate([b.antigen_types for b in batch_list]),
            assembly_ids=np.concatenate([b.assembly_ids for b in batch_list]),
            compound_keys=np.concatenate([b.compound_keys for b in batch_list]),
            archetypes=np.concatenate([b.archetypes for b in batch_list]),
            alignment_scores=np.concatenate([b.alignment_scores for b in batch_list]),
            pmids=np.concatenate([b.pmids for b in batch_list]),
            years=np.concatenate([b.years for b in batch_list]),
            is_repeating=np.concatenate([b.is_repeating for b in batch_list]),
            glycans=Glycans.concat([b.glycans for b in batch_list]),
            loci=Loci.concat([b.loci for b in batch_list]),
            assignments=GeneGlycanAssignments.concat([b.assignments for b in batch_list]),
            structure_raws=np.concatenate([b.structure_raws for b in batch_list]),
            structure_linears=np.concatenate([b.structure_linears for b in batch_list]),
            dois=np.concatenate([b.dois for b in batch_list]),
            glytoucan_ids=np.concatenate([b.glytoucan_ids for b in batch_list]),
            notes=np.concatenate([b.notes for b in batch_list]),
        )

    @classmethod
    def build(cls, records: Iterable[GlycanRecord]) -> Self:
        r"""Compiles an iterable of `GlycanRecord` instances into a pure SoA `GlycanRecords` batch."""
        raw_recs = list(records)
        n = len(raw_recs)
        if n == 0:
            return cls.empty()

        assigner = BiosyntheticAssigner()

        source_dbs = np.empty(n, dtype="S16")
        record_ids = np.empty(n, dtype="S32")
        organisms = np.empty(n, dtype="S64")
        strains = np.empty(n, dtype="S64")
        serotypes = np.empty(n, dtype="S32")
        antigen_types = np.empty(n, dtype="S32")
        assembly_ids = np.empty(n, dtype="S32")
        compound_keys = np.empty(n, dtype="S48")
        archetypes = np.empty(n, dtype=np.int8)
        alignment_scores = np.empty(n, dtype=np.float32)
        pmids = np.empty(n, dtype=np.int64)
        years = np.empty(n, dtype=np.int16)
        is_repeating = np.empty(n, dtype=bool)

        structure_raw_list: list[bytes] = []
        structure_linear_list: list[bytes] = []
        doi_list: list[bytes] = []
        glytoucan_id_list: list[bytes] = []
        notes_list: list[bytes] = []

        empty_g = Glycan(b"", (), ())
        empty_loc = Locus(nodes=(), edges=())
        glycan_objs: list[Glycan] = []
        locus_objs: list[Locus] = []
        assign_lists: list[tuple[Any, ...]] = []

        for i, r in enumerate(raw_recs):
            if r.locus is not None and len(r.locus.nodes) > 0:
                loc = r.locus.canonicalize_strand()
                if not loc.name:
                    label = (r.serotype or r.strain or r.record_id).decode("utf-8", errors="replace").strip()
                    if label:
                        loc = loc.with_name(label)
                if loc is not r.locus:
                    r = replace(r, locus=loc)
            raw_recs[i] = r

            source_dbs[i] = r.source_db
            record_ids[i] = r.record_id
            organisms[i] = r.organism
            strains[i] = r.strain
            serotypes[i] = r.serotype
            antigen_types[i] = r.antigen_type
            assembly_ids[i] = r.assembly_id
            compound_keys[i] = r.compound_key
            archetypes[i] = int(r.archetype)
            alignment_scores[i] = float(r.alignment_score)
            pmids[i] = r.pmid
            years[i] = r.publication_year
            is_repeating[i] = r.is_repeating

            structure_raw_list.append(r.structure_raw)
            structure_linear_list.append(r.structure_linear)
            doi_list.append(r.doi)
            glytoucan_id_list.append(r.glytoucan_id)
            notes_list.append(r.notes)

            glycan_objs.append(r.glycan if r.glycan is not None else empty_g)
            locus_objs.append(r.locus if r.locus is not None else empty_loc)

        # 1. Cluster genes via MCL across all paired loci FIRST so every GeneNode carries a homology cluster_id
        loci_batch = Loci.from_list(locus_objs)
        if (
            len(loci_batch.node_types) > 0
            and len(loci_batch.genes) == len(loci_batch.node_types)
            and bool(np.all(loci_batch.cluster_ids == 0))
        ):
            loci_batch = loci_batch.cluster_genes(
                mode="mcl",
                min_identity=28.0,
                min_bitscore=40.0,
                inflation=2.0,
                by_role=True,
            )

        # 2. Fit cross-record PanGraphConsensusPrior & rotate glycans to biological reducing ends
        pan_prior, rotated_glycans = PanGraphConsensusPrior.fit(
            glycan_objs,
            loci_batch,
            archetypes,
            is_repeating,
            assigner=assigner,
        )
        if len(pan_prior.rescued_loci) == len(raw_recs):
            loci_batch = Loci.from_list(pan_prior.rescued_loci)

        # 3. Complete locus modifiers and resolve pan-graph-informed GeneGlycanAssignments
        final_glycans: list[Glycan] = []
        for i, r in enumerate(raw_recs):
            gly = rotated_glycans[i]
            loc = loci_batch[i] if len(loci_batch) > i else empty_loc
            completed_mod_genes: dict[int, int] | None = None

            if len(loc.nodes) > 0 and len(gly.nodes) > 0:
                comp_gly, comp_mods, comp_map = assigner.complete_glycan_modifiers(gly, loc)
                if comp_mods:
                    gly = comp_gly
                    completed_mod_genes = comp_map
                    notes_b = notes_list[i]
                    if b"[locus_completed=" not in notes_b:
                        tag = b" [locus_completed=" + b",".join(b"+" + m for m in comp_mods) + b"]"
                        notes_list[i] = notes_b + tag

            final_glycans.append(gly)
            if not r.assignments and len(gly.nodes) > 0:
                inferred = assigner.assign_record(
                    gly,
                    loc if len(loc.nodes) > 0 else None,
                    archetype=r.archetype,
                    is_repeating=r.is_repeating,
                    completed_modifier_genes=completed_mod_genes,
                    pan_prior=pan_prior,
                )
                assign_lists.append(inferred)
            else:
                assign_lists.append(r.assignments)

        glycans_batch = Glycans.from_list(final_glycans)
        assignments_batch = GeneGlycanAssignments.from_list(assign_lists)

        return cls(
            source_dbs=source_dbs,
            record_ids=record_ids,
            organisms=organisms,
            strains=strains,
            serotypes=serotypes,
            antigen_types=antigen_types,
            assembly_ids=assembly_ids,
            compound_keys=compound_keys,
            archetypes=archetypes,
            alignment_scores=alignment_scores,
            pmids=pmids,
            years=years,
            is_repeating=is_repeating,
            glycans=glycans_batch,
            loci=loci_batch,
            assignments=assignments_batch,
            structure_raws=np.array(structure_raw_list, dtype="S"),
            structure_linears=np.array(structure_linear_list, dtype="S"),
            dois=np.array(doi_list, dtype="S"),
            glytoucan_ids=np.array(glytoucan_id_list, dtype="S"),
            notes=np.array(notes_list, dtype="S"),
        )

    @classmethod
    async def abuild(cls, records: AsyncIterable[GlycanRecord]) -> Self:
        r"""Asynchronously compiles an async stream of `GlycanRecord` instances into a `GlycanRecords` batch."""
        collected: list[GlycanRecord] = []
        async for r in records:
            collected.append(r)
        return cls.build(collected)

    def save(self, path: Path | str | None = None) -> Path:
        r"""Serializes the `GlycanRecords` instance to disk as a compressed NumPy `.npz` archive (zero pickle)."""
        p = Path(path) if path is not None else DEFAULT_DB_PATH
        p.parent.mkdir(parents=True, exist_ok=True)
        tmp_p = p.with_name(f".{p.name}.{os.getpid()}.tmp.npz")

        np.savez_compressed(
            tmp_p,
            # Record scalar columns
            source_dbs=self.source_dbs,
            record_ids=self.record_ids,
            organisms=self.organisms,
            strains=self.strains,
            serotypes=self.serotypes,
            antigen_types=self.antigen_types,
            assembly_ids=self.assembly_ids,
            compound_keys=self.compound_keys,
            archetypes=self.archetypes,
            alignment_scores=self.alignment_scores,
            pmids=self.pmids,
            years=self.years,
            is_repeating=self.is_repeating,
            structure_raws=self.structure_raws,
            structure_linears=self.structure_linears,
            dois=self.dois,
            glytoucan_ids=self.glytoucan_ids,
            notes=self.notes,
            # Glycans SoA
            glycan_node_types=self.glycans.node_types,
            glycan_node_offsets=self.glycans.node_offsets,
            glycan_edge_sources=self.glycans.edge_sources,
            glycan_edge_targets=self.glycans.edge_targets,
            glycan_edge_types=self.glycans.edge_types,
            glycan_edge_offsets=self.glycans.edge_offsets,
            glycan_canonicals=self.glycans.canonicals,
            # Loci & Genes SoA
            **_loci_to_npz_dict(self.loci, prefix="locus_"),
            # GeneGlycanAssignments SoA
            assignment_gene_indices=self.assignments.gene_indices,
            assignment_roles=self.assignments.roles,
            assignment_bipartite_indices=self.assignments.bipartite_indices,
            assignment_target_types=self.assignments.target_types,
            assignment_components=self.assignments.components,
            assignment_edge_indices=self.assignments.edge_indices,
            assignment_evidences=self.assignments.evidences,
            assignment_scores=self.assignments.scores,
            assignment_offsets=self.assignments.offsets,
            assignment_lengths=self.assignments.lengths,
        )
        os.replace(tmp_p, p)
        return p

    @classmethod
    def load(cls, path: Path | str | None = None) -> Self:
        r"""Deserializes a `GlycanRecords` instance from a compressed NumPy `.npz` archive (zero pickle, zero eager Python loop)."""
        p = Path(path) if path is not None else DEFAULT_DB_PATH
        with np.load(p, allow_pickle=False) as data:
            source_dbs = data["source_dbs"]
            n = len(source_dbs)
            if n == 0:
                return cls.empty()

            record_ids = data["record_ids"]
            organisms = data["organisms"]
            strains = data["strains"]
            serotypes = data["serotypes"]
            antigen_types = data["antigen_types"]
            assembly_ids = data["assembly_ids"]
            compound_keys = data["compound_keys"]
            archetypes = data["archetypes"]
            alignment_scores = data["alignment_scores"]
            pmids = data["pmids"]
            years = data["years"]
            is_repeating = data["is_repeating"]

            structure_raws = data["structure_raws"]
            structure_linears = data["structure_linears"]
            dois = data["dois"]
            glytoucan_ids = data["glytoucan_ids"]
            notes = data["notes"]

            glycans = Glycans(
                node_types=data["glycan_node_types"],
                node_offsets=data["glycan_node_offsets"],
                edge_sources=data["glycan_edge_sources"],
                edge_targets=data["glycan_edge_targets"],
                edge_types=data["glycan_edge_types"],
                edge_offsets=data["glycan_edge_offsets"],
                canonicals=data["glycan_canonicals"],
            )

            if "locus_gene_starts" in data.files:
                loci = _loci_from_npz_dict(data, prefix="locus_")
            else:
                gene_intervals = Intervals(
                    starts=data["gene_starts"],
                    ends=data["gene_ends"],
                    strands=data["gene_strands"],
                )
                gene_seqs = Sequences(
                    ids=data["gene_seq_ids"],
                    seqs=data["gene_seq_bytes"],
                    offsets=data["gene_seq_offsets"],
                    lengths=data["gene_seq_lengths"],
                )
                gene_indels = Indels(
                    positions=data["gene_indel_positions"],
                    kinds=data["gene_indel_kinds"],
                    offsets=data["gene_indel_offsets"],
                    lengths=data["gene_indel_lengths"],
                )
                genes = Genes(
                    intervals=gene_intervals,
                    seqs=gene_seqs if len(gene_seqs) > 0 else None,
                    contig_indices=data["gene_contig_indices"],
                    indels=gene_indels,
                )
                loci = Loci(
                    node_types=data["locus_node_types"],
                    node_offsets=data["locus_node_offsets"],
                    edge_sources=data["locus_edge_sources"],
                    edge_targets=data["locus_edge_targets"],
                    edge_types=data["locus_edge_types"],
                    edge_offsets=data["locus_edge_offsets"],
                    genes=genes,
                    intervals=Intervals(
                        starts=data["locus_starts"],
                        ends=data["locus_ends"],
                        strands=data["locus_strands"],
                    ),
                    contig_indices=data["locus_contig_indices"],
                    archetypes=data["locus_archetypes"],
                    completeness=data["locus_completeness"],
                    scores=data["locus_scores"],
                    is_tandem_split=data["locus_is_tandem_split"],
                )

            has_saved_assignments = "assignment_offsets" in data.files
            if has_saved_assignments:
                assignments = GeneGlycanAssignments(
                    gene_indices=data["assignment_gene_indices"],
                    roles=data["assignment_roles"],
                    bipartite_indices=data["assignment_bipartite_indices"],
                    target_types=data["assignment_target_types"],
                    components=data["assignment_components"],
                    edge_indices=data["assignment_edge_indices"],
                    evidences=data["assignment_evidences"],
                    scores=data["assignment_scores"],
                    offsets=data["assignment_offsets"],
                    lengths=data["assignment_lengths"],
                )
            else:
                assigner = BiosyntheticAssigner()
                fallback_lists = [
                    assigner.assign_record(
                        glycans[i] if len(glycans[i].nodes) > 0 else None,
                        loci[i] if len(loci[i].nodes) > 0 else None,
                        archetype=LocusArchetype(int(archetypes[i])),
                        is_repeating=bool(is_repeating[i]),
                    )
                    for i in range(n)
                ]
                assignments = GeneGlycanAssignments.from_list(fallback_lists)

            return cls(
                source_dbs=source_dbs,
                record_ids=record_ids,
                organisms=organisms,
                strains=strains,
                serotypes=serotypes,
                antigen_types=antigen_types,
                assembly_ids=assembly_ids,
                compound_keys=compound_keys,
                archetypes=archetypes,
                alignment_scores=alignment_scores,
                pmids=pmids,
                years=years,
                is_repeating=is_repeating,
                glycans=glycans,
                loci=loci,
                assignments=assignments,
                structure_raws=structure_raws,
                structure_linears=structure_linears,
                dois=dois,
                glytoucan_ids=glytoucan_ids,
                notes=notes,
            )

    @classmethod
    def pretrained(cls, path: Path | str | None = None) -> Self:
        r"""Loads the bundled curated `GlycanRecords` database (`src/kbank/data/glycans.npz`)."""
        return cls.load(path=path)