Update genesis/api_clients/chembl_api.py

genesis/api_clients/chembl_api.py

@@ -1,98 +1,88 @@
 # genesis/api_clients/chembl_api.py
-import requests
 import os
+import requests
+from typing import List, Dict, Optional
+
+CHEMBL_BASE = "https://www.ebi.ac.uk/chembl/api/data"
+
+def _fetch_paginated(endpoint: str, params: dict) -> List[dict]:
+    """
+    Handles paginated ChEMBL API responses.
+    """
+    results = []
+    url = f"{CHEMBL_BASE}/{endpoint}"
+    
+    while url:
+        r = requests.get(url, params=params)
+        r.raise_for_status()
+        data = r.json()
+        page_items = data.get(endpoint, [])
+        results.extend(page_items)
+        url = data.get("page_meta", {}).get("next")
+        params = {}  # clear params after first page to avoid duplication
+    return results
+
+
+def search_molecule_by_name(name: str, max_results: int = 10) -> List[dict]:
+    """
+    Search molecules in ChEMBL by name or synonym.
+    """
+    params = {"molecule_synonyms__icontains": name, "limit": max_results}
+    return _fetch_paginated("molecule", params)
+
+
+def get_molecule_details(chembl_id: str) -> dict:
+    """
+    Retrieve detailed information for a molecule by ChEMBL ID.
+    """
+    r = requests.get(f"{CHEMBL_BASE}/molecule/{chembl_id}")
+    r.raise_for_status()
+    return r.json()
 
-CHEMBL_BASE_URL = "https://www.ebi.ac.uk/chembl/api/data"
 
-def search_drug(drug_name: str, limit: int = 5):
+def get_molecule_bioactivities(chembl_id: str, max_results: int = 50) -> List[dict]:
     """
-    Search ChEMBL for drugs by name.
+    Retrieve bioactivity records for a given molecule.
     """
-    url = f"{CHEMBL_BASE_URL}/molecule/search.json?q={drug_name}&limit={limit}"
-    response = requests.get(url)
-    response.raise_for_status()
-    data = response.json()
-    return [
-        {
-            "molecule_chembl_id": mol.get("molecule_chembl_id"),
-            "pref_name": mol.get("pref_name"),
-            "max_phase": mol.get("max_phase"),
-            "therapeutic_flag": mol.get("therapeutic_flag"),
-            "molecule_type": mol.get("molecule_type"),
-        }
-        for mol in data.get("molecules", [])
-    ]
+    params = {"molecule_chembl_id": chembl_id, "limit": max_results}
+    return _fetch_paginated("activity", params)
 
 
-def get_drug_details(chembl_id: str):
+def search_target_by_name(name: str, max_results: int = 10) -> List[dict]:
     """
-    Retrieve detailed drug information by ChEMBL ID.
+    Search for biological targets by name.
     """
-    url = f"{CHEMBL_BASE_URL}/molecule/{chembl_id}.json"
-    response = requests.get(url)
-    response.raise_for_status()
-    return response.json()
+    params = {"target_synonym__icontains": name, "limit": max_results}
+    return _fetch_paginated("target", params)
 
 
-def get_target_info(chembl_id: str):
+def get_target_details(chembl_id: str) -> dict:
     """
-    Get targets for a given drug (by ChEMBL ID).
+    Retrieve detailed target information by ChEMBL ID.
     """
-    url = f"{CHEMBL_BASE_URL}/target.json?molecule_chembl_id={chembl_id}"
-    response = requests.get(url)
-    response.raise_for_status()
-    return response.json().get("targets", [])
+    r = requests.get(f"{CHEMBL_BASE}/target/{chembl_id}")
+    r.raise_for_status()
+    return r.json()
 
 
-def get_mechanism_of_action(chembl_id: str):
+def search_assays_by_target(target_chembl_id: str, max_results: int = 50) -> List[dict]:
     """
-    Retrieve mechanism of action for a drug.
+    Retrieve assays related to a specific target.
     """
-    url = f"{CHEMBL_BASE_URL}/mechanism.json?molecule_chembl_id={chembl_id}"
-    response = requests.get(url)
-    response.raise_for_status()
-    return [
-        {
-            "mechanism_of_action": m.get("mechanism_of_action"),
-            "target_name": m.get("target_name"),
-            "target_type": m.get("target_type"),
-        }
-        for m in response.json().get("mechanisms", [])
-    ]
+    params = {"target_chembl_id": target_chembl_id, "limit": max_results}
+    return _fetch_paginated("assay", params)
 
 
-def get_bioactivity_data(chembl_id: str, limit: int = 10):
+def get_drugs_in_clinical_phase(phase: int, max_results: int = 20) -> List[dict]:
     """
-    Retrieve bioactivity data for a given drug.
+    Retrieve drugs by clinical trial phase.
     """
-    url = f"{CHEMBL_BASE_URL}/activity.json?molecule_chembl_id={chembl_id}&limit={limit}"
-    response = requests.get(url)
-    response.raise_for_status()
-    return [
-        {
-            "assay_description": a.get("assay_description"),
-            "standard_type": a.get("standard_type"),
-            "standard_value": a.get("standard_value"),
-            "units": a.get("units"),
-            "activity_comment": a.get("activity_comment"),
-        }
-        for a in response.json().get("activities", [])
-    ]
+    params = {"max_phase": phase, "limit": max_results}
+    return _fetch_paginated("molecule", params)
 
 
-def search_by_target(target_name: str, limit: int = 5):
+def get_molecule_image_url(chembl_id: str) -> str:
     """
-    Search drugs by target name.
+    Generate a URL to the 2D structure image of a molecule.
     """
-    url = f"{CHEMBL_BASE_URL}/target/search.json?q={target_name}&limit={limit}"
-    response = requests.get(url)
-    response.raise_for_status()
-    return [
-        {
-            "target_chembl_id": t.get("target_chembl_id"),
-            "pref_name": t.get("pref_name"),
-            "target_type": t.get("target_type"),
-            "organism": t.get("organism"),
-        }
-        for t in response.json().get("targets", [])
-    ]
+    return f"https://www.ebi.ac.uk/chembl/api/data/image/{chembl_id}.svg"