cdm_data_loaders.parsers.uniprot.uniprot_kb ¶

def parse_identifiers(entry: Mapping[str, Any]) -> list[dict[str, Any]]:
    """Parse the UniProt accessions from an XSD-converted entry dictionary.

    :param entry: XSD-converted entry dictionary
    :type entry: Mapping[str, Any]
    :return: list of identifiers in "db" / "xref" format
    :rtype: list[dict[str, Any]]
    """
    return [{DB: "UniProt", XREF: acc, DESCRIPTION: "UniProt accession"} for acc in _extract_accessions(entry)]

def parse_names(entry: Mapping[str, Any]) -> list[dict[str, Any]]:
    """Parse the names from an XSD-converted entry dictionary.

    :param entry: XSD-converted entry dictionary
    :type entry: Mapping[str, Any]
    :return: list of names with a description of the name type
    :rtype: list[dict[str, Any]]
    """
    name_list: list[dict[str, Any]] = [
        {NAME: n.strip(), DESCRIPTION: "UniProt entry name"}
        for n in entry.get("name") or []
        if isinstance(n, str) and n.strip()
    ]
    protein = entry.get("protein")
    if isinstance(protein, Mapping):
        for name_type in ("recommended", "alternative", "submitted"):
            name_groups = list(_iter_mappings(protein.get(f"{name_type}Name")))
            for name_length in ("full", "short"):
                for group in name_groups:
                    for name_obj in _iter_mappings(group.get(f"{name_length}Name")):
                        text = name_obj.get("#text")
                        if isinstance(text, str) and text.strip():
                            name_list.append(
                                {
                                    NAME: text.strip(),
                                    DESCRIPTION: f"UniProt {name_type} {name_length} name",
                                }
                            )

    # gene names
    name_list.extend(
        {
            NAME: gene_name.get("#text"),
            DESCRIPTION: f"UniProt gene name, {gene_name.get('_type')}",
        }
        for gene in _iter_mappings(entry.get("gene"))
        for gene_name in _iter_mappings(gene.get("name"))
    )

    return name_list

def parse_cross_references(entry: Mapping[str, Any]) -> list[dict[str, Any]]:
    """Parse the dbReferences from an XSD-converted entry dictionary.

    :param entry: XSD-converted entry dictionary
    :type entry: Mapping[str, Any]
    :return: list of cross references in "db" / "xref" / "description" / "relationship" format
    :rtype: list[dict[str, Any]]
    """
    refs: list[dict[str, Any]] = []
    entry_acc: str | None = None
    for dbxref in _iter_mappings(entry.get("dbReference")):
        db: str | None = dbxref.get("_type")
        xref: str | None = dbxref.get("_id")
        if not db or not xref:
            continue
        molecules = [mol_id for mol in _iter_mappings(dbxref.get("molecule")) if (mol_id := mol.get("_id"))]
        if len(molecules) > 1:
            if not entry_acc:
                accessions = _extract_accessions(entry)
                entry_acc = accessions[0] if accessions else None
            logger.warning("Found many molecules for entry %s", entry_acc)
        description_suffix = f" for UniProt:{molecules[0]}" if molecules else ""

        prop_list = [
            {"type": p.get("_type"), "value": p.get("_value")}
            for p in _iter_mappings(dbxref.get("property"))
            if p.get("_type") is not None and p.get("_value") is not None
        ]
        prop_list_dispatch_hash = {
            "Ensembl": parse_ensembl_dbxref,
            # embl/genbank
            "EMBL": parse_embl_dbxref,
            "RefSeq": parse_refseq_dbxref,
            "MANE-Select": parse_mane_dbxref,
        }

        if prop_list and db in prop_list_dispatch_hash:
            refs.extend(prop_list_dispatch_hash[db](xref, prop_list, description_suffix))
            continue

        if db == "GO":
            refs.extend([{DB: "GO", XREF: xref.removeprefix("GO:")}])
            continue

        if description_suffix:
            refs.extend([{DB: db, XREF: xref, DESCRIPTION: f"{db} ID{description_suffix}"}])
            continue

        # default parse
        refs.extend([{DB: db, XREF: xref}])
    return refs

def parse_ensembl_dbxref(xref: str, prop_list: list[dict[str, str]], description_suffix: str) -> list[dict[str, str]]:
    """Parse an Ensembl dbxref.

    :param xref: primary Ensembl cross-reference
    :type xref: str
    :param prop_list: property dicts attached to the dbReference
    :type prop_list: list[dict[str, str]]
    :param description_suffix: optional molecule suffix
    :type description_suffix: str
    :return: list of parsed Ensembl cross-reference dicts
    :rtype: list[dict[str, str]]
    """
    return _parse_typed_dbxref("ensembl", "Ensembl", _ensembl_type(xref), xref, prop_list, description_suffix)

