fhir-patient-api/
├── app/
│   ├── main.py                   # FastAPI应用
│   ├── config.py                 # 配置
│   ├── fhir/
│   │   ├── __init__.py
│   │   ├── core.py               # FHIR核心元素
│   │   ├── patient.py            # Patient资源模型
│   │   └── validators.py         # FHIR验证器
│   ├── api/
│   │   └── patients.py           # 患者API路由
│   ├── services/
│   │   ├── patient_service.py    # 业务逻辑
│   │   └── audit_service.py      # 审计日志
│   └── database.py               # 数据库连接
├── tests/
│   ├── test_patients.py
│   └── test_fhir_validations.py
├── requirements.txt
└── README.md

from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List, Any
from datetime import date, datetime
from enum import Enum

class CodeableConcept(BaseModel):
    """FHIR CodeableConcept - 编码概念"""
    coding: Optional[List["Coding"]] = None
    text: Optional[str] = None

class Coding(BaseModel):
    """FHIR Coding - 编码"""
    system: Optional[str] = Field(
        None,
        description="术语系统URI"
    )
    version: Optional[str] = None
    code: Optional[str] = None
    display: Optional[str] = None
    userSelected: Optional[bool] = None

class Identifier(BaseModel):
    """FHIR Identifier - 标识符"""
    use: Optional[str] = Field(
        None,
        pattern="^(usual|official|temp|secondary|old)$"
    )
    type: Optional[CodeableConcept] = None
    system: str = Field(..., description="标识符系统")
    value: str = Field(..., description="标识符值")
    period: Optional["Period"] = None
    assigner: Optional["Reference"] = None

class HumanName(BaseModel):
    """FHIR HumanName - 人名"""
    use: Optional[str] = Field(
        None,
        pattern="^(usual|official|temp|nickname|anonymous|old|maiden)$"
    )
    text: Optional[str] = None
    family: Optional[str] = Field(None, description="姓")
    given: Optional[List[str]] = Field(None, description="名")
    prefix: Optional[List[str]] = None
    suffix: Optional[List[str]] = None
    period: Optional["Period"] = None

class ContactPoint(BaseModel):
    """FHIR ContactPoint - 联系方式"""
    system: str = Field(
        ...,
        pattern="^(phone|fax|email|pager|url|sms|other)$"
    )
    value: str = Field(..., description="联系方式值")
    use: Optional[str] = Field(
        None,
        pattern="^(home|work|temp|old|mobile)$"
    )
    rank: Optional[int] = Field(None, ge=1, le=1)
    period: Optional["Period"] = None

class Address(BaseModel):
    """FHIR Address - 地址"""
    use: Optional[str] = Field(
        None,
        pattern="^(home|work|temp|old|billing)$"
    )
    type: Optional[str] = Field(
        None,
        pattern="^(postal|physical|both)$"
    )
    text: Optional[str] = None
    line: Optional[List[str]] = None
    city: Optional[str] = None
    district: Optional[str] = None
    state: Optional[str] = None
    postalCode: Optional[str] = None
    country: Optional[str] = None
    period: Optional["Period"] = None

class Period(BaseModel):
    """FHIR Period - 时间段"""
    start: Optional[datetime] = None
    end: Optional[datetime] = None

class Reference(BaseModel):
    """FHIR Reference - 引用"""
    reference: Optional[str] = Field(
        None,
        description="引用引用,如Patient/123"
    )
    type: Optional[str] = Field(None, description="资源类型")
    identifier: Optional[Identifier] = None
    display: Optional[str] = None

class Attachment(BaseModel):
    """FHIR Attachment - 附件"""
    contentType: Optional[str] = None
    language: Optional[str] = None
    data: Optional[str] = Field(None, description="Base64编码数据")
    url: Optional[str] = None
    size: Optional[int] = Field(None, ge=0)
    title: Optional[str] = None
    creation: Optional[datetime] = None

class Meta(BaseModel):
    """FHIR Meta - 元数据"""
    versionId: Optional[str] = Field(None, alias="versionId")
    lastUpdated: Optional[datetime] = Field(None, alias="lastUpdated")
    source: Optional[str] = None
    profile: Optional[List[str]] = None
    security: Optional[List["Coding"]] = None
    tag: Optional[List["Coding"]] = None

