kaptive.db¶

@classmethod
def from_genbank(cls, file: str | Path) -> "Database":
    r"""Compiles a Database object by parsing a legacy GenBank format file and its associated TOML metadata.

    This is an expensive compilation step that converts the nested, string-heavy GenBank structure
    into the flat, integer-based SoA layout defined by this class.

    The method performs the following tasks:

    1.  Iterates through loci in the `.gbk` file, extracting sequence data.
    2.  Parses CDS (Coding Sequence) features to extract gene coordinates, clusters, and descriptions.
    3.  Builds the string-to-integer vocabularies (`cluster_vocab`, `description_vocab`).
    4.  Constructs the flat numpy arrays (e.g., `gene_cluster_ids`, `gene_positions`).
    5.  Creates `Sequences` and `Intervals` objects for vectorized sequence operations.
    6.  Translates all extracted gene nucleotide sequences into protein sequences.
    7.  Loads and validates metadata from a companion `.toml` file (which must have the same name as
        the `.gbk` file).
    8.  Parses complex phenotype logic from the metadata.

    Args:
        file (str | Path): The path to the `.gbk` database file. A companion `.toml` file
            must exist in the same directory.

    Returns:
        Database: The newly compiled, optimized Database object.

    Raises:
        DatabaseError: If loci lack required qualifiers ('note', 'locus'), or if the associated
            `.toml` metadata file is missing.
    """
    file = cls._check_file(file)
    from gb_io import iter as GenbankIterator

    _LOCUS_REGEX = re_compile(r"locus:\s?(.*)$")
    _SEROTYPE_REGEX = re_compile(r"type:\s?(.*)$")
    _EXTRA_REGEX = re_compile(r"Extra genes:\s?(.*)$")

    global_gene_idx = 0

    # Locus trackers
    locus_records, serotype_names, locus_gene_offsets, locus_gene_lengths, locus_intervals = (
        [],
        [],
        [],
        [],
        [],
    )

    # Gene trackers
    gene_ids, extra_genes = [], []
    gene_cluster_ids, gene_description_ids, gene_expected_positions = [], [], []

    # Global Vocabulary tracker
    cluster_vocab, description_vocab = {}, {}

    with file.open("rb") as fh:
        for rec in GenbankIterator(fh):
            locus_name, serotype, extra = None, None, False

            if not (notes := [i.value for i in rec.features[0].qualifiers if i.key == "note"]):
                raise DatabaseError(f'Locus has no "note" qualifiers: {rec.name}')

            # Iterate over notes to extract locus and type names, or whether the locus is an "Extra genes" locus
            for note in notes:
                if match := _EXTRA_REGEX.search(note):  # type: ignore
                    extra = True
                    locus_name = match.group(1)
                    break

                if not locus_name and (match := _LOCUS_REGEX.search(note)):  # type: ignore
                    locus_name = match.group(1)

                if not serotype and (match := _SEROTYPE_REGEX.search(note)):  # type: ignore
                    serotype = match.group(1)

            if not locus_name:
                raise DatabaseError(f'Locus has no valid "locus" qualifiers: {rec.name}')

            locus_record = SeqRecord(locus_name, rec.sequence.upper())  # type: ignore

            # Local trackers for the current locus
            starts, ends, strands = [], [], []

            local_gene_idx = 0
            locus_start_idx = global_gene_idx

            for feat in rec.features[1:]:
                if feat.kind != "CDS":
                    continue

                cluster, description = "", ""
                for i in feat.qualifiers:
                    if not cluster and i.key == "gene":
                        cluster = i.value
                    if not description and i.key == "product":
                        description = i.value

                if not extra:
                    gene_id = f"{locus_name}_{local_gene_idx + 1:02}_{cluster}"
                else:
                    gene_id = cluster

                if cluster not in cluster_vocab:
                    cluster_vocab[cluster] = len(cluster_vocab)
                cluster_id = cluster_vocab[cluster]

                if description not in description_vocab:
                    description_vocab[description] = len(description_vocab)
                description_id = description_vocab[description]

                expected_pos = local_gene_idx + 1  # Standardizing biological position

                # Coordinate parsing
                loc = feat.location
                start, end = sorted((loc.start, loc.end))  # type: ignore
                strand_val = -1 if loc.strand in (-1, "-") else 1  # type: ignore

                # Append to flat gene arrays
                starts.append(start)
                ends.append(end)
                strands.append(strand_val)
                gene_ids.append(gene_id)
                gene_cluster_ids.append(cluster_id)
                gene_description_ids.append(description_id)

                # Extra genes do not have expected biological positions or synteny matrices
                gene_expected_positions.append(0 if extra else expected_pos)

                local_gene_idx += 1
                global_gene_idx += 1

            if local_gene_idx == 0:
                continue

            locus_gene_offsets.append(locus_start_idx)
            locus_gene_lengths.append(local_gene_idx)

            # 2. Handle Sequence Extraction via Intervals
            intervals = Intervals(
                np.array(starts, dtype=np.int32),
                np.array(ends, dtype=np.int32),
                np.array(strands, dtype=np.int8),
            )

            # 4. Append final locus metadata
            locus_records.append(locus_record)
            serotype_names.append(serotype or "")
            locus_intervals.append(intervals)
            extra_genes.extend([extra] * local_gene_idx)

        # Pre-compute an array mapping every global gene index back to its locus index
        gene_locus_indices = np.zeros(global_gene_idx, dtype=np.uint16)
        for i, (o, length) in enumerate(zip(locus_gene_offsets, locus_gene_lengths)):
            gene_locus_indices[o : o + length] = i

    db_gene_ids = tuple(gene_ids)
    loci = Sequences.from_records(locus_records)
    cluster_keys = tuple(cluster_vocab.keys())
    phenotype_objs = []
    if (metadata_file := file.with_suffix(".toml")).is_file():
        with metadata_file.open("rb") as fp:
            metadata = DatabaseMetadata.from_dict(tomllib.load(fp))
            for k, v in metadata.phenotype_logic.items():
                phenotype_objs.append(cls._parse_phenotype(k, v, loci.ids, cluster_keys))
    else:
        raise DatabaseError("Missing required TOML metadata file alongside Genbank file.")

    # Initialize Phenotype SoA arrays
    n_pheno, n_loci, n_clusters = len(phenotype_objs), len(loci), len(cluster_keys)
    pheno_ids = []
    locus_vocab = {name: i for i, name in enumerate(loci.ids)}
    locus_masks = np.zeros((n_pheno, n_loci), dtype=bool)
    extra_masks = np.zeros((n_pheno, n_clusters), dtype=np.int8)
    inactive_masks = np.zeros((n_pheno, n_clusters), dtype=np.int8)
    priorities = np.zeros(n_pheno, dtype=np.int8)
    as_suffix = np.zeros(n_pheno, dtype=bool)

    for i, p in enumerate(phenotype_objs):
        pheno_ids.append(p.id)
        for loc in p.loci:
            locus_masks[i, locus_vocab[loc]] = True
        for ext in p.extra_genes:
            extra_masks[i, cluster_vocab[ext]] = 1
        for ina in p.inactive_genes:
            inactive_masks[i, cluster_vocab[ina]] = 1
        priorities[i] = p.priority
        as_suffix[i] = p.as_suffix

    global_intervals = Intervals.concat(locus_intervals)
    genes = loci.extract_intervals(gene_locus_indices, global_intervals, new_ids=db_gene_ids)
    translations = genes.translate()

    return cls(
        metadata=metadata,
        loci=loci,
        serotypes=tuple(serotype_names),
        locus_gene_offsets=np.array(locus_gene_offsets, dtype=np.uint32),
        locus_gene_lengths=np.array(locus_gene_lengths, dtype=np.uint32),
        gene_intervals=global_intervals,
        genes=genes,
        translations=translations,
        extra_genes=np.array(extra_genes, dtype=bool),
        gene_locus_indices=gene_locus_indices,
        cluster_keys=cluster_keys,
        gene_cluster_ids=np.array(gene_cluster_ids, dtype=np.uint16),
        description_keys=tuple(description_vocab.keys()),
        gene_description_ids=np.array(gene_description_ids, dtype=np.uint16),
        gene_positions=np.array(gene_expected_positions, dtype=np.uint16),
        phenotypes=Phenotypes(
            ids=np.array([p.encode("utf-8") for p in pheno_ids], dtype="S32"),
            locus_masks=locus_masks,
            extra_masks=extra_masks,
            inactive_masks=inactive_masks,
            extra_counts=extra_masks.sum(axis=1, dtype=np.int8),
            priorities=priorities,
            as_suffix=as_suffix,
        ),
        loci_sketches=FracMinHashIndex.build(loci, sort_by_hash=False),
    )

