File size: 4,776 Bytes
3ed6b66 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """pathways_for_genes — which Reactome pathways each gene's protein belongs to.
Membership, never enrichment: a row says "Reactome annotates this protein to this pathway", and
nothing is computed over the gene list. ``shared_by_all`` lists the pathways every resolved gene
belongs to — a set intersection over Reactome's own annotation, with no statistic attached.
Which pathways are *enriched* in a gene list is DecoupleRpy's question.
Symbols resolve through HGNC to UniProt accessions; accessions may also be given directly.
Reactome's UniProt mapping returns the lowest-level pathways (Phase 0 Q9: ISG15, RSAD2 and OAS1
in "Interferon alpha/beta signaling").
"""
from __future__ import annotations
from typing import Any
from src import payloads
from src.endpoints._common import (
Budget,
Collector,
according_to,
get_client,
resolution_rows,
resolve_symbols,
stamped,
)
from src.identifiers import parse_gene_list
ENDPOINT = "pathways_for_genes"
REACTOME_MAP = "Reactome_map_uniprot_to_pathways"
def pathways_for_genes(genes: str = "") -> dict[str, Any]:
parsed = parse_gene_list(genes, allow_uniprot=True, allow_ensembl=False)
if not parsed.ok:
return payloads.refused(ENDPOINT, parsed.reason, parsed.message)
client, budget, out = get_client(), Budget(), Collector()
symbols = [i.value for i in parsed.identifiers if i.kind == "symbol"]
resolved = resolve_symbols(symbols, client, budget, out)
proteins: list[tuple[str, str]] = [] # (gene label, UniProt accession)
for gene in resolved:
if not gene.uniprot_ids:
out.unresolved_input(
gene.input,
f"HGNC links no UniProt entry (locus type: {gene.locus_type or 'not stated'})",
)
proteins += [(gene.symbol, acc) for acc in gene.uniprot_ids]
proteins += [(i.value, i.value) for i in parsed.identifiers if i.kind == "uniprot"]
by_gene: dict[str, set[str]] = {}
names: dict[str, str] = {}
for label, acc in proteins:
result = out.call(client, budget, REACTOME_MAP, {"uniprot_id": acc}, unknown=("404", label))
if result is None:
continue
pathways = result.data or []
by_gene.setdefault(label, set())
for pw in pathways:
st_id = pw["stId"]
names[st_id] = pw["displayName"]
if st_id in by_gene[label]:
continue
by_gene[label].add(st_id)
out.rows.append(
stamped(
{
"gene": label,
"uniprot_accession": acc,
"pathway_id": st_id,
"pathway_name": pw["displayName"],
"is_in_disease": bool(pw.get("isInDisease")),
"species": pw.get("speciesName"),
},
result,
ref=st_id,
url=f"https://reactome.org/content/detail/{st_id}",
)
)
if not out.rows and out.source_errors and not by_gene:
return payloads.error(
ENDPOINT,
"all_sources_failed",
"No source answered; nothing was looked up.",
source_errors=out.source_errors,
)
shared = sorted(set.intersection(*by_gene.values())) if len(by_gene) > 1 else []
lead = according_to(["reactome"], out.retrieved)
sentences = [
f"{lead}, {label} is annotated to {len(ids)} lowest-level pathway"
f"{'s' if len(ids) != 1 else ''}."
for label, ids in by_gene.items()
]
if len(by_gene) > 1:
sentences.append(
f"Pathways all {len(by_gene)} genes belong to: "
+ (", ".join(f"{names[s]} ({s})" for s in shared) if shared else "none")
+ "."
)
sentences.append("This is membership, not enrichment: no statistic was computed.")
if out.unresolved:
sentences.append(
"Not resolved: "
+ "; ".join(f"{u['input']} ({u['reason']})" for u in out.unresolved)
+ "."
)
if out.source_errors:
sentences.append(
"Source problems: "
+ "; ".join(f"{e['source']} {e['reason']}" for e in out.source_errors)
+ "."
)
return payloads.ok(
ENDPOINT,
query={"genes": [i.value for i in parsed.identifiers]},
rows=out.rows,
sources_used=out.sources_used or {"reactome"},
narrative=" ".join(sentences),
unresolved=out.unresolved,
source_errors=out.source_errors,
resolved_identifiers=resolution_rows(resolved),
shared_by_all=[{"pathway_id": s, "pathway_name": names[s]} for s in shared],
)
|