class Narrative(BaseModel):
    """FHIR Narrative - 叙述性文本"""
    status: str = Field(..., pattern="^(generated|extensions|additional|empty)$")
    div: str = Field(..., description="XHTML内容")

class Resource(BaseModel):
    """FHIR Resource基础类"""
    resourceType: str = Field(..., description="资源类型")
    id: Optional[str] = None
    meta: Optional[Meta] = None
    implicitRules: Optional[str] = Field(None, alias="implicitRules")
    language: Optional[str] = None

# 更新前向引用
HumanName.model_rebuild()
ContactPoint.model_rebuild()
Address.model_rebuild()
Identifier.model_rebuild()
Reference.model_rebuild()
Period.model_rebuild()
Coding.model_rebuild()
CodeableConcept.model_rebuild()

from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import date, datetime
from enum import Enum

from app.fhir.core import (
    Identifier, HumanName, ContactPoint,
    Address, Reference, Attachment, Meta, Resource
)

class AdministrativeGender(str, Enum):
    """FHIR性别代码"""
    MALE = "male"
    FEMALE = "female"
    OTHER = "other"
    UNKNOWN = "unknown"

class PatientLink(BaseModel):
    """FHIR Patient.link - 患者链接"""
    other: Reference = Field(..., description="关联的患者引用")
    type: str = Field(
        ...,
        pattern="^(replaced-by|replaces|refer|seealso)$"
    )

class PatientCommunication(BaseModel):
    """FHIR Patient.communication - 沟通语言"""
    language: CodeableConcept = Field(..., description="语言")
    preferred: Optional[bool] = Field(None, description="是否首选")

class Contact(BaseModel):
    """FHIR Patient.contact - 紧急联系人"""
    relationship: Optional[List[CodeableConcept]] = None
    name: Optional[HumanName] = None
    telecom: Optional[List[ContactPoint]] = None
    address: Optional[Address] = None
    gender: Optional[AdministrativeGender] = None
    organization: Optional[Reference] = None
    period: Optional[Period] = None

class Patient(Resource):
    """FHIR Patient资源 - 完整模型"""
    resourceType: str = Field(default="Patient", const=True)

    # 核心元素
    identifier: List[Identifier] = Field(
        ...,
        description="患者标识符(至少一个)"
    )
    active: Optional[bool] = Field(None, description="是否活跃")

    # 基本信息
    name: List[HumanName] = Field(
        ...,
        description="患者姓名(至少一个)"
    )
    telecom: Optional[List[ContactPoint]] = Field(
        None,
        description="联系方式"
    )
    gender: Optional[AdministrativeGender] = Field(
        None,
        description="性别"
    )
    birthDate: Optional[date] = Field(None, description="出生日期")
    deceasedBoolean: Optional[bool] = Field(
        None,
        alias="deceasedBoolean",
        description="是否已死亡(布尔)"
    )
    deceasedDateTime: Optional[datetime] = Field(
        None,
        alias="deceasedDateTime",
        description="死亡时间"
    )

    # 地址和照片
    address: Optional[List[Address]] = Field(None, description="地址")
    photo: Optional[List[Attachment]] = Field(None, description="照片")

    # 管理信息
    maritalStatus: Optional[CodeableConcept] = Field(None, description="婚姻状况")
    multipleBirthBoolean: Optional[bool] = Field(None, description="是否多胎")
    multipleBirthInteger: Optional[int] = Field(None, ge=1, description="出生顺序")
    communication: Optional[List[PatientCommunication]] = None

    # 关联
    generalPractitioner: Optional[List[Reference]] = Field(
        None,
        description="全科医生"
    )
    managingOrganization: Optional[Reference] = Field(
        None,
        description="管理机构"
    )
    link: Optional[List[PatientLink]] = Field(None, description="关联患者")

    # 扩展支持
    extension: Optional[List[dict]] = Field(None, description="扩展数据")

    class Config:
        json_schema_extra = {
            "example": {
                "resourceType": "Patient",
                "id": "example",
                "identifier": [{
                    "use": "usual",
                    "type": {
                        "coding": [{
                            "system": "http://terminology.hl7.org/CodeSystem/v2-0203",
                            "code": "MR",
                            "display": "Medical record number"
                        }]
                    },
                    "system": "http://hospital.example.org/mrn",
                    "value": "12345"
                }],
                "active": True,
                "name": [{
                    "use": "official",
                    "family": "Zhang",
                    "given": ["San"]
                }],
                "gender": "male",
                "birthDate": "1990-01-15"
            }
        }