@classmethod
def from_pickle(cls, file: str | Path) -> "Database":
    r"""Loads a pre-compiled Database object from a serialized pickle file.

    Args:
        file (str | Path): Path to `.pkl` database file.

    Returns:
        Database: Deserialized [`Database`][kaptive.db.core.Database] instance.

    Raises:
        FileNotFoundError: If the file does not exist.

    See Also:
        [`load`][kaptive.db.core.Database.load]
    """
    return pickle.loads(cls._check_file(file).read_bytes())

def get_locus_data(self, locus_name: str) -> "LocusData":
    r"""Extracts locus data including protein sequences, coordinate backbone, gene states, and product descriptions for a specific locus.

    Args:
        locus_name (str): Identifier of the locus sequence (e.g., 'K1').

    Returns:
        LocusData: Container object populated with locus proteins, coordinate backbone, gene states, and product descriptions
            ([`LocusData`][kaptive.compare.LocusData]).

    Raises:
        ValueError: If `locus_name` is not present in `self.loci.ids`.

    See Also:
        [`LocusData`][kaptive.compare.LocusData]
    """
    from kaptive.compare import LocusData
    from kaptive.serotyping.models import GeneState

    locus_idx = self.loci.ids.index(locus_name)
    start = self.locus_gene_offsets[locus_idx]
    length = self.locus_gene_lengths[locus_idx]

    desc_ids = self.gene_description_ids[start : start + length]
    raw_descs = [
        self.description_keys[i].decode("utf-8")
        if isinstance(self.description_keys[i], bytes)
        else str(self.description_keys[i])
        for i in desc_ids
    ]
    descriptions = np.asarray(raw_descs, dtype=object)
    states = np.full(length, GeneState.NORMAL.value, dtype=np.int8)

    return LocusData(
        proteins=self.translations[start : start + length],  # type: ignore
        name=locus_name,
        backbone=self.gene_intervals[start : start + length],  # type: ignore
        pieces=None,
        gene_ctg_indices=None,
        gene_states=states,
        gene_descriptions=descriptions,
    )

