Update genesis/api_clients/bioportal_api.py

genesis/api_clients/bioportal_api.py

@@ -1,81 +1,78 @@
 # genesis/api_clients/bioportal_api.py
-import os
 import requests
+import os
 from urllib.parse import quote
 
-BIOPORTAL_API_KEY = os.getenv("BIOPORTAL_API_KEY")
+BIOPORTAL_API_KEY = os.getenv("BIOPORTAL_API_KEY")  # Store this in Hugging Face secrets
 BIOPORTAL_BASE_URL = "https://data.bioontology.org"
 
 if not BIOPORTAL_API_KEY:
-    raise ValueError("Missing BIOPORTAL_API_KEY in environment variables")
-
+    raise ValueError("BIOPORTAL_API_KEY not found in environment variables.")
 
-def _bioportal_get(endpoint: str, params: dict = None):
-    """
-    Generic BioPortal GET request helper.
-    """
-    headers = {"Authorization": f"apikey token={BIOPORTAL_API_KEY}"}
-    url = f"{BIOPORTAL_BASE_URL}{endpoint}"
-    res = requests.get(url, headers=headers, params=params)
-    res.raise_for_status()
-    return res.json()
+HEADERS = {
+    "Authorization": f"apikey token={BIOPORTAL_API_KEY}"
+}
 
 
-def search_terms(query: str, ontologies: list = None, pagesize: int = 10):
+def search_concept(term: str, ontology: str = None, max_results: int = 10):
     """
-    Search for biomedical terms in BioPortal.
+    Search for a biomedical concept across BioPortal ontologies.
+    
+    Args:
+        term (str): The search term (e.g., "BRCA1", "glioblastoma").
+        ontology (str): Optional ontology acronym (e.g., "NCIT", "SNOMEDCT").
+        max_results (int): Maximum number of results.
     """
     params = {
-        "q": query,
-        "pagesize": pagesize
+        "q": term,
+        "pagesize": max_results
     }
-    if ontologies:
-        params["ontologies"] = ",".join(ontologies)
-    return _bioportal_get("/search", params)
+    if ontology:
+        params["ontology"] = ontology
 
+    url = f"{BIOPORTAL_BASE_URL}/search"
+    response = requests.get(url, headers=HEADERS, params=params)
+    response.raise_for_status()
+    return response.json().get("collection", [])
 
-def get_term_details(ontology_acronym: str, term_id: str):
-    """
-    Retrieve details for a specific ontology term.
-    """
-    encoded_term_id = quote(term_id, safe="")
-    return _bioportal_get(f"/ontologies/{ontology_acronym}/classes/{encoded_term_id}")
 
-
-def map_to_standard_ontology(term: str, ontologies: list = None):
+def get_concept_details(ontology_acronym: str, concept_id: str):
     """
-    Map a free-text biomedical term to standardized ontology concepts.
+    Fetch detailed information about a concept given ontology acronym and concept ID.
     """
-    search_results = search_terms(term, ontologies=ontologies, pagesize=1)
-    if search_results.get("collection"):
-        top_hit = search_results["collection"][0]
-        return {
-            "label": top_hit.get("prefLabel"),
-            "ontology": top_hit.get("links", {}).get("ontology"),
-            "definition": top_hit.get("definition"),
-            "synonyms": top_hit.get("synonym", []),
-            "cui": top_hit.get("@id")
-        }
-    return None
+    encoded_id = quote(concept_id, safe="")
+    url = f"{BIOPORTAL_BASE_URL}/ontologies/{ontology_acronym}/classes/{encoded_id}"
+    response = requests.get(url, headers=HEADERS)
+    response.raise_for_status()
+    return response.json()
 
 
-def search_disease_to_gene(disease_term: str):
+def get_ontologies():
     """
-    Example: find genes linked to a disease using ontology mappings.
+    List all available ontologies in BioPortal.
     """
-    # First, normalize the disease term
-    mapped = map_to_standard_ontology(disease_term, ontologies=["DOID", "MONDO"])
-    if not mapped:
-        return None
-
-    # For demo, this function could be extended to query
-    # linked datasets or internal Neo4j graph for related genes
-    return mapped
+    url = f"{BIOPORTAL_BASE_URL}/ontologies"
+    response = requests.get(url, headers=HEADERS)
+    response.raise_for_status()
+    return [
+        {"acronym": o.get("acronym"), "name": o.get("name"), "description": o.get("description")}
+        for o in response.json()
+    ]
 
 
-def search_pathway_for_disease(disease_term: str):
+def map_term_across_ontologies(term: str, ontologies: list = None):
     """
-    Example: find pathways linked to a disease term.
+    Map a term across multiple ontologies to unify synonyms and cross-references.
     """
-    mapped = map_to_standard_ontology(disease_term, ontologies=["PW", "Reactome"])
-    return mapped
+    mappings = {}
+    target_ontologies = ontologies or [o["acronym"] for o in get_ontologies()]
+    
+    for ont in target_ontologies:
+        results = search_concept(term, ontology=ont, max_results=1)
+        if results:
+            mappings[ont] = {
+                "label": results[0].get("prefLabel"),
+                "uri": results[0].get("@id"),
+                "synonyms": results[0].get("synonym", []),
+            }
+    return mappings