Ontology

The Ontology class retrieves GO term metadata and gene associations from the official Gene Ontology API. GO accessions are authoritative; phenotype text is used only as a display label or a locally registered alias.

Import

from neko._annotations.gene_ontology import Ontology

Quick example

from neko._annotations.gene_ontology import Ontology

onto = Ontology(taxon_id=9606)

# Structured records retain both the gene symbol and source identifier.
genes = onto.fetch_go_genes("GO:0062043")
print([(gene.symbol, gene.gene_id) for gene in genes])

# The backward-compatible helper returns symbols only.
markers = onto.get_markers(id_accession="GO:0062043")

By default, only associations whose object is the requested GO term are returned. Set include_descendants=True to include annotations propagated from more specific terms. The default taxon is human (NCBITaxon:9606). Numeric taxonomy IDs and complete NCBITaxon: CURIEs are both accepted.

Automatic assertions are included by default. Pass exclude_automatic_assertions=True to remove ECO:0000501 associations. This and the exact-term filter are enforced locally because deployments of the upstream API have not always applied their corresponding query flags.

HTTP, decoding, and response-schema failures raise GeneOntologyError. Unknown accessions raise GeneOntologyNotFoundError; a valid term with no matching genes returns an empty list.

---

Class reference

Ontology

Ontology(taxon_id=9606, timeout=30.0, user_agent='NeKo (https://github.com/sysbio-curie/Neko)', session=None)

class that stores some functionalities to connect phenotypes to nodes and to associate information for each node at tissue level

Source code in neko/_annotations/gene_ontology.py

def __init__(
        self,
        taxon_id=9606,
        timeout=30.0,
        user_agent=(
            "NeKo (https://github.com/sysbio-curie/Neko)"
        ),
        session=None,
    ):
    self.taxon_id = self._normalize_taxon_id(taxon_id)
    self.timeout = timeout
    self.session = session or _new_go_session(user_agent)
    self._term_cache = {}
    self.accession_to_phenotype_dict = {
        "GO:0010718": (
            "positive regulation of epithelial to mesenchymal transition"
        ),
    }

Methods:

get_term

get_term(go_id)

Return canonical metadata for a GO accession.

Source code in neko/_annotations/gene_ontology.py

def get_term(self, go_id):
    """Return canonical metadata for a GO accession."""
    go_id = self._normalize_go_id(go_id)
    if go_id in self._term_cache:
        return self._term_cache[go_id]

    payload = self._request_json(
        f"{_GO_API_BASE}/ontology/term/{go_id}",
        go_id,
    )
    if not isinstance(payload, dict):
        raise GeneOntologyError(
            "Unexpected GO term response: expected a JSON object."
        )

    returned_id = payload.get("goid")
    label = payload.get("label")
    try:
        returned_id = self._normalize_go_id(returned_id)
    except ValueError as exc:
        raise GeneOntologyError(
            f"GO term response for {go_id} has no valid accession."
        ) from exc
    if not isinstance(label, str) or not label.strip():
        raise GeneOntologyError(
            f"GO term response for {go_id} has no valid label."
        )

    term = GOTerm(returned_id, label.strip())
    self._term_cache[go_id] = term
    self._term_cache[returned_id] = term
    self.accession_to_phenotype_dict[returned_id] = term.label
    return term

fetch_go_genes

fetch_go_genes(go_id, *, taxon_id=None, include_descendants=False, exclude_automatic_assertions=False, page_size=_GO_PAGE_SIZE, max_pages=_GO_MAX_PAGES)

Fetch unique genes associated with a GO term.

Source code in neko/_annotations/gene_ontology.py

