| |
| import httpx, random |
|
|
| V2_URL = "https://clinicaltrials.gov/api/v2/studies" |
| V1_URL = "https://clinicaltrials.gov/api/query/study_fields" |
| UA_LIST = [ |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64)…Chrome/125…", |
| "Mozilla/5.0 (Macintosh)…Safari/605…", |
| "Mozilla/5.0 (X11)…Firefox/126…" |
| ] |
|
|
| async def search_trials(term: str, max_studies: int = 20) -> list[dict]: |
| ua = random.choice(UA_LIST) |
| try: |
| async with httpx.AsyncClient(headers={"User-Agent": ua}, timeout=12) as c: |
| r = await c.get(V2_URL, params={"query": term, "pageSize": max_studies}) |
| if r.status_code == 200: |
| return r.json().get("studies", []) |
| raise httpx.HTTPStatusError("v2 failed", request=r.request, response=r) |
| except Exception: |
| params = { |
| "expr": term, |
| "fields": "NCTId,BriefTitle,Phase,OverallStatus,StartDate", |
| "max_rnk": max_studies, |
| "min_rnk": 1, |
| "fmt": "json" |
| } |
| async with httpx.AsyncClient(headers={"User-Agent": ua}, timeout=12) as c: |
| r = await c.get(V1_URL, params=params) |
| r.raise_for_status() |
| return r.json().get("StudyFieldsResponse", {}).get("StudyFields", []) |
|
|
