api_clients/rxnorm_client.py

# 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