@classmethod
def load(cls, file: str | Path) -> "Database":
    r"""Loads a Database from a file path (GenBank or Pickle).

    Factory entry point that delegates loading based on file extension:
    1. If `.gbk` file path is provided, invokes [`from_genbank`][kaptive.db.core.Database.from_genbank].
    2. If `.pkl` file path is provided, invokes [`from_pickle`][kaptive.db.core.Database.from_pickle].

    Args:
        file (str | Path): Path to `.gbk`/`.pkl` file.

    Returns:
        Database: Loaded and initialized [`Database`][kaptive.db.core.Database] instance.

    Raises:
        DatabaseError: If file extension is unsupported.
        FileNotFoundError: If the file does not exist.

    See Also:
        [`from_genbank`][kaptive.db.core.Database.from_genbank],
        [`from_pickle`][kaptive.db.core.Database.from_pickle]
    """
    file_path = cls._check_file(file)
    if file_path.suffix == ".gbk":
        return cls.from_genbank(file_path)
    elif file_path.suffix == ".pkl":
        return cls.from_pickle(file_path)
    raise DatabaseError(f"File {file} not supported")

@classmethod
def add(
    cls,
    owner: str,
    repo_name: str,
    db_name: str,
    branch: str = "main",
    local_meta: DatabaseMetadata | None = None,
) -> Database | None:
    r"""Add or update a database directly from a specified remote Git repository.

    This is the primary method for adding custom or official databases from GitHub. The procedure:

    1. Constructs raw GitHub URL endpoints for the repository's `.toml` metadata and `.gbk` GenBank files.
    2. Downloads and parses the remote TOML metadata to extract version information.
    3. Compares the remote version against local metadata (if available). If up-to-date, skips remaining steps and
       returns `None`.
    4. Downloads the raw GenBank file content over HTTP.
    5. Writes source files into a temporary directory and compiles them using
       [`from_genbank`][kaptive.db.core.Database.from_genbank].
    6. Serializes and caches the compiled [`Database`][kaptive.db.core.Database] object into the local storage
       directory.

    Args:
        owner (str): Owner or organization of the GitHub repository (e.g., `'klebgenomics'`).
        repo_name (str): Name of the GitHub repository (e.g., `'KpSC_surface_antigen_loci'`).
        db_name (str): Base name of the database files in the repository
            (e.g., `'Klebsiella_pneumoniae_Species_Complex_K'`).
        branch (str): Git branch name to fetch from. Defaults to `'main'`.
        local_meta (DatabaseMetadata | None): Pre-loaded metadata of local database installation.
            Defaults to `None`.

    Returns:
        Database | None: The newly compiled [`Database`][kaptive.db.core.Database] object if installed or updated,
            or `None` if the local version was already up-to-date.

    Raises:
        DatabaseError: If repository files are not found, network issues occur, or file compilation fails.

    See Also:
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`Database`][kaptive.db.core.Database],
        [`DatabaseMetadata`][kaptive.db.models.DatabaseMetadata],
        [`DatabaseError`][kaptive.db.models.DatabaseError]
    """
    fetched = cls._fetch_files(owner, repo_name, db_name, branch=branch, local_meta=local_meta)
    if fetched is None:
        return None
    return cls._compile_and_save(*fetched)