class PatientCreate(BaseModel):
    """创建患者请求(不含id和meta)"""
    identifier: List[Identifier]
    active: Optional[bool] = True
    name: List[HumanName]
    telecom: Optional[List[ContactPoint]] = None
    gender: Optional[AdministrativeGender] = None
    birthDate: Optional[date] = None
    address: Optional[List[Address]] = None
    extension: Optional[List[dict]] = None

class PatientUpdate(BaseModel):
    """更新患者请求(所有字段可选)"""
    identifier: Optional[List[Identifier]] = None
    active: Optional[bool] = None
    name: Optional[List[HumanName]] = None
    telecom: Optional[List[ContactPoint]] = None
    gender: Optional[AdministrativeGender] = None
    birthDate: Optional[date] = None
    address: Optional[List[Address]] = None
    extension: Optional[List[dict]] = None

from typing import List, Optional
from datetime import date, datetime
from pydantic import validator

from app.fhir.patient import Patient, PatientCreate

class FHIRValidator:
    """FHIR资源验证器"""

    @staticmethod
    def validate_patient(patient: PatientCreate) -> List[str]:
        """验证Patient资源"""
        errors = []

        # 1. 必须有至少一个标识符
        if not patient.identifier:
            errors.append("Patient必须包含至少一个identifier")

        # 2. 必须有至少一个姓名
        if not patient.name:
            errors.append("Patient必须包含至少一个name")

        # 3. 检查标识符系统
        for idx, identifier in enumerate(patient.identifier):
            if not identifier.system:
                errors.append(f"identifier[{idx}]: system是必填项")
            if not identifier.value:
                errors.append(f"identifier[{idx}]: value是必填项")

        # 4. 检查姓名
        for idx, name in enumerate(patient.name):
            if not name.family and not name.given:
                errors.append(f"name[{idx}]: family或given至少需要一个")

        # 5. 检查出生日期
        if patient.birthDate:
            if patient.birthDate > date.today():
                errors.append("birthDate不能是未来日期")
            if patient.birthDate.year ＜ 1900:
                errors.append("birthDate年份不能早于1900")

        # 6. 检查死亡字段互斥
        deceased_fields = [
            hasattr(patient, 'deceasedBoolean') and patient.deceasedBoolean is not None,
            hasattr(patient, 'deceasedDateTime') and patient.deceasedDateTime is not None
        ]
        if sum(deceased_fields) > 1:
            errors.append("deceasedBoolean和deceasedDateTime互斥,只能设置一个")

        return errors

    @staticmethod
    def validate_identifier_system(system: str) -> bool:
        """验证标识符系统格式"""
        # FHIR建议使用URI格式
        if system and not (system.startswith('http://') or
                          system.startswith('https://') or
                          system.startswith('urn:oid:') or
                          system.startswith('urn:uuid:')):
            return False
        return True

    @staticmethod
    def validate_phone_number(value: str) -> bool:
        """验证电话号码"""
        # 简化验证,实际应该根据国家代码验证
        if not value:
            return False
        digits = ''.join(filter(str.isdigit, value))
        return 10 <= len(digits) <= 15

    @staticmethod
    def validate_email(value: str) -> bool:
        """验证邮箱地址"""
        import re
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, value))

from fastapi import APIRouter, HTTPException, Response, Query, Depends
from typing import List, Optional
from datetime import datetime

from app.fhir.patient import Patient, PatientCreate, PatientUpdate
from app.fhir.validators import FHIRValidator
from app.services.patient_service import PatientService
from app.services.audit_service import AuditService, AuditEvent

router = APIRouter(prefix="/fhir/R4/Patient", tags=["FHIR-Patient"])

# 依赖注入
async def get_patient_service() -> PatientService:
    return PatientService()

async def get_audit_service() -> AuditService:
    return AuditService()