def fetch_go_genes(
        self,
        go_id,
        *,
        taxon_id=None,
        include_descendants=False,
        exclude_automatic_assertions=False,
        page_size=_GO_PAGE_SIZE,
        max_pages=_GO_MAX_PAGES,
    ):
    """Fetch unique genes associated with a GO term."""
    term = self.get_term(go_id)
    taxon_id = self.taxon_id if taxon_id is None else (
        self._normalize_taxon_id(taxon_id)
    )
    if not isinstance(page_size, int) or page_size <= 0:
        raise ValueError("page_size must be a positive integer.")
    if not isinstance(max_pages, int) or max_pages <= 0:
        raise ValueError("max_pages must be a positive integer.")
    page_size = min(page_size, _GO_PAGE_SIZE)

    url = (
        f"{_GO_API_BASE}/bioentity/function/"
        f"{term.go_id}/genes"
    )
    start = 0
    page_count = 0
    previous_signature = None
    genes_by_id = {}

    while page_count < max_pages:
        payload = self._request_json(
            url,
            term.go_id,
            params={
                "start": start,
                "rows": page_size,
                "taxon": taxon_id,
                "relationship_type": "involved_in",
            },
        )
        if not isinstance(payload, dict):
            raise GeneOntologyError(
                "Unexpected GO association response: expected a JSON "
                "object."
            )
        associations = payload.get("associations")
        if not isinstance(associations, list):
            raise GeneOntologyError(
                "Unexpected GO association response: 'associations' "
                "must be a list."
            )
        if not associations:
            break

        signature = tuple(
            (
                association.get("id")
                or (
                    (association.get("subject") or {}).get("id"),
                    (association.get("object") or {}).get("id"),
                    association.get("evidence"),
                    association.get("negated"),
                )
            )
            if isinstance(association, dict) else None
            for association in associations
        )
        if signature == previous_signature:
            raise GeneOntologyError(
                "GO API pagination did not advance."
            )
        previous_signature = signature

        for association in associations:
            if not isinstance(association, dict):
                raise GeneOntologyError(
                    "GO API returned a malformed association."
                )
            if association.get("negated", False):
                continue
            if (
                exclude_automatic_assertions
                and _IEA_EVIDENCE in self._association_evidence_ids(
                    association
                )
            ):
                continue

            subject = association.get("subject")
            obj = association.get("object")
            if not isinstance(subject, dict) or not isinstance(obj, dict):
                continue
            if not include_descendants and obj.get("id") != term.go_id:
                continue

            taxon = subject.get("taxon")
            if not isinstance(taxon, dict) or taxon.get("id") != taxon_id:
                continue
            gene_id = subject.get("id")
            symbol = subject.get("label")
            gene_id = gene_id if isinstance(gene_id, str) else None
            symbol = symbol if isinstance(symbol, str) else None
            if not gene_id and not symbol:
                continue

            key = gene_id or f"{taxon_id}:{symbol}"
            gene = GOGene(
                gene_id=gene_id,
                symbol=symbol,
                taxon_id=taxon_id,
                taxon_label=(
                    taxon.get("label")
                    if isinstance(taxon.get("label"), str)
                    else None
                ),
            )
            existing = genes_by_id.get(key)
            if existing is None or (
                existing.symbol is None and gene.symbol is not None
            ):
                genes_by_id[key] = gene

        page_count += 1
        start += len(associations)
        if len(associations) < page_size:
            break
    else:
        raise GeneOntologyError(
            f"GO API pagination exceeded {max_pages} pages for "
            f"{term.go_id}."
        )

    return sorted(
        genes_by_id.values(),
        key=lambda gene: (
            gene.symbol is None,
            (gene.symbol or gene.gene_id or "").casefold(),
        ),
    )

get_markers

get_markers(phenotype=None, id_accession=None, *, taxon_id=None, include_descendants=False, exclude_automatic_assertions=False)

Return gene symbols associated with a GO term.

Source code in neko/_annotations/gene_ontology.py

def get_markers(
        self,
        phenotype=None,
        id_accession=None,
        *,
        taxon_id=None,
        include_descendants=False,
        exclude_automatic_assertions=False,
    ):
    """Return gene symbols associated with a GO term."""
    id_accession = self.resolve_accession(
        phenotype=phenotype,
        id_accession=id_accession,
    )
    genes = self.fetch_go_genes(
        id_accession,
        taxon_id=taxon_id,
        include_descendants=include_descendants,
        exclude_automatic_assertions=exclude_automatic_assertions,
    )
    return sorted({gene.symbol for gene in genes if gene.symbol})