@classmethod
def get(cls, file_or_keyword: str | Path) -> Database:
    r"""Load a Database from a file path or resolve and load it by keyword.

    If `file_or_keyword` points to an existing file, it is loaded directly.
    Otherwise, it is treated as a keyword. If the keyword is not installed locally,
    it will be automatically downloaded and installed.

    Args:
        file_or_keyword (str | Path): File path or recognized database keyword.

    Returns:
        Database: The loaded [`Database`][kaptive.db.core.Database] instance.
    """
    from kaptive.db.core import Database

    try:
        file_path = Path(file_or_keyword)
        if file_path.is_file():
            return Database.load(file_path)
    except (TypeError, ValueError, OSError):
        pass

    try:
        return cls.load(str(file_or_keyword))
    except DatabaseError:
        result = cls.install(str(file_or_keyword))
        if isinstance(result, list):
            result = result[0]
        if result is None:
            return cls.load(str(file_or_keyword))
        return result

@classmethod
def install(cls, kwd: str | list[str]) -> Database | list[Database | None]:
    r"""Install known, officially supported databases by keyword.

    Looks up the repository details (owner, repo, database name) associated with the provided keyword(s)
    in the internal registry (`_KNOWN`) and delegates file retrieval
    and compilation to [`add`][kaptive.db.manager.DatabaseManager.add]. If `'all'` or a list of keywords is
    supplied, fetching is performed concurrently via a thread pool executor.

    Args:
        kwd (str | list[str]): The keyword(s) of the known database(s) to install (e.g., `'kpsc_k'`,
            `['kpsc_k', 'ab_k']`, or `'all'`).

    Returns:
        Database | list[Database | None]: For a single keyword, returns the compiled
            [`Database`][kaptive.db.core.Database] object (or `None` if already up-to-date). For a list of
            keywords or `'all'`, returns a list of compiled [`Database`][kaptive.db.core.Database] objects
            (or `None` for entries that were up-to-date).

    Raises:
        DatabaseError: If any keyword is not recognized in the list of known databases, or if network/parsing
            errors occur.

    See Also:
        [`known`][kaptive.db.manager.DatabaseManager.known],
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`add`][kaptive.db.manager.DatabaseManager.add],
        [`Database`][kaptive.db.core.Database],
        [`DatabaseError`][kaptive.db.models.DatabaseError]
    """
    if kwd == "all":
        kwd = list(cls._KNOWN.keys())

    if isinstance(kwd, list):

        def _fetch_one(k: str):
            if (known_info := cls._KNOWN.get(k, None)) is None:
                raise DatabaseError(f'"{k}" is not a known database, choose from {list(cls._KNOWN.keys())}')
            return cls._fetch_files(*known_info)

        with concurrent.futures.ThreadPoolExecutor() as executor:
            fetched_list = list(executor.map(_fetch_one, kwd))

        results = []
        for fetched in fetched_list:
            if fetched is None:
                results.append(None)
            else:
                results.append(cls._compile_and_save(*fetched))
        return results

    if (known_info := cls._KNOWN.get(kwd, None)) is None:
        raise DatabaseError(f'"{kwd}" is not a known database, choose from {list(cls._KNOWN.keys())}')
    return cls.add(*known_info)  # type: ignore

@classmethod
def installed(cls) -> list[str]:
    r"""Return a list of keywords for all currently installed databases.

    Scans the local storage directory for `.pkl` files and extracts their keywords from the file stems.

    Returns:
        list[str]: A list of database keywords corresponding to installed `.pkl` database files.
            Returns an empty list if no databases are installed or if the storage directory does not exist.

    See Also:
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`known`][kaptive.db.manager.DatabaseManager.known]
    """
    if not cls._DB_DIR.exists():
        return []
    return [p.stem for p in cls._DB_DIR.glob("*.pkl")]

