genesis/api_clients/ncbi_api.py

# genesis/api_clients/ncbi_api.py
import os
import requests
from urllib.parse import urlencode

NCBI_API_KEY = os.getenv("NCBI_API_KEY")
NCBI_BASE_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/"

if not NCBI_API_KEY:
    raise ValueError("Missing NCBI_API_KEY in environment variables")


def _ncbi_request(endpoint: str, params: dict):
    """
    Generic helper for calling NCBI E-Utilities API.
    """
    params["api_key"] = NCBI_API_KEY
    url = f"{NCBI_BASE_URL}{endpoint}?{urlencode(params)}"
    res = requests.get(url)
    res.raise_for_status()
    return res.text


def search_pubmed(query: str, max_results: int = 10):
    """
    Search PubMed articles via NCBI.
    """
    params = {"db": "pubmed", "term": query, "retmax": max_results, "retmode": "json"}
    res = requests.get(f"{NCBI_BASE_URL}esearch.fcgi", params=params)
    res.raise_for_status()
    ids = res.json().get("esearchresult", {}).get("idlist", [])
    return ids


def fetch_pubmed_details(pubmed_ids: list):
    """
    Fetch PubMed details given a list of IDs.
    """
    if not pubmed_ids:
        return []
    params = {
        "db": "pubmed",
        "id": ",".join(pubmed_ids),
        "retmode": "xml"
    }
    res = requests.get(f"{NCBI_BASE_URL}efetch.fcgi", params=params)
    res.raise_for_status()
    return res.text  # XML response


def search_gene(query: str):
    """
    Search for a gene in NCBI Gene database.
    """
    params = {"db": "gene", "term": query, "retmode": "json", "retmax": 5}
    res = requests.get(f"{NCBI_BASE_URL}esearch.fcgi", params=params)
    res.raise_for_status()
    return res.json()


def fetch_gene_summary(gene_id: str):
    """
    Get detailed gene summary from NCBI.
    """
    params = {"db": "gene", "id": gene_id, "retmode": "json"}
    res = requests.get(f"{NCBI_BASE_URL}esummary.fcgi", params=params)
    res.raise_for_status()
    return res.json()


def search_protein(query: str):
    """
    Search proteins in NCBI Protein database.
    """
    params = {"db": "protein", "term": query, "retmode": "json", "retmax": 5}
    res = requests.get(f"{NCBI_BASE_URL}esearch.fcgi", params=params)
    res.raise_for_status()
    return res.json()


def search_pubchem_compound(query: str):
    """
    Search compounds in PubChem via NCBI.
    """
    params = {"db": "pccompound", "term": query, "retmode": "json", "retmax": 5}
    res = requests.get(f"{NCBI_BASE_URL}esearch.fcgi", params=params)
    res.raise_for_status()
    return res.json()


def search_clinical_trials(query: str):
    """
    Search clinical trials in ClinicalTrials.gov via NCBI.
    """
    params = {"db": "clinicaltrials", "term": query, "retmode": "json", "retmax": 5}
    res = requests.get(f"{NCBI_BASE_URL}esearch.fcgi", params=params)
    res.raise_for_status()
    return res.json()