@router.post(
    "",
    response_model=Patient,
    status_code=201,
    summary="创建患者资源",
    description="根据FHIR R4规范创建新的Patient资源"
)
async def create_patient(
    patient: PatientCreate,
    patient_service: PatientService = Depends(get_patient_service),
    audit_service: AuditService = Depends(get_audit_service)
):
    """
    创建患者

    - **identifier**: 必须,至少一个患者标识符
    - **name**: 必须,至少一个姓名
    - **gender**: 可选,性别(male/female/other/unknown)
    - **birthDate**: 可选,出生日期(YYYY-MM-DD)

    返回创建的完整Patient资源,包含生成的resourceType和id
    """
    # FHIR验证
    errors = FHIRValidator.validate_patient(patient)
    if errors:
        raise HTTPException(
            status_code=422,
            detail={
                "resourceType": "OperationOutcome",
                "issue": [{
                    "severity": "error",
                    "code": "invalid",
                    "diagnostics": error
                } for error in errors]
            }
        )

    # 创建患者
    created = await patient_service.create(patient)

    # 审计日志
    await audit_service.log_event(AuditEvent(
        event_type="Patient.create",
        resource_id=created.id,
        resource_type="Patient",
        timestamp=datetime.utcnow(),
        details={"operation": "create"}
    ))

    return created

@router.get(
    "/{id}",
    response_model=Patient,
    summary="读取患者资源",
    description="根据ID读取Patient资源"
)
async def read_patient(
    id: str,
    patient_service: PatientService = Depends(get_patient_service)
):
    """读取患者"""
    patient = await patient_service.read(id)
    if not patient:
        raise HTTPException(
            status_code=404,
            detail={
                "resourceType": "OperationOutcome",
                "issue": [{
                    "severity": "error",
                    "code": "not-found",
                    "diagnostics": f"Patient/{id}不存在"
                }]
            }
        )
    return patient

@router.put(
    "/{id}",
    response_model=Patient,
    summary="更新患者资源",
    description="完整更新Patient资源"
)
async def update_patient(
    id: str,
    patient: PatientCreate,
    patient_service: PatientService = Depends(get_patient_service),
    audit_service: AuditService = Depends(get_audit_service)
):
    """更新患者"""
    # 验证患者存在
    existing = await patient_service.read(id)
    if not existing:
        raise HTTPException(status_code=404, detail="患者不存在")

    # 验证数据
    errors = FHIRValidator.validate_patient(patient)
    if errors:
        raise HTTPException(status_code=422, detail=errors)

    # 更新
    updated = await patient_service.update(id, patient)

    # 审计
    await audit_service.log_event(AuditEvent(
        event_type="Patient.update",
        resource_id=id,
        resource_type="Patient",
        timestamp=datetime.utcnow(),
        details={"operation": "update"}
    ))

    return updated

@router.patch(
    "/{id}",
    response_model=Patient,
    summary="修补患者资源",
    description="部分更新Patient资源"
)
async def patch_patient(
    id: str,
    patient: PatientUpdate,
    patient_service: PatientService = Depends(get_patient_service)
):
    """修补患者"""
    existing = await patient_service.read(id)
    if not existing:
        raise HTTPException(status_code=404, detail="患者不存在")

    updated = await patient_service.patch(id, patient)
    return updated

@router.delete(
    "/{id}",
    status_code=204,
    summary="删除患者资源",
    description="删除Patient资源(谨慎使用)"
)
async def delete_patient(
    id: str,
    patient_service: PatientService = Depends(get_patient_service),
    audit_service: AuditService = Depends(get_audit_service)
):
    """删除患者"""
    success = await patient_service.delete(id)
    if not success:
        raise HTTPException(status_code=404, detail="患者不存在")

    # 审计
    await audit_service.log_event(AuditEvent(
        event_type="Patient.delete",
        resource_id=id,
        resource_type="Patient",
        timestamp=datetime.utcnow(),
        details={"operation": "delete"}
    ))

    return Response(status_code=204)