resolve_accession

resolve_accession(phenotype=None, id_accession=None)

Resolve an explicit accession or a registered phenotype alias.

Source code in neko/_annotations/gene_ontology.py

def resolve_accession(self, phenotype=None, id_accession=None):
    """Resolve an explicit accession or a registered phenotype alias."""
    if id_accession is not None:
        return self._normalize_go_id(id_accession)
    if phenotype is None:
        raise ValueError(
            "Provide at least one of id_accession or phenotype."
        )
    normalized_phenotype = _normalize_annotation_value(phenotype)
    matches = [
        accession
        for accession, description
        in self.accession_to_phenotype_dict.items()
        if _normalize_annotation_value(description)
        == normalized_phenotype
    ]
    if not matches:
        raise ValueError(
            "No locally registered GO accession was found for "
            f"phenotype {phenotype!r}."
        )
    id_accession = matches[0]
    return self._normalize_go_id(id_accession)

check_tissue_annotations

check_tissue_annotations(genes_df, tissue)

Check whether genes have detected HPA expression in a tissue.

Args: genes_df (DataFrame): DataFrame containing gene symbols. tissue (str): Tissue to match exactly after case/whitespace normalization.

Returns: DataFrame: Gene symbols and their detected-expression status.

Raises: AnnotationServiceError: If OmniPath is unavailable or returns an unexpected annotation schema.

Source code in neko/_annotations/gene_ontology.py

def check_tissue_annotations(self, genes_df, tissue):
    """
    Check whether genes have detected HPA expression in a tissue.

    Args:
    genes_df (DataFrame): DataFrame containing gene symbols.
    tissue (str): Tissue to match exactly after case/whitespace normalization.

    Returns:
    DataFrame: Gene symbols and their detected-expression status.

    Raises:
    AnnotationServiceError: If OmniPath is unavailable or returns an
        unexpected annotation schema.
    """

    if not isinstance(genes_df, pd.DataFrame):
        raise TypeError(
            "genes_df must be a pandas DataFrame containing a "
            "'Genesymbol' column."
        )
    if 'Genesymbol' not in genes_df:
        raise ValueError("genes_df must contain a 'Genesymbol' column.")
    if not isinstance(tissue, str) or not tissue.strip():
        raise ValueError("tissue must be a non-empty string.")

    raw_symbols = genes_df['Genesymbol']
    if raw_symbols.isna().any():
        raise ValueError("genes_df contains a missing gene symbol.")

    gene_symbols = raw_symbols.astype(str).tolist()
    if any(not symbol.strip() for symbol in gene_symbols):
        raise ValueError("genes_df contains an empty gene symbol.")
    if not gene_symbols:
        return pd.DataFrame({
            'Genesymbol': pd.Series(dtype='object'),
            'in_tissue': pd.Series(dtype='bool'),
        })

    unique_symbols = list(dict.fromkeys(gene_symbols))
    normalized_tissue = _normalize_annotation_value(tissue)

    if normalized_tissue in _HPA_CANCER_TISSUES:
        try:
            expressed = _hpa_cancer_expressed_genes(
                unique_symbols,
                tissue,
            )
        except Exception as exc:
            raise AnnotationServiceError(
                "Unable to retrieve cancer expression data from the "
                "Human Protein Atlas. No genes were classified as absent."
            ) from exc
    else:
        # A single resource-restricted request is faster, cacheable as one
        # unit, and less likely to fail than one request per gene.
        try:
            annotations_df = op.requests.Annotations.get(
                proteins=unique_symbols,
                resources=_HPA_TISSUE_RESOURCE,
            )
        except Exception as exc:
            raise AnnotationServiceError(
                "Unable to retrieve HPA tissue annotations from OmniPath. "
                "The service may be temporarily unavailable; no genes "
                "were classified as absent."
            ) from exc
        expressed = _expressed_genes(annotations_df, tissue)

    return pd.DataFrame({
        'Genesymbol': gene_symbols,
        'in_tissue': [
            _normalize_annotation_value(symbol) in expressed
            for symbol in gene_symbols
        ],
    })