Spaces:
Running
Running
| # api_clients/clinicaltrials_client.py | |
| """ | |
| Client for the ClinicalTrials.gov API (Version 2). | |
| This client is optimized to find active, recruiting studies for a given condition, | |
| which is the most relevant information for patients and caregivers. | |
| """ | |
| import aiohttp | |
| from urllib.parse import quote_plus | |
| from .config import CLINICALTRIALS_BASE_URL, REQUEST_HEADERS | |
| async def find_trials(session: aiohttp.ClientSession, condition_query: str, max_results: int = 5) -> list[dict]: | |
| """ | |
| Searches ClinicalTrials.gov for active, recruiting studies related to a condition. | |
| Args: | |
| session (aiohttp.ClientSession): The active HTTP session. | |
| condition_query (str): The medical condition to search for (e.g., "Type 2 Diabetes"). | |
| max_results (int): The maximum number of trial results to return. | |
| Returns: | |
| list[dict]: A list of dictionaries, where each dictionary represents a clinical trial. | |
| Returns an empty list if no trials are found or an error occurs. | |
| """ | |
| if not condition_query: | |
| return [] | |
| # The API is powerful. We will specifically query for trials that are currently recruiting, | |
| # as this is the most actionable information for a user. | |
| # The `quote_plus` function ensures the query is safely encoded for the URL. | |
| formatted_query = quote_plus(condition_query) | |
| # Construct the query parameters | |
| params = { | |
| 'query.term': formatted_query, | |
| 'filter.overallStatus': 'RECRUITING', # The "wow" factor: finding ACTIONABLE trials | |
| 'pageSize': max_results, | |
| 'format': 'json' | |
| } | |
| url = f"{CLINICALTRIALS_BASE_URL}/studies" | |
| try: | |
| async with session.get(url, params=params, headers=REQUEST_HEADERS, timeout=15) as resp: | |
| # Raise an exception for bad status codes (4xx or 5xx) | |
| resp.raise_for_status() | |
| data = await resp.json() | |
| studies_list = data.get('studies', []) | |
| if not studies_list: | |
| return [] | |
| # Parse the complex response into a simple, clean list of dictionaries | |
| parsed_trials = [] | |
| for study in studies_list: | |
| # Use nested .get() calls to safely access potentially missing keys | |
| protocol = study.get('protocolSection', {}) | |
| id_module = protocol.get('identificationModule', {}) | |
| status_module = protocol.get('statusModule', {}) | |
| nct_id = id_module.get('nctId', '') | |
| if not nct_id: | |
| continue # Skip if the trial has no ID | |
| parsed_trials.append({ | |
| "title": id_module.get('briefTitle', 'Title Not Available'), | |
| "nct_id": nct_id, | |
| "status": status_module.get('overallStatus', 'Status Unknown'), | |
| "url": f"https://clinicaltrials.gov/study/{nct_id}", | |
| "condition": ", ".join(protocol.get('conditionsModule', {}).get('conditions', ['N/A'])), | |
| "intervention": ", ".join(i.get('name', 'N/A') for i in protocol.get('armsAndInterventionsModule', {}).get('interventions', [])) | |
| }) | |
| return parsed_trials | |
| except aiohttp.ClientError as e: | |
| print(f"An error occurred while fetching data from ClinicalTrials.gov: {e}") | |
| return [] | |
| except Exception as e: | |
| print(f"A general error occurred in the clinicaltrials_client: {e}") | |
| return [] |