"""trials_for_genes — ClinicalTrials.gov studies whose intervention matches the gene. Memo decision 3: ToolUniverse for interactive lookups; lit-agent's own harvester keeps the weekly digest. ToolUniverse narrows an intervention search to the intervention's name and other name (Phase 0 finding 5: "CD40" gives the one pancreas-specific mitazalimab trial, where a broad API query also returns two solid-tumour basket trials). That rewrite is helpful but silent, so every payload states it: ``query_semantics`` in words, and ``executed_queries`` with the exact query ClinicalTrials.gov ran, taken from the response itself and pinned by a recorded fixture. ClinicalTrials.gov's terms ask a redistributor to show the date ClinicalTrials.gov processed the data and to state any changes: once a search has answered, one more call reads that date (``dataTimestamp``, src/direct_calls.py) and the payload carries the notice of :mod:`src.ctgov_terms` in ``source_terms`` and as the narrative's last sentence. Each row also carries ``ctgov_data_processed_on``, so a relay that shows only rows keeps the date. """ from __future__ import annotations from typing import Any from src import ctgov_terms, payloads from src.endpoints._common import Budget, Collector, according_to, get_client, stamped from src.identifiers import parse_disease, parse_gene_list ENDPOINT = "trials_for_genes" CT_SEARCH = "ClinicalTrials_search_studies" DEFAULT_CONDITION = "pancreatic cancer" DEFAULT_STATUS = "RECRUITING" STATUSES = ( "RECRUITING", "NOT_YET_RECRUITING", "ACTIVE_NOT_RECRUITING", "ENROLLING_BY_INVITATION", "COMPLETED", "ANY", ) PAGE_SIZE = 100 QUERY_SEMANTICS = ( "A study is listed when its intervention name or intervention other-name matches the gene " "symbol (ToolUniverse's rewrite of the ClinicalTrials.gov query). Studies that mention the " "gene only in their title, outcomes or eligibility are not listed, and a match on the name is " "not evidence that the intervention acts on the gene's product." ) def trials_for_genes(genes: str = "", condition: str = "", status: str = "") -> dict[str, Any]: parsed = parse_gene_list(genes, allow_ensembl=False) if not parsed.ok: return payloads.refused(ENDPOINT, parsed.reason, parsed.message) cond = parse_disease(condition or DEFAULT_CONDITION) if not cond.ok: return payloads.refused(ENDPOINT, cond.reason, cond.message) if cond.identifiers[0].kind != "disease_name": return payloads.refused( ENDPOINT, "invalid_identifier", "Give the condition as a name, e.g. 'pancreatic cancer'.", ) wanted = (status or DEFAULT_STATUS).strip().upper() if wanted not in STATUSES: return payloads.refused( ENDPOINT, "invalid_argument", f"status must be one of {', '.join(STATUSES)}; got {wanted[:24]!r}.", ) condition_name = cond.identifiers[0].value client, budget, out = get_client(), Budget(), Collector() executed: list[dict[str, Any]] = [] per_gene: list[tuple[str, int, int]] = [] symbols = [i.value for i in parsed.identifiers] for idx, gene in enumerate(symbols): if budget.spent: out.out_of_budget(symbols[idx:]) break args: dict[str, Any] = { "query_cond": condition_name, "query_intr": gene, "page_size": PAGE_SIZE, } if wanted != "ANY": args["filter_status"] = wanted result = out.call(client, budget, CT_SEARCH, args) if result is None: continue data = result.data executed.append({"gene": gene, "executed_query": data.get("executed_query")}) studies = data.get("studies") or [] for study in studies: nct = study["nct_id"] out.rows.append( stamped( { "gene": gene, "nct_id": nct, "title": study.get("brief_title"), "status": study.get("status"), "phases": study.get("phases") or [], "study_type": study.get("study_type"), "conditions": study.get("conditions") or [], "interventions": study.get("interventions") or [], "sponsor": study.get("sponsor"), "start_date": study.get("start_date"), "completion_date": study.get("completion_date"), }, result, ref=nct, url=f"https://clinicaltrials.gov/study/{nct}", ) ) per_gene.append((gene, len(studies), int(data.get("total_count") or len(studies)))) if not out.rows and out.source_errors and not per_gene: return payloads.error( ENDPOINT, "all_sources_failed", "ClinicalTrials.gov did not answer; nothing was looked up.", source_errors=out.source_errors, ) terms = _terms_notice(client, budget, out) if per_gene else None if terms is not None: for row in out.rows: row["ctgov_data_processed_on"] = terms["data_processed_on"] lead = according_to(["clinicaltrials"], out.retrieved) kind = "of any status" if wanted == "ANY" else wanted clauses = [] for gene, shown, total in per_gene: ids = [r["nct_id"] for r in out.rows if r["gene"] == gene] listed = f" ({', '.join(ids[:10])}" + ( f" and {len(ids) - 10} more" if len(ids) > 10 else "" ) listed += ")" if ids else "" more = f", showing the first {shown}" if total > shown else "" noun = "study" if total == 1 else "studies" clauses.append(f"{gene}: {total} {kind} {noun}{listed}{more}") sentences = [] if clauses: sentences.append( f"{lead}, studies for '{condition_name}' with an intervention matching each gene — " + "; ".join(clauses) + "." ) sentences.append("Matching is by intervention name only; see query_semantics.") if out.source_errors: sentences.append( "Source problems: " + "; ".join(f"{e['source']} {e['reason']}" for e in out.source_errors) + "." ) if terms is not None: sentences.append(terms["sentence"]) return payloads.ok( ENDPOINT, query={"genes": symbols, "condition": condition_name, "status": wanted}, rows=out.rows, sources_used=out.sources_used or {"clinicaltrials"}, narrative=" ".join(sentences), unresolved=out.unresolved, source_errors=out.source_errors, query_semantics=QUERY_SEMANTICS, executed_queries=executed, source_terms=[terms] if terms is not None else [], ) def _terms_notice(client, budget: Budget, out: Collector) -> dict[str, Any]: """The ClinicalTrials.gov notice, with the processed date read from ClinicalTrials.gov. A failure here never hides the studies already found: the notice says the date could not be read and why, and the failure is not a ``source_errors`` entry (the search itself answered). """ if budget.spent: return ctgov_terms.notice(None, "the call budget ran out before it could be read") result, err = out.attempt(client, budget, ctgov_terms.DATE_TOOL, {}) if err is not None: return ctgov_terms.notice(None, f"{err.reason}: {err.detail}") return ctgov_terms.notice(result.data["dataTimestamp"])