def parse_refseq_dbxref(xref: str, prop_list: list[dict[str, str]], description_suffix: str) -> list[dict[str, str]]:
    """Parse a RefSeq dbxref.

    :param xref: primary RefSeq cross-reference
    :type xref: str
    :param prop_list: property dicts attached to the dbReference
    :type prop_list: list[dict[str, str]]
    :param description_suffix: optional molecule suffix
    :type description_suffix: str
    :return: list of parsed RefSeq cross-reference dicts
    :rtype: list[dict[str, str]]
    """
    return _parse_typed_dbxref("refseq", "RefSeq", "protein sequence", xref, prop_list, description_suffix)

def parse_embl_dbxref(xref: str, prop_list: list[dict[str, str]], description_suffix: str) -> list[dict[str, str]]:
    """Parse an EMBL / GenBank dbxref.

    :param xref: primary EMBL cross-reference
    :type xref: str
    :param prop_list: property dicts attached to the dbReference
    :type prop_list: list[dict[str, str]]
    :param description_suffix: optional molecule suffix
    :type description_suffix: str
    :return: list of parsed EMBL/GenBank cross-reference dicts
    :rtype: list[dict[str, str]]
    """
    mol_types = [p["value"] for p in prop_list if p["type"] == "molecule type"]
    description = f"EMBL/GenBank {mol_types[0]} ID" if mol_types and len(mol_types) == 1 else "EMBL/GenBank ID"
    return [
        {DB: "genbank", XREF: xref, DESCRIPTION: f"{description}{description_suffix}"},
        *[
            {DB: "genbank", XREF: p["value"], DESCRIPTION: f"EMBL/GenBank {p['type']}{description_suffix}"}
            for p in prop_list
            if p["type"] not in ("molecule type", "status")
        ],
    ]

def parse_mane_dbxref(xref: str, prop_list: list[dict[str, str]], description_suffix: str) -> list[dict[str, str]]:
    """Parse a MANE-Select dbxref.

    :param xref: primary MANE-Select cross-reference
    :type xref: str
    :param prop_list: property dicts attached to the dbReference
    :type prop_list: list[dict[str, str]]
    :param description_suffix: optional molecule suffix
    :type description_suffix: str
    :return: list of parsed MANE-Select cross-reference dicts
    :rtype: list[dict[str, str]]
    """
    likely_type = _ensembl_type(xref)
    refs = [
        {
            DB: "ensembl",
            XREF: xref,
            DESCRIPTION: f"Ensembl {likely_type} ID{description_suffix}, via MANE-Select",
        }
    ]
    for p in prop_list:
        if p["type"].startswith("RefSeq "):
            refs.append(
                {
                    DB: "refseq",
                    XREF: p["value"],
                    DESCRIPTION: f"{p['type']}{description_suffix}, via MANE-Select",
                }
            )
        elif p["value"].startswith("ENS"):
            refs.append(
                {
                    DB: "ensembl",
                    XREF: p["value"],
                    DESCRIPTION: f"Ensembl {p['type']}{description_suffix}, via MANE-Select",
                }
            )
    return refs

def parse_protein_info(entry: Mapping[str, Any]) -> dict[str, Any]:
    """Extract information for the protein table from an XSD-converted entry dictionary.

    :param entry: XSD-converted entry dictionary
    :type entry: Mapping[str, Any]
    :return: dictionary of protein properties
    :rtype: dict[str, Any]
    """
    protein: dict[str, Any] = {}
    protein_existence = entry.get("proteinExistence")
    if isinstance(protein_existence, Mapping) and protein_existence.get("_type") is not None:
        protein["evidence_for_existence"] = protein_existence.get("_type")

    seq_elem = entry.get("sequence")
    if isinstance(seq_elem, Mapping):
        protein.update(
            {
                "length": int(seq_elem.get("_length") or 0),
                "hash": seq_elem.get("_checksum"),
            }
        )
        sequence = seq_elem.get("#text")
        if isinstance(sequence, str) and sequence:
            protein["sequence"] = sequence
    return protein

def parse_organism(entry: Mapping[str, Any]) -> list[dict[str, str]]:
    """Parse the organism taxon designation from an XSD-converted entry dictionary.

    :param entry: XSD-converted entry dictionary
    :type entry: Mapping[str, Any]
    :return: list of organism taxon cross-reference dicts
    :rtype: list[dict[str, str]]
    """
    organism = entry.get("organism")
    if isinstance(organism, Mapping):
        for taxon_ref in _iter_mappings(organism.get("dbReference")):
            if taxon_ref.get("_type") == "NCBI Taxonomy" and (taxon_id := taxon_ref.get("_id")):
                return [
                    {
                        DB: NCBI_TAXON,
                        XREF: taxon_id,
                        DESCRIPTION: "UniProt taxon designation",
                        "relationship": "RO:0002162: in taxon",
                    }
                ]
    return []

