cdm_data_loaders.parsers.uniprot.uniref ¶

def generate_dbxref(db: str, acc: str) -> str:
    """Generate a database reference that uses BioRegistry prefixes."""
    return f"{PREFIX_TRANSLATION.get(db, db)}:{acc}"

def extract_cluster(entry: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    """Extract the UniRef cluster ID and name from an XSD-converted entry dictionary."""
    entry_id = entry.get("_id")
    if not entry_id:
        return ({}, {})
    cluster_id = f"uniref:{entry_id}"
    updated_str = _format_date(entry.get("_updated"))
    cluster = {
        "cluster_id": cluster_id,
        "name": entry.get("name"),
        "cluster_type": "Protein",
        "description": None,
        # fill in protocol afterwards
    }
    entity = {
        "entity_id": cluster_id,
        "entity_type": "Cluster",
        "data_source": UNIREF,
        "data_source_entity_id": entry_id,
        "data_source_updated": updated_str,
        # fill in other columns later
    }

    return (cluster, entity)

def extract_cross_refs(
    dbref: Mapping[str, Any],
    cluster_id: str,
    is_representative: bool,  # noqa: FBT001
) -> dict[str, Any]:
    """Extract UniRef90/50/UniParc cross references from a single ``dbReference`` dictionary."""
    entity_db = dbref.get("_type")
    entity_xref = dbref.get("_id")
    if not entity_xref or not entity_db:
        # should never happen!
        logger.warning("Incomplete dbReference in %s", cluster_id)
        return {}

    entity_id = generate_dbxref(entity_db, entity_xref)
    properties = list(_iter_mappings(dbref.get("property")))

    # if this is a UniProtKB ID, check whether there is a UniProtKB accession
    if entity_db == "UniProtKB ID":
        uniprot_id = next(
            (prop.get("_value") for prop in properties if prop.get("_type") == "UniProtKB accession"),
            None,
        )
        if uniprot_id:
            entity_id = f"uniprot:{uniprot_id}"
        else:
            logger.warning("Could not find UniProt accession for UniProtKB ID %s", entity_xref)

    # is this a seed member?
    is_seed_val = next(
        (prop.get("_value") for prop in properties if prop.get("_type") == "isSeed"),
        None,
    )
    is_seed = is_seed_val == "true"

    return {
        "entity_id": entity_id,
        "cluster_id": cluster_id,
        "is_representative": is_representative,
        "is_seed": is_seed,
    }

def parse_uniref_entry(
    entry: Mapping[str, Any],
    timestamp: dt.datetime,
    uniref_variant: str,
) -> dict[str, list[dict[str, Any]]]:
    """Parse a single XSD-converted UniRef ``entry`` dictionary into CDM-friendly rows."""
    # Cluster basic info
    (cluster, entity) = extract_cluster(entry)
    cluster_id = cluster.get("cluster_id")
    if not cluster_id:
        logger.debug("No cluster ID found in entry")
        return make_rejected_table_result("entry", entry, "No cluster ID found in entry")

    # add the current timestamp
    entity["updated"] = timestamp
    # add in the protocol (which version of UniRef this is)
    cluster["protocol"] = uniref_variant

    # Cross references from representative and members
    cluster_members = [
        *(
            extract_cross_refs(repr_db, cluster_id, is_representative=True)
            for repr_member in _iter_mappings(entry.get("representativeMember"))
            for repr_db in _iter_mappings(repr_member.get("dbReference"))
        ),
        *(
            extract_cross_refs(mem_db, cluster_id, is_representative=False)
            for member in _iter_mappings(entry.get("member"))
            for mem_db in _iter_mappings(member.get("dbReference"))
        ),
    ]

    return {
        "entity": [entity],
        "cluster": [cluster],
        "clustermember": [c for c in cluster_members if c],
    }

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

    :param uniref_variant: protocol label for the UniRef variant (e.g. ``"UniRef 50"``)
    :type uniref_variant: str
    :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
    """
    resolved_ts = timestamp if timestamp is not None else dt.datetime.now(dt.UTC)
    return functools.partial(
        parse_uniref_entry,
        timestamp=resolved_ts,
        uniref_variant=uniref_variant,
    )