@router.get(
    "",
    response_model=List[Patient],
    summary="搜索患者",
    description="根据参数搜索Patient资源"
)
async def search_patients(
    identifier: Optional[str] = Query(None, description="标识符值"),
    family: Optional[str] = Query(None, description="姓氏"),
    given: Optional[str] = Query(None, description="名字"),
    birthdate: Optional[str] = Query(None, description="出生日期(eq/YYYY-MM-DD)"),
    gender: Optional[str] = Query(None, description="性别"),
    _count: int = Query(50, ge=1, le=100, description="返回数量"),
    patient_service: PatientService = Depends(get_patient_service)
):
    """
    搜索患者

    支持的搜索参数:
    - **identifier**: 患者标识符
    - **family**: 姓氏
    - **given**: 名字
    - **birthdate**: 出生日期
    - **gender**: 性别
    - **_count**: 返回数量限制

    返回匹配的Patient资源列表
    """
    results = await patient_service.search(
        identifier=identifier,
        family=family,
        given=given,
        birthdate=birthdate,
        gender=gender,
        limit=_count
    )

    return results

@router.get(
    "/{id}/$everything",
    response_model=dict,
    summary="获取患者所有相关资源",
    description="返回患者及其所有关联资源"
)
async def patient_everything(
    id: str,
    _count: int = Query(50, ge=1, le=100),
    patient_service: PatientService = Depends(get_patient_service)
):
    """获取患者所有数据"""
    patient = await patient_service.read(id)
    if not patient:
        raise HTTPException(status_code=404, detail="患者不存在")

    return await patient_service.get_everything(id, limit=_count)

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.api.patients import router as patient_router
from app.fhir.patient import Patient
from datetime import datetime

app = FastAPI(
    title="FHIR Patient API",
    description="符合FHIR R4标准的患者资源API",
    version="1.0.0",
    terms_of_service="https://example.org/terms",
    contact={
        "name": "API Support",
        "email": "support@example.org"
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html"
    }
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(patient_router)

@app.get("/fhir/R4/metadata", tags=["FHIR-Metadata"])
async def capability_statement():
    """FHIR CapabilityStatement - 服务能力声明"""
    return {
        "resourceType": "CapabilityStatement",
        "status": "active",
        "date": datetime.now().isoformat(),
        "publisher": "WellAlly",
        "kind": "instance",
        "implementation": {
            "description": "FHIR Patient API Server"
        },
        "fhirVersion": "4.0.1",
        "format": ["application/fhir+json", "application/json+fhir"],
        "rest": [{
            "mode": "server",
            "resource": [{
                "type": "Patient",
                "interaction": [
                    {"code": "read"},
                    {"code": "create"},
                    {"code": "update"},
                    {"code": "delete"},
                    {"code": "search-type"}
                ],
                "searchParam": [
                    {
                        "name": "identifier",
                        "type": "token",
                        "definition": "http://hl7.org/fhir/SearchParameter/Patient-identifier"
                    },
                    {
                        "name": "family",
                        "type": "string",
                        "definition": "http://hl7.org/fhir/SearchParameter/Patient-family"
                    },
                    {
                        "name": "given",
                        "type": "string",
                        "definition": "http://hl7.org/fhir/SearchParameter/Patient-given"
                    },
                    {
                        "name": "birthdate",
                        "type": "date",
                        "definition": "http://hl7.org/fhir/SearchParameter/Patient-birthdate"
                    },
                    {
                        "name": "gender",
                        "type": "token",
                        "definition": "http://hl7.org/fhir/SearchParameter/Patient-gender"
                    }
                ]
            }]
        }]
    }

@app.get("/health")
async def health_check():
    """健康检查"""
    return {
        "status": "healthy",
        "fhirVersion": "4.0.1",
        "timestamp": datetime.now().isoformat()
    }

# FHIR异常处理
@app.exception_handler(ValueError)
async def fhir_validation_error(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=422,
        content={
            "resourceType": "OperationOutcome",
            "issue": [{
                "severity": "error",
                "code": "invalid",
                "diagnostics": str(exc)
            }]
        }
    )

POST /fhir/R4/Patient
Content-Type: application/fhir+json

{
  "resourceType": "Patient",
  "identifier": [{
    "system": "http://hospital.example.org/mrn",
    "value": "12345"
  }],
  "name": [{
    "use": "official",
    "family": "张",
    "given": ["三"]
  }],
  "gender": "male",
  "birthDate": "1990-01-15"
}

GET /fhir/R4/Patient?family=张&gender=male&_count=10

GET /fhir/R4/Patient/{id}

GET /fhir/R4/metadata