@classmethod
def known(cls) -> list[str]:
    r"""Return a list of keywords for all currently known, officially supported databases.

    These databases can be installed directly by providing their keyword to
    [`install`][kaptive.db.manager.DatabaseManager.install].

    Returns:
        list[str]: A list of known database keywords (e.g., `['kpsc_k', 'kpsc_o', 'kosc_k', ...]`).

    See Also:
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`install`][kaptive.db.manager.DatabaseManager.install]
    """
    return list(cls._KNOWN.keys())

@classmethod
def load(cls, kwd: str) -> Database:
    r"""Load a locally installed, compiled database using its keyword.

    Reads and unpickles the serialized `.pkl` database file from the local cache directory.

    Args:
        kwd (str): The keyword identifier of the database to load (e.g., `'kpsc_k'`).

    Returns:
        Database: The deserialized [`Database`][kaptive.db.core.Database] instance.

    Raises:
        DatabaseError: If the specified database is not installed locally.

    See Also:
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`Database`][kaptive.db.core.Database],
        [`DatabaseError`][kaptive.db.models.DatabaseError]
    """
    return pickle.loads(cls._get_existing_db_path(kwd).read_bytes())

@classmethod
def reset(cls) -> None:
    r"""Remove all installed databases by deleting their compiled files from the local directory.

    This clears the user's `~/.kaptive` cache directory of any `.pkl` database files and `.json` metadata
    sidecar files, effectively uninstalling all downloaded and compiled databases.

    Returns:
        None

    See Also:
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`uninstall`][kaptive.db.manager.DatabaseManager.uninstall]
    """
    if cls._DB_DIR.exists():
        for file_path in cls._DB_DIR.glob("*.pkl"):
            file_path.unlink()
        for file_path in cls._DB_DIR.glob("*.json"):
            file_path.unlink()

@classmethod
def save(cls, db: Database) -> int:
    r"""Serialize and save a compiled Database object and its metadata to local storage.

    Saves the database as a `.pkl` file named `{keyword}.pkl` in the local cache directory (`_DB_DIR`).
    Also writes a companion `{keyword}.json` file containing the serialized metadata for fast version checking.

    Args:
        db (Database): The compiled [`Database`][kaptive.db.core.Database] object to save.

    Returns:
        int: The total number of bytes written to the `.pkl` database file.

    See Also:
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`Database`][kaptive.db.core.Database],
        [`DatabaseMetadata`][kaptive.db.models.DatabaseMetadata]
    """
    db_path = cls._get_db_path(db.metadata.keyword)
    db_path.with_suffix(".json").write_text(json.dumps(asdict(db.metadata)))
    return db_path.write_bytes(pickle.dumps(db, protocol=pickle.HIGHEST_PROTOCOL))

@classmethod
def uninstall(cls, kwd: str) -> None:
    r"""Uninstall a specific database by removing its compiled local `.pkl` and `.json` files.

    Args:
        kwd (str): The keyword of the database to uninstall (e.g., `'kpsc_k'`).

    Returns:
        None

    Raises:
        DatabaseError: If the specified database is not currently installed locally.

    See Also:
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`DatabaseError`][kaptive.db.models.DatabaseError]
    """
    db_path = cls._get_existing_db_path(kwd)
    db_path.unlink()
    if db_path.with_suffix(".json").exists():
        db_path.with_suffix(".json").unlink()