def parse_references(entry: Mapping[str, Any]) -> list[dict[str, str]]:
    """Parse the publications for an entry and extract any that have prioritized database identifiers.

    :param entry: the XSD-converted UniProt entry being parsed
    :type entry: Mapping[str, Any]
    :return: list of publications associated with the entry
    :rtype: list[dict[str, str]]
    """
    publications: set[str] = set()
    for reference in _iter_mappings(entry.get("reference")):
        citation = reference.get("citation")
        if not isinstance(citation, Mapping):
            continue
        citation_refs = [
            {DB: str(ref_type).lower(), XREF: str(ref_id)}
            for ref in _iter_mappings(citation.get("dbReference"))
            if (ref_type := ref.get("_type")) and (ref_id := ref.get("_id"))
        ]
        if not citation_refs:
            continue

        all_ref_types = {c[DB] for c in citation_refs}
        ref_db_priority_order = ["doi", "pmcid", "pmid", "pubmed", "agricola"]
        unexpected_types = all_ref_types - set(ref_db_priority_order)
        if unexpected_types:
            # oh no!!
            logger.warning("Unexpected dbxref types in publications: %s", ", ".join(sorted(unexpected_types)))

        ref_found = False
        for ref_type in ref_db_priority_order:
            refs = [ref[XREF] for ref in citation_refs if ref[DB] == ref_type]
            if refs:
                publications.add(f"{PREFIX_TRANSLATION.get(ref_type, ref_type.upper())}:{refs[0]}")
                ref_found = True
                break

        # oh no!!
        if not ref_found:
            logger.warning("Could not find priority ref type for reference: %s", reference)

    return [{"publication_id": pub} for pub in publications]

def parse_uniprot_entry(
    entry: Mapping[str, Any],
    timestamp: dt.datetime,
) -> dict[str, list[dict[str, Any]]]:
    """Parse an XSD-converted UniProt entry dictionary into CDM schema tables.

    :param entry: an XSD-converted entry dictionary from a UniProt XML file
    :type entry: Mapping[str, Any]
    :param timestamp: current UTC timestamp
    :type timestamp: dt.datetime
    :return: dictionary of parsed data, indexed by table name
    :rtype: dict[str, list[dict[str, Any]]]
    """
    try:
        accessions = _extract_accessions(entry)
        if not accessions:
            logger.error("No accession found for entry")
            return make_rejected_table_result("entry", entry, "No accession found")

        first_acc = accessions[0]
        uniprot_id = f"uniprot:{first_acc}"
        raw_version = entry.get("_version")
        entity = {
            ENTITY_ID: uniprot_id,
            "entity_type": "protein",
            "data_source_entity_id": first_acc,
            "data_source_created": _format_date(entry.get("_created")),
            "data_source_modified": _format_date(entry.get("_modified")),
            "data_source_entity_version": str(raw_version) if raw_version is not None else None,
            "data_source_id": None,  # SwissProt / TrEMBL
            "created": None,
            "data_source": f"UniProt/{entry.get('_dataset')}",
            "updated": timestamp,
        }

        return {
            "entity": [entity],
            "identifier": [
                {ENTITY_ID: uniprot_id, **e}
                for e in [*parse_identifiers(entry), *parse_cross_references(entry), *parse_organism(entry)]
            ],
            "name": [{ENTITY_ID: uniprot_id, **e} for e in parse_names(entry)],
            "protein": [{"protein_id": uniprot_id, **parse_protein_info(entry)}],
            "entity_x_publication": [{ENTITY_ID: uniprot_id, **e} for e in parse_references(entry)],
        }
    except Exception as e:
        logger.exception("Parse error while processing entry")
        return make_rejected_table_result("entry", entry, str(e))

def make_uniprot_cdm_transform(
    timestamp: dt.datetime | None = None,
) -> RecordTransformFn:
    """Create a ``RecordTransformFn`` that transforms XSD-parsed UniProt ``entry`` dicts into CDM tables.

    :param timestamp: UTC timestamp assigned to ``entity.updated`` (defaults to ``datetime.now(UTC)``)
    :type timestamp: dt.datetime | None
    :return: ``RecordTransformFn`` transforming ``entry`` records into CDM table rows
    :rtype: RecordTransformFn
    """
    resolved_ts = timestamp if timestamp is not None else dt.datetime.now(dt.UTC)
    return functools.partial(parse_uniprot_entry, timestamp=resolved_ts)