1. 引言
aidbox-python-sdk 是 Aidbox 官方提供的 Python 软件开发工具包,用于简化 Python 应用与 Aidbox 医疗数据平台之间的交互。Aidbox 是一个基于 FHIR(Fast Healthcare Interoperability Resources)标准的医疗数据平台,而 aidbox-python-sdk 封装了与 Aidbox REST API 通信的底层细节,让开发者可以更高效地在 Python 项目中完成 FHIR 资源的创建、查询、更新和删除等操作。
本文将从功能特性、安装方式、核心语法与参数、9 个实际应用案例以及常见错误与使用注意事项五个方面,系统性地介绍 aidbox-python-sdk 的使用方法。
2. 功能概述
aidbox-python-sdk 的核心功能可以概括为以下几个方面:
- FHIR 资源操作:支持对 Patient、Observation、Encounter 等标准 FHIR 资源进行增删改查操作。
- REST API 封装:将 Aidbox REST API 的 HTTP 请求封装为简洁的 Python 方法调用。
- 身份认证:内置基于 Basic Auth 和 Token 的认证机制,支持与 Aidbox 的访问控制体系对接。
- 查询构建器:提供链式查询语法,支持按资源类型、字段条件、分页和排序进行灵活查询。
- 批量操作:支持通过 Bundle 事务批量提交多个资源操作,提升数据同步效率。
- 异步支持:提供异步客户端,方便在 asyncio 应用中集成。
- 错误处理:封装了 Aidbox API 返回的错误码和异常类型,便于开发者捕获和处理。
3. 安装与环境准备
3.1 环境要求
aidbox-python-sdk 要求 Python 3.8 及以上版本,并依赖 requests 或 httpx 库进行 HTTP 通信。建议使用虚拟环境进行安装,避免与系统环境产生依赖冲突。
3.2 安装命令
通过 pip 安装最新版本:
pip install aidbox-python-sdk如果需要安装指定版本:
pip install aidbox-python-sdk==0.4.2安装完成后,可以通过以下命令验证是否安装成功:
python -c "import aidbox; print(aidbox.__version__)"3.3 配置连接参数
在使用 SDK 之前,需要准备 Aidbox 服务的地址、客户端 ID 和客户端密钥。通常在 Aidbox 管理控制台中创建客户端后获取这些信息。
4. 核心语法与参数详解
4.1 初始化客户端
使用同步客户端时,通过 AidboxClient 类进行初始化:
from aidbox import AidboxClient client = AidboxClient( base_url="https://your-aidbox-instance.aidbox.app", client_id="your-client-id", client_secret="your-client-secret" )参数说明:
- base_url:Aidbox 服务的根地址,必填。
- client_id:Aidbox 客户端 ID,必填。
- client_secret:客户端密钥,必填。
- timeout:请求超时时间,默认 30 秒,可选。
- verify_ssl:是否校验 SSL 证书,默认 True,可选。
4.2 资源创建
创建 FHIR 资源使用 create 方法,传入资源类型和资源数据字典:
patient_data = { "resourceType": "Patient", "name": [{"family": "张", "given": ["伟"]}], "gender": "male", "birthDate": "1990-01-01" } patient = client.resource("Patient").create(patient_data)4.3 资源查询
查询资源使用 search 方法,支持链式条件过滤:
patients = client.resource("Patient").search( name="张", gender="male", _count=20 ).execute()常用查询参数:
- _count:每页返回的资源数量。
- _sort:排序字段,如 _sort=-birthDate 表示按出生日期降序。
- _page:页码,用于分页查询。
- _include:关联资源包含查询。
4.4 资源更新与删除
# 更新资源 patient["name"] = [{"family": "张", "given": ["伟", "明"]}] updated = client.resource("Patient").update(patient) 删除资源 client.resource("Patient").delete(patient["id"])4.5 批量操作
通过 Bundle 事务批量提交多个操作:
bundle = client.bundle() bundle.add_create("Patient", patient_data) bundle.add_update("Observation", observation_data) result = bundle.execute()5. 实际应用案例
5.1 案例一:患者注册
在门诊系统中注册新患者,并自动生成患者编号:
from aidbox import AidboxClient client = AidboxClient( base_url="https://aidbox.example.com", client_id="clinic-app", client_secret="secret-key" ) def register_patient(family_name, given_name, gender, birth_date): patient_data = { "resourceType": "Patient", "name": [{"family": family_name, "given": [given_name]}], "gender": gender, "birthDate": birth_date, "active": True } patient = client.resource("Patient").create(patient_data) print(f"患者已注册,ID: {patient['id']}") return patient register_patient("李", "娜", "female", "1988-05-12")5.2 案例二:患者信息查询
根据姓名和性别组合条件查询患者列表:
def search_patients(name=None, gender=None, limit=10): query = client.resource("Patient").search(_count=limit) if name: query = query.search(name=name) if gender: query = query.search(gender=gender) result = query.execute() return result.get("entry", []) patients = search_patients(name="王", gender="male", limit=5) for item in patients: p = item["resource"] print(p["id"], p["name"][0]["family"])5.3 案例三:生命体征记录
为患者创建血压和心率观察记录:
def create_vital_signs(patient_id, systolic, diastolic, heart_rate): observation_data = { "resourceType": "Observation", "status": "final", "subject": {"reference": f"Patient/{patient_id}"}, "code": { "coding": [{ "system": "http://loinc.org", "code": "85354-9", "display": "Blood pressure panel" }] }, "component": [ { "code": {"coding": [{"system": "http://loinc.org", "code": "8480-6"}]}, "valueQuantity": {"value": systolic, "unit": "mmHg"} }, { "code": {"coding": [{"system": "http://loinc.org", "code": "8462-4"}]}, "valueQuantity": {"value": diastolic, "unit": "mmHg"} } ] } return client.resource("Observation").create(observation_data) create_vital_signs("patient-123", 120, 80, 72)5.4 案例四:批量导入历史病历
将 CSV 文件中的历史病历批量导入 Aidbox:
import csv def batch_import_patients(csv_path): bundle = client.bundle() with open(csv_path, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: patient_data = { "resourceType": "Patient", "name": [{"family": row["family"], "given": [row["given"]]}], "gender": row["gender"], "birthDate": row["birth_date"] } bundle.add_create("Patient", patient_data) result = bundle.execute() print(f"成功导入 {len(result.get('entry', []))} 条记录") batch_import_patients("patients.csv")5.5 案例五:预约挂号管理
创建预约资源并关联患者和医生:
def create_appointment(patient_id, practitioner_id, start_time, end_time): appointment_data = { "resourceType": "Appointment", "status": "booked", "participant": [ {"actor": {"reference": f"Patient/{patient_id}"}, "status": "accepted"}, {"actor": {"reference": f"Practitioner/{practitioner_id}"}, "status": "accepted"} ], "start": start_time, "end": end_time } return client.resource("Appointment").create(appointment_data) create_appointment( "patient-456", "practitioner-789", "2026-10-06T09:00:00Z", "2026-10-06T09:30:00Z" )5.6 案例六:检验报告上传
上传实验室检验结果并关联到患者:
def upload_lab_result(patient_id, lab_code, value, unit, reference_range): observation_data = { "resourceType": "Observation", "status": "final", "subject": {"reference": f"Patient/{patient_id}"}, "code": { "coding": [{ "system": "http://loinc.org", "code": lab_code }] }, "valueQuantity": {"value": value, "unit": unit}, "referenceRange": [{"low": {"value": reference_range["low"]}, "high": {"value": reference_range["high"]}}] } return client.resource("Observation").create(observation_data) upload_lab_result("patient-456", "718-7", 13.5, "g/dL", {"low": 12.0, "high": 16.0})5.7 案例七:用药处方管理
创建 MedicationRequest 资源记录医生处方:
def create_prescription(patient_id, practitioner_id, medication_code, dosage): prescription_data = { "resourceType": "MedicationRequest", "status": "active", "intent": "order", "subject": {"reference": f"Patient/{patient_id}"}, "requester": {"reference": f"Practitioner/{practitioner_id}"}, "medicationCodeableConcept": { "coding": [{ "system": "http://www.nlm.nih.gov/research/umls/rxnorm", "code": medication_code }] }, "dosageInstruction": [{"text": dosage}] } return client.resource("MedicationRequest").create(prescription_data) create_prescription( "patient-456", "practitioner-789", "312961", "每日一次,每次一片,饭后服用" )5.8 案例八:异步批量数据同步
使用异步客户端并发同步多个患者数据:
import asyncio from aidbox import AsyncAidboxClient async def sync_patients(): client = AsyncAidboxClient( base_url="https://aidbox.example.com", client_id="sync-service", client_secret="sync-secret" ) tasks = [] for i in range(10): patient_data = { "resourceType": "Patient", "name": [{"family": f"测试{i}", "given": ["用户"]}], "gender": "unknown" } tasks.append(client.resource("Patient").create(patient_data)) results = await asyncio.gather(*tasks) print(f"同步完成,共创建 {len(results)} 个患者") await client.close() asyncio.run(sync_patients())5.9 案例九:数据导出与报表生成
查询指定时间段内的所有就诊记录并生成统计报表:
from datetime import datetime def generate_visit_report(start_date, end_date): encounters = client.resource("Encounter").search( _count=100, _sort="period.start" ).execute() total_visits = 0 male_count = 0 female_count = 0 for item in encounters.get("entry", []): encounter = item["resource"] period = encounter.get("period", {}) start = period.get("start", "") if start_date <= start <= end_date: total_visits += 1 subject = encounter.get("subject", {}).get("reference", "") patient = client.resource("Patient").get(subject.split("/")[-1]) if patient.get("gender") == "male": male_count += 1 elif patient.get("gender") == "female": female_count += 1 print(f"总就诊次数: {total_visits}") print(f"男性患者: {male_count}") print(f"女性患者: {female_count}") return {"total": total_visits, "male": male_count, "female": female_count} report = generate_visit_report("2026-09-01", "2026-09-30")6. 常见错误与使用注意事项
6.1 常见错误
| 错误类型 | 错误信息 | 原因与解决方法 |
|---|---|---|
| 认证失败 | 401 Unauthorized | client_id 或 client_secret 配置错误,检查 Aidbox 控制台中的客户端凭据。 |
| 资源不存在 | 404 Not Found | 请求的资源 ID 不存在,或资源类型拼写错误,检查资源 ID 和类型名称。 |
| 参数校验失败 | 422 Unprocessable Entity | FHIR 资源数据不符合规范,如缺少必填字段或字段类型错误,对照 FHIR 规范检查数据。 |
| 请求超时 | TimeoutError | 网络延迟或 Aidbox 服务负载过高,适当增加 timeout 参数或重试。 |
| 批量操作部分失败 | Bundle 返回部分错误 | Bundle 中某个操作失败不影响其他操作,需逐个检查 entry 中的 response 状态。 |
6.2 使用注意事项
- 敏感信息保护:client_secret 等敏感凭据不要硬编码在代码中,建议使用环境变量或密钥管理服务存储。
- 分页处理:查询大量数据时务必使用分页参数,避免一次性拉取过多数据导致内存溢出。
- 资源 ID 管理:创建资源后应妥善保存返回的 id,后续更新和删除操作需要依赖该 ID。
- 时区处理:FHIR 日期时间字段使用 ISO 8601 格式,建议统一使用 UTC 时间存储,展示时再转换为本地时区。
- 错误重试:对于网络抖动或临时服务不可用,建议实现指数退避重试机制。
- 版本兼容:SDK 版本升级可能带来 API 变更,升级前阅读官方更新日志,并在测试环境验证。
《AI提示工程必知必会》为读者提供了丰富的AI提示工程知识与实战技能,主要包括各类提示词的应用,如问答式、指令式、状态类、建议式、安全类和感谢类提示词,以及如何通过实战演练掌握提示词的使用技巧;使用提示词进行文本摘要、改写重述、语法纠错、机器翻译等语言处理任务,以及在数据挖掘、程序开发等领域的应用;AI在绘画创作上的应用,百度文心一言和阿里通义大模型这两大智能平台的特性与功能,以及市场调研中提示词的实战应用。通过阅读《AI提示工程必知必会》,读者可掌握如何有效利用AI提示工程提升工作效率,创新工作流程,并在职场中脱颖而出。