@classmethod
def update(cls, kwd: str | list[str] = "all") -> Generator[Database, None, None]:
    r"""Update installed databases by checking against their remote GitHub repositories.

    Extracts local metadata to determine the source repository and version, then checks the remote GitHub
    repository for a newer version. If a newer version is available, the source files (`.gbk` and `.toml`)
    are fetched, compiled into a new [`Database`][kaptive.db.core.Database] object, and saved to disk. When
    updating multiple databases or `"all"`, remote fetches are executed concurrently using a thread pool executor.

    Args:
        kwd (str | list[str]): The keyword(s) of the database to update (e.g., `'kpsc_k'`,
            `['kpsc_k', 'ab_k']`, or `'all'`). Defaults to `"all"`, which updates all currently installed databases.

    Yields:
        Database: The newly compiled [`Database`][kaptive.db.core.Database] object for each database that
            required an update. Databases that are already up-to-date yield nothing.

    Raises:
        DatabaseError: If a requested database is not installed locally, or if network/parsing failures occur
            during update.

    See Also:
        [`installed`][kaptive.db.manager.DatabaseManager.installed],
        [`add`][kaptive.db.manager.DatabaseManager.add],
        [`DatabaseManager`][kaptive.db.manager.DatabaseManager],
        [`Database`][kaptive.db.core.Database],
        [`DatabaseMetadata`][kaptive.db.models.DatabaseMetadata],
        [`DatabaseError`][kaptive.db.models.DatabaseError]
    """
    if kwd == "all":
        kwd = cls.installed()
        if not kwd:
            return

    if isinstance(kwd, list):

        def _fetch_update_one(k: str):
            db_path = cls._get_existing_db_path(k)
            json_path = db_path.with_suffix(".json")
            if json_path.is_file():
                meta = DatabaseMetadata.from_dict(json.loads(json_path.read_text()))
            else:
                meta = pickle.loads(db_path.read_bytes()).metadata
            db_name = Path(meta.genbank).with_suffix("").name
            return cls._fetch_files(meta.owner, meta.repo, db_name, branch=meta.branch, local_meta=meta)

        with concurrent.futures.ThreadPoolExecutor() as executor:
            fetched_list = list(executor.map(_fetch_update_one, kwd))

        for fetched in fetched_list:
            if fetched is not None:
                yield cls._compile_and_save(*fetched)
    else:
        db_path = cls._get_existing_db_path(kwd)
        json_path = db_path.with_suffix(".json")
        if json_path.is_file():
            meta = DatabaseMetadata.from_dict(json.loads(json_path.read_text()))
        else:
            meta = pickle.loads(db_path.read_bytes()).metadata
        db_name = Path(meta.genbank).with_suffix("").name
        if (res := cls.add(meta.owner, meta.repo, db_name, branch=meta.branch, local_meta=meta)) is not None:
            yield res

@classmethod
def from_dict(cls, data: dict) -> "DatabaseMetadata":  # type: ignore
    r"""Instantiates a `DatabaseMetadata` object from a dictionary.

    Validates required fields, casts numeric types, and sets default fallback dictionaries for
    phenotype logic and antigenic units.

    Args:
        data (dict): Dictionary containing metadata fields (e.g., parsed from JSON or TOML).

    Returns:
        DatabaseMetadata: Validated [`DatabaseMetadata`][kaptive.db.models.DatabaseMetadata] instance.

    Raises:
        DatabaseError: If `data` is not a dict, missing required keys, or contains invalid attribute types.
    """
    if not isinstance(data, dict):
        raise DatabaseError("Metadata must be a dictionary.")

    try:
        meta = cls(
            name=data["name"],
            keyword=data["keyword"],
            genbank=data["genbank"],
            organism=data["organism"],
            taxon=int(data["taxon"]),
            antigen=data["antigen"],
            pathway=data["pathway"],
            version=data["version"],
            id_threshold=float(data["id_threshold"]),
            doi=data["doi"],
            owner=data["owner"],
            repo=data["repo"],
            branch=data["branch"],
            contact=data["contact"],
            phenotype_logic=data.get("phenotype_logic", data.get("logic", {})),
            antigenic_units=data.get("antigenic_units", data.get("units", {})),
        )
    except KeyError as e:
        raise DatabaseError(f"Metadata is missing required field: {e.args[0]!r}")
    except ValueError as e:
        raise DatabaseError(f"Metadata has an invalid value type: {e}")

    return meta

def __getitem__(self, item: int | slice | npt.NDArray[Any] | list[int]) -> "Any | Phenotypes":
    r"""Slices or masks the `Phenotypes` container batch along the primary dimension.

    Args:
        item (int | slice | npt.NDArray | list): Slice object, boolean mask array, or list of indices to select.

    Returns:
        Any | Phenotypes: A new [`Phenotypes`][kaptive.db.models.Phenotypes] container containing
            the selected subset of records.

    Raises:
        NotImplementedError: If a single integer index is provided, as scalar indexing is not supported
            on SoA containers.
    """
    if isinstance(item, (int, np.integer)):
        raise NotImplementedError("Single item access not implemented for Phenotypes")
    return Phenotypes(
        ids=self.ids[item],
        locus_masks=self.locus_masks[item],
        extra_masks=self.extra_masks[item],
        inactive_masks=self.inactive_masks[item],
        extra_counts=self.extra_counts[item],
        priorities=self.priorities[item],
        as_suffix=self.as_suffix[item],
    )

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

    Returns:
        int: Number of phenotype records.
    """
    return len(self.ids)