Update api_clients/rxnorm_client.py

api_clients/rxnorm_client.py

@@ -0,0 +1,124 @@
+# api_clients/rxnorm_client.py
+"""
+Client for the NLM RxNorm API.
+This module is essential for the Drug Interaction & Safety Analyzer. It provides
+functionality to standardize drug names to RxCUIs and then checks for
+potential interactions between a list of those drugs.
+"""
+import asyncio
+import aiohttp
+from .config import RXNORM_BASE_URL, REQUEST_HEADERS
+
+async def get_rxcui(session: aiohttp.ClientSession, drug_name: str) -> str | None:
+    """
+    Converts a drug name string into its corresponding RxNorm Concept Unique Identifier (RxCUI).
+
+    An RxCUI is a standardized code necessary for accurately checking interactions.
+
+    Args:
+        session (aiohttp.ClientSession): The active HTTP session.
+        drug_name (str): The name of the drug (e.g., "Lipitor", "atorvastatin").
+
+    Returns:
+        str | None: The RxCUI as a string if found, otherwise None.
+    """
+    if not drug_name:
+        return None
+        
+    url = f"{RXNORM_BASE_URL}/rxcui.json"
+    params = {'name': drug_name, 'search': 1} # search=1 for exact match first
+
+    try:
+        async with session.get(url, params=params, headers=REQUEST_HEADERS, timeout=10) as resp:
+            resp.raise_for_status()
+            data = await resp.json()
+            # The RxCUI is nested within the 'idGroup'
+            id_group = data.get('idGroup', {})
+            if id_group and 'rxnormId' in id_group:
+                return id_group['rxnormId'][0] # Return the first and most likely CUI
+            return None
+    except aiohttp.ClientError as e:
+        print(f"Error fetching RxCUI for '{drug_name}': {e}")
+        return None
+
+async def get_interactions(session: aiohttp.ClientSession, list_of_rxcuis: list[str]) -> list[dict]:
+    """
+    Checks for interactions among a list of RxCUIs.
+
+    Args:
+        session (aiohttp.ClientSession): The active HTTP session.
+        list_of_rxcuis (list[str]): A list of standardized drug RxCUI strings.
+
+    Returns:
+        list[dict]: A list of interaction pairs, each with a description and severity.
+                    Returns an empty list if no interactions are found or an error occurs.
+    """
+    if not list_of_rxcuis or len(list_of_rxcuis) < 2:
+        return [] # Interactions require at least two drugs
+
+    # The API expects a space-separated string of RxCUIs
+    rxcuis_str = " ".join(list_of_rxcuis)
+    url = f"{RXNORM_BASE_URL}/interaction/list.json"
+    params = {'rxcuis': rxcuis_str}
+
+    try:
+        async with session.get(url, params=params, headers=REQUEST_HEADERS, timeout=15) as resp:
+            resp.raise_for_status()
+            data = await resp.json()
+            
+            interaction_groups = data.get('fullInteractionTypeGroup', [])
+            parsed_interactions = []
+
+            # The API response is deeply nested, so we parse it carefully
+            for group in interaction_groups:
+                for interaction_type in group.get('fullInteractionType', []):
+                    for pair in interaction_type.get('interactionPair', []):
+                        # Extract the critical information into a clean format
+                        description = pair.get('description', 'No description available.')
+                        severity = pair.get('severity', 'Severity unknown')
+                        
+                        # Identify the two drugs involved in this specific interaction
+                        drug1_name = pair['interactionConcept'][0]['minConceptItem']['name']
+                        drug2_name = pair['interactionConcept'][1]['minConceptItem']['name']
+                        
+                        parsed_interactions.append({
+                            "pair": f"{drug1_name} / {drug2_name}",
+                            "severity": severity,
+                            "description": description
+                        })
+            return parsed_interactions
+            
+    except aiohttp.ClientError as e:
+        print(f"Error fetching interactions for RxCUIs '{rxcuis_str}': {e}")
+        return []
+
+async def run_interaction_check(drug_names: list[str]) -> list[dict]:
+    """
+    High-level orchestrator for a complete drug interaction check.
+
+    This function handles the full workflow:
+    1. Takes a list of human-readable drug names.
+    2. Concurrently converts them all to RxCUIs.
+    3. Feeds the valid RxCUIs into the interaction checker.
+
+    Args:
+        drug_names (list[str]): A list of drug names from user input.
+
+    Returns:
+        list[dict]: A list of found interactions, ready for display or AI synthesis.
+    """
+    async with aiohttp.ClientSession() as session:
+        # Step 1: Concurrently get RxCUIs for all provided drug names
+        rxcui_tasks = [get_rxcui(session, name) for name in drug_names]
+        resolved_rxcuis = await asyncio.gather(*rxcui_tasks)
+        
+        # Step 2: Filter out any drugs that were not found (returned None)
+        valid_rxcuis = [rxcui for rxcui in resolved_rxcuis if rxcui]
+        
+        if len(valid_rxcuis) < 2:
+            print("Fewer than two valid drugs found, cannot check for interactions.")
+            return []
+            
+        # Step 3: Check for interactions using the list of valid RxCUIs
+        interactions = await get_interactions(session, valid_rxcuis)
+        return interactions