Query gnomAD GraphQL API for population allele frequencies.
API: POST https://gnomad.broadinstitute.org/api
Note: Requires variant_id (chr-pos-ref-alt format) from FAVOR Step 2.
Return:
Allele frequencies across populations (global, AFR, AMR, ASJ, EAS, FIN, NFE, SAS, etc.)
import requests
import json
from datetime import datetime
rs_id = "rs7412"
results = {"query_rsid": rs_id, "timestamp": datetime.now().isoformat()}
def safe_request(name, func, fallback=None):
"""统一的容错请求包装器。任一数据库超时/报错不会中断整个流程。"""
try:
return func()
except requests.exceptions.Timeout:
print(f"[{name}] ⚠ 连接超时，跳过")
results.setdefault("errors", {})[name] = "timeout"
return fallback
except Exception as e:
print(f"[{name}] ⚠ 请求失败: {e}，跳过")
results.setdefault("errors", {})[name] = str(e)
return fallback
rsid_num = rs_id.replace("rs", "")
dbsnp_url = f"https://api.ncbi.nlm.nih.gov/variation/v0/refsnp/{rsid_num}"
dbsnp = safe_request("dbSNP",
lambda: requests.get(dbsnp_url, timeout=30).json(), fallback={})
results["dbsnp"] = dbsnp
print(f"[dbSNP] {rs_id} 查询{'成功' if dbsnp else '失败'}")
snapshot = dbsnp.get("primary_snapshot_data", {})
clinvar_records = []
for ann in snapshot.get("allele_annotations", []):
for clin in ann.get("clinical", []):
clinvar_records.append({
"accession": clin.get("accession_version", ""),
"significances": clin.get("clinical_significances", []),
"diseases": clin.get("disease_names", []),
"review_status": clin.get("review_status", "")
})
results["clinvar"] = clinvar_records
print(f"[ClinVar] RCV 记录数: {len(clinvar_records)}")
favor_url = f"https://api.genohub.org/v1/rsids/{rs_id}"
favor = safe_request("FAVOR",
lambda: requests.get(favor_url, timeout=30).json(), fallback={})
results["favor"] = favor
print(f"[FAVOR] 功能注释{'获取成功' if favor else '获取失败'}")
variant_id = None
favor_results = favor if isinstance(favor, list) else favor.get("results", [favor]) if favor else []
if favor_results and isinstance(favor_results, list):
first = favor_results[0] if favor_results else {}
chrom = str(first.get("chromosome", ""))
pos = str(first.get("position", ""))
ref = first.get("ref", "")
alt = first.get("alt", "")
if chrom and pos and ref and alt:
variant_id = f"{chrom}-{pos}-{ref}-{alt}"
if variant_id:
gnomad_query = """
query($variantId: String!) {
variant(variantId: $variantId, dataset: gnomad_r4) {
variant_id
genome {
ac
an
af
populations { id ac an af }
}
}
}
"""
gnomad_data = safe_request("gnomAD",
lambda: requests.post(
"https://gnomad.broadinstitute.org/api",
json={"query": gnomad_query, "variables": {"variantId": variant_id}},
timeout=30
).json(), fallback={})
results["gnomad"] = (gnomad_data or {}).get("data", {}).get("variant", {})
genome = results["gnomad"].get("genome", {}) if results["gnomad"] else {}
print(f"[gnomAD] AF={genome.get('af', 'N/A')}")
else:
results["gnomad"] = {"error": "无法从 FAVOR 提取 variant_id"}
print("[gnomAD] 跳过: 无 variant_id")
gwas_url = f"https://www.ebi.ac.uk/gwas/rest/api/associations/search/findByRsId?rsId={rs_id}"
gwas = safe_request("GWAS Catalog",
lambda: requests.get(gwas_url, headers={"Accept": "application/json"}, timeout=30).json(),
fallback={})
associations = (gwas or {}).get("_embedded", {}).get("associations", [])
results["gwas_catalog"] = {
"association_count": len(associations),
"associations": associations
}
print(f"[GWAS Catalog] 关联数={len(associations)}")
pgx_url = f"https://api.clinpgx.org/v1/data/clinicalAnnotation?location.fingerprint={rs_id}&view=base"
pgx_resp = safe_request("PharmGKB",
lambda: requests.get(pgx_url, timeout=30).json(), fallback=[])
pgx_annotations = pgx_resp if isinstance(pgx_resp, list) else (pgx_resp or {}).get("data", [])
results["pharmgkb"] = pgx_annotations
print(f"[PharmGKB] 药物基因组注释数: {len(pgx_annotations)}")
clingen_url = f"https://reg.genome.network/alleles?dbSNP.rs={rs_id}"
clingen_resp = None
for attempt in range(2):
clingen_resp = safe_request("ClinGen",
lambda: requests.get(clingen_url,
headers={"Accept": "application/json"}, timeout=60).json(),
fallback=None)
if clingen_resp is not None:
break
print(f"[ClinGen] 第 {attempt+1} 次尝试失败，重试中...")
if clingen_resp is not None:
if not isinstance(clingen_resp, list):
clingen_resp = [clingen_resp]
clingen_alleles = []
for allele in clingen_resp:
titles = allele.get("communityStandardTitle", [])
if titles and any("=" in t for t in titles):
continue
clingen_alleles.append({
"ca_id": allele.get("@id", "").split("/")[-1],
"title": titles,
"externalRecords": allele.get("externalRecords", {})
})
results["clingen"] = clingen_alleles
print(f"[ClinGen] 等位基因数: {len(clingen_alleles)}")
else:
results["clingen"] = {"error": "ClinGen API 连接超时，可稍后单独使用 variant-cross-database-ids skill 重试"}
print("[ClinGen] ⚠ 所有尝试均超时，已跳过")
output_file = f"{rs_id}_annotation.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump(results, f, indent=2, ensure_ascii=False)
errors = results.get("errors", {})
if errors:
print(f"\n⚠ 以下数据库查询失败: {list(errors.keys())}，其余数据库结果正常")
print(f"✓ 结果已保存: {output_file}")
