avoigt1121
Claude Opus 5.5
feat: the five knowledge endpoints, gradio_ui and frozen descriptor (s2)
3ed6b66 | """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], | |
| ) | |