fastapi-healthcare 0.0.1

Healthcare compliance extensions for FastAPI — audit logging, PHI protection, FHIR models, and more

fastapi-healthcare

Healthcare compliance extensions for FastAPI -- audit logging, PHI protection, structured errors, health checks, clinical AI serving, and more.

Why fastapi-healthcare?

Building healthcare APIs with FastAPI means implementing HIPAA audit trails, PHI redaction in logs and error responses, safe ML inference pipelines, structured error formats, and Kubernetes health probes -- all from scratch, across every service. This is security-critical boilerplate that's easy to get wrong: a single unredacted SSN in a log line or stack trace is a compliance violation.

fastapi-healthcare gives you all of this as drop-in FastAPI middleware and Pydantic models. Each module is independent -- use one or all five, with zero overhead for modules you don't import.

What You Get

Module	What it does	Standard
audit	Automatic HIPAA audit trail for every request -- who did what, when, to which resource	HIPAA §164.312(b)
phi	Declarative PHI field annotations with role-based redaction, AES-256-GCM encryption, and safe logging	HIPAA §164.514
errors	Machine-readable error responses with audit correlation	RFC 9457
health	/health, /ready, /live endpoints with parallel check execution	Kubernetes probes
ai	Unified ML inference client with automatic PHI scrubbing before payloads leave your service	Vertex AI, SageMaker, Azure ML

Getting Started

A complete patient API with audit logging, PHI protection, structured errors, and health checks:

# main.py
from fastapi import Depends, FastAPI

from fastapi_healthcare.audit import (
    AuditContext,
    AuditMiddleware,
    FileAuditStorage,
    get_audit_context,
)
from fastapi_healthcare.errors import install_problem_handlers, ProblemDetails
from fastapi_healthcare.health import HealthCheckRouter, DiskSpaceCheck, MemoryCheck
from fastapi_healthcare.phi import (
    PHIClassification,
    PHISafeLogger,
    PHISafeModel,
    RedactionStrategy,
    phi_field,
)

app = FastAPI(title="Patient Service")

# 1. Audit logging -- every request is recorded with actor, resource, and outcome
storage = FileAuditStorage("./audit_logs", retention_days=90)
app.add_middleware(AuditMiddleware, storage=storage)

# 2. Structured errors -- all exceptions become RFC 9457 JSON responses
install_problem_handlers(app)

# 3. Health checks -- Kubernetes probes out of the box
app.include_router(HealthCheckRouter(
    checks=[DiskSpaceCheck(path="/"), MemoryCheck()],
    version="1.0.0",
))

# 4. PHI-safe logging -- SSNs, emails, and dates are auto-redacted
logger = PHISafeLogger("patients")


# 5. PHI-annotated model -- each field declares its sensitivity and redaction rule
class Patient(PHISafeModel):
    id: str
    name: str = phi_field(PHIClassification.DIRECT_IDENTIFIER, redaction=RedactionStrategy.PARTIAL,
                          partial_visible=1, partial_position="start")
    ssn: str = phi_field(PHIClassification.DIRECT_IDENTIFIER, redaction=RedactionStrategy.MASK)
    diagnosis: str = phi_field(PHIClassification.SENSITIVE, redaction=RedactionStrategy.HASH)
    blood_type: str  # not PHI -- visible to everyone


PATIENTS: dict[str, Patient] = {}


@app.post("/patients", status_code=201)
async def create_patient(
    patient: Patient,
    audit: AuditContext = Depends(get_audit_context),
):
    audit.set_resource("patient", patient.id)
    PATIENTS[patient.id] = patient
    logger.info("Created patient %s", patient.id)
    return patient.safe_dict(role="nurse")


@app.get("/patients/{patient_id}")
async def get_patient(patient_id: str):
    if patient_id not in PATIENTS:
        raise ProblemDetails(
            type_uri="https://api.example.com/errors/patient-not-found",
            title="Patient Not Found",
            status=404,
            detail=f"No patient with ID '{patient_id}'.",
        )
    return PATIENTS[patient_id].safe_dict(role="nurse")

Run it:

pip install fastapi-healthcare uvicorn
uvicorn main:app --reload

Try it:

# Create a patient
curl -X POST http://localhost:8000/patients \
  -H "Content-Type: application/json" \
  -d '{"id":"pt-1","name":"Jane Doe","ssn":"123-45-6789","diagnosis":"Hypertension","blood_type":"O+"}'

# Response (nurse role): name is partially redacted, SSN is masked, diagnosis visible
# {"id": "pt-1", "name": "J*******", "ssn": "*******6789", "diagnosis": "Hypertension", "blood_type": "O+"}

# Check health
curl http://localhost:8000/health

# Request a missing patient -- returns RFC 9457 JSON, not a generic 500
curl http://localhost:8000/patients/pt-999
# {"type": "https://api.example.com/errors/patient-not-found", "title": "Patient Not Found", "status": 404, ...}

What happened under the hood:

• The POST /patients request was recorded in ./audit_logs/audit-YYYY-MM-DD.jsonl with the actor, resource type, resource ID, status code, and duration
• The logger call redacted any PHI that might appear in the patient ID or log message
• The GET /patients/pt-999 error was returned as application/problem+json with an audit_id linking it to the audit trail
• repr(patient) and str(patient) never expose PHI -- safe for debuggers and print statements

Installation

pip install fastapi-healthcare

With all optional dependencies:

pip install fastapi-healthcare[all]

Modules

Audit Logging (fastapi_healthcare.audit)

HIPAA-compliant audit trail middleware that automatically captures who did what, when, and where for every request.

from fastapi import Depends
from fastapi_healthcare.audit import (
    AuditMiddleware,
    FileAuditStorage,
    AuditContext,
    get_audit_context,
)

# Basic setup with file-based storage
storage = FileAuditStorage(
    base_dir="./audit_logs",
    max_file_size_mb=100,
    retention_days=90,
)
app.add_middleware(AuditMiddleware, storage=storage)

# Extract actor identity from JWT tokens
async def extract_actor(request):
    token = request.headers.get("Authorization", "")
    # ... decode JWT ...
    return ("user-123", "physician")

app.add_middleware(
    AuditMiddleware,
    storage=storage,
    actor_extractor=extract_actor,
    include_paths=["/api/*"],
    exclude_paths=["/api/health*"],
)

# Enrich audit events from within endpoints
@app.post("/api/v1/patients")
async def create_patient(
    patient: PatientCreate,
    audit: AuditContext = Depends(get_audit_context),
):
    audit.set_actor("dr-smith", "physician")
    audit.set_resource("patient", patient.id)
    audit.add_metadata("department", "cardiology")
    # ... create patient ...

Audit events are stored as JSON Lines with date-partitioned files:

audit_logs/
  audit-2026-06-06.jsonl
  audit-2026-06-07.jsonl

Each event captures: audit_id, timestamp, action, resource_type, resource_id, actor_id, actor_role, source_ip, user_agent, request_method, request_path, status_code, duration_ms, outcome, and metadata.

Cloud Logging backend -- for GKE deployments, swap FileAuditStorage for CloudLoggingAuditStorage to send audit events to Google Cloud Logging as structured entries:

pip install fastapi-healthcare[cloud-logging]

from fastapi_healthcare.audit import AuditMiddleware, CloudLoggingAuditStorage

storage = CloudLoggingAuditStorage(
    project="my-gcp-project",
    log_name="hipaa-audit",
    labels={"service": "patient-api"},
)
app.add_middleware(AuditMiddleware, storage=storage)

See the GKE deployment guide for the full walkthrough.

Customizing URL parsing and search detection:

import re

app.add_middleware(
    AuditMiddleware,
    storage=storage,
    # Custom regex for non-REST URL structures (must have 2 capture groups)
    resource_pattern=re.compile(r"/services/([a-z]+)/items/([a-zA-Z0-9_-]+)"),
    # Custom query params that trigger SEARCH action (empty set disables)
    search_indicators={"lookup", "find", "term"},
)

PHI Protection (fastapi_healthcare.phi)

Declarative PHI field annotations with role-based redaction, encryption, and safe logging.

from fastapi_healthcare.phi import (
    PHISafeModel,
    phi_field,
    PHIClassification,
    RedactionStrategy,
    PHISafeLogger,
    PHISafeErrorHandler,
)

# Define models with PHI annotations
class Patient(PHISafeModel):
    id: str
    name: str = phi_field(
        PHIClassification.DIRECT_IDENTIFIER,
        redaction=RedactionStrategy.PARTIAL,
        partial_visible=1,
        partial_position="start",
        description="Patient full name",
    )
    ssn: str = phi_field(
        PHIClassification.DIRECT_IDENTIFIER,
        redaction=RedactionStrategy.MASK,
        description="Social Security Number",
    )
    date_of_birth: str = phi_field(
        PHIClassification.QUASI_IDENTIFIER,
        redaction=RedactionStrategy.MASK,
        description="Date of birth",
    )
    diagnosis: str = phi_field(
        PHIClassification.SENSITIVE,
        redaction=RedactionStrategy.HASH,
        description="Primary diagnosis",
    )

patient = Patient(
    id="pt-001",
    name="Jane Doe",
    ssn="123-45-6789",
    date_of_birth="1990-05-14",
    diagnosis="Hypertension",
)

# Role-based redaction
nurse_view = patient.redacted(role="nurse")
# nurse_view.name -> "J*******"  (nurses can't see DIRECT_IDENTIFIERs)
# nurse_view.ssn -> "*******6789"
# nurse_view.diagnosis -> "Hypertension"  (nurses CAN see SENSITIVE)

public_view = patient.redacted(role="public")
# public_view.name -> "J*******"
# public_view.ssn -> "*******6789"
# public_view.diagnosis -> "sha256:a1b2c3..."

# Safe JSON serialization
safe_json = patient.safe_json(role="billing")

# PHI-safe repr (never leaks PHI in logs/debugger)
print(patient)
# Patient(id='pt-001', name='[REDACTED]', ssn='[REDACTED]', ...)

PHI-safe logging automatically scrubs sensitive patterns from log messages:

logger = PHISafeLogger("myapp.patients")

logger.info("Processing patient SSN: 123-45-6789")
# Output: {"message": "Processing patient SSN: ***-**-6789", ...}

logger.info("Contact email: [e-mail], phone: [phone]")
# Output: {"message": "Contact email: j***@***.*** phone: (***) ***-4567", ...}

PHI-safe error handling prevents PHI leaks in exception responses:

handler = PHISafeErrorHandler(include_traceback=False)
handler.install(app)
# Exceptions like ValueError("Invalid DOB '1990-05-14' for SSN 123-45-6789")
# become: "Invalid DOB '****-**-**' for SSN ***-**-6789"

Role permission matrix (built-in, customizable):

Role	DIRECT_IDENTIFIER	QUASI_IDENTIFIER	SENSITIVE	DE_IDENTIFIED
administrator	x	x	x	x
physician	x	x	x	x
nurse		x	x	x
billing	x			x
researcher				x
public				x

Custom roles -- extend the default matrix without replacing it:

from fastapi_healthcare.phi import merge_role_permissions, PHIClassification

permissions = merge_role_permissions({
    "pharmacist": {PHIClassification.DIRECT_IDENTIFIER, PHIClassification.DE_IDENTIFIED},
    "nurse": {  # override the built-in nurse role
        PHIClassification.DIRECT_IDENTIFIER,
        PHIClassification.QUASI_IDENTIFIER,
        PHIClassification.SENSITIVE,
        PHIClassification.DE_IDENTIFIED,
    },
})

patient.redacted(role="pharmacist", role_permissions=permissions)

Custom PHI patterns -- replace the built-in US-centric patterns for non-US deployments:

import re
from fastapi_healthcare.phi import PHISafeLogger

nhs_patterns = [
    ("NHS", "NHS-***-***-****", re.compile(r"\b\d{3}\s?\d{3}\s?\d{4}\b")),
]

logger = PHISafeLogger(
    "myapp",
    extra_patterns=nhs_patterns,
    replace_default_patterns=True,  # use ONLY custom patterns
)

Structured Errors (fastapi_healthcare.errors)

RFC 9457 Problem Details for HTTP APIs -- machine-readable error responses for healthcare API interoperability.

from fastapi_healthcare.errors import (
    ProblemDetails,
    install_problem_handlers,
)

# Install handlers (replaces FastAPI defaults)
install_problem_handlers(app)

# Raise structured errors
raise ProblemDetails(
    type_uri="https://api.example.com/errors/patient-not-found",
    title="Patient Not Found",
    status=404,
    detail=f"No patient with ID {patient_id} exists in the system.",
    instance=f"/api/v1/patients/{patient_id}",
    extensions={"suggested_action": "Verify the patient ID and try again."},
)

# Convenience method
ProblemDetails.raise_for(
    status=409,
    detail="Active prescription conflicts with new order.",
    extensions={
        "conflicting_rx": "RX-2024-001",
        "interaction_severity": "high",
    },
)

All error responses use application/problem+json content type:

{
  "type": "https://api.example.com/errors/patient-not-found",
  "title": "Patient Not Found",
  "status": 404,
  "detail": "No patient with ID pt-999 exists in the system.",
  "instance": "/api/v1/patients/pt-999",
  "audit_id": "a1b2c3d4e5f6",
  "suggested_action": "Verify the patient ID and try again."
}

Validation errors, HTTP exceptions, and unhandled exceptions are all converted to RFC 9457 format automatically.

Custom error type URIs -- point the predefined error types at your organization's domain:

from fastapi_healthcare.errors import HealthcareErrorTypes

HealthcareErrorTypes.configure("https://api.myorg.com/errors")
# All 13 error type URIs now use your base URI:
# HealthcareErrorTypes.PATIENT_NOT_FOUND -> "https://api.myorg.com/errors/patient-not-found"

Health Checks (fastapi_healthcare.health)

Kubernetes-ready health check endpoints with parallel execution and structured responses.

from fastapi_healthcare.health import (
    HealthCheckRouter,
    BaseHealthCheck,
    HealthCheckResult,
    HealthStatus,
    DatabaseCheck,
    CacheCheck,
    ExternalServiceCheck,
    DiskSpaceCheck,
    MemoryCheck,
)

# Built-in checks
health_router = HealthCheckRouter(
    checks=[
        DatabaseCheck(dsn="postgresql://localhost/mydb", tags=["readiness"]),
        CacheCheck(url="redis://localhost:6379", tags=["readiness"]),
        ExternalServiceCheck(
            name="auth-service",
            url="https://auth.internal/health",
            tags=["readiness"],
        ),
        DiskSpaceCheck(path="/data", min_free_gb=1.0, warn_free_gb=5.0),
        MemoryCheck(max_rss_mb=512.0, warn_rss_mb=384.0),
    ],
    version="1.0.0",
)
app.include_router(health_router)
# Provides: GET /health, GET /ready, GET /live

# Custom checks
class FHIRServerCheck(BaseHealthCheck):
    def __init__(self, fhir_url: str):
        super().__init__(name="fhir-server", tags=["readiness"])
        self.fhir_url = fhir_url

    async def check(self) -> HealthCheckResult:
        # ... check FHIR server connectivity ...
        return HealthCheckResult(
            name=self.name,
            status=HealthStatus.HEALTHY,
            latency_ms=15.2,
            message="FHIR R4 server responding",
            details={"version": "4.0.1"},
            tags=self.tags,
        )

health_router.add_check(FHIRServerCheck("https://fhir.example.com/r4"))

Endpoints:

Endpoint	Purpose	Checks
GET /health	Full health status	All registered checks (parallel)
GET /ready	Readiness probe	Only checks tagged readiness
GET /live	Liveness probe	None (returns 200 if process is running)

Response format:

{
  "status": "healthy",
  "checks": [
    {
      "name": "database",
      "status": "healthy",
      "latency_ms": 12.5,
      "message": "PostgreSQL connection OK",
      "details": {"pool_size": 10, "active_connections": 3}
    },
    {
      "name": "cache",
      "status": "healthy",
      "latency_ms": 1.2
    }
  ],
  "version": "1.0.0",
  "uptime_seconds": 86400.5,
  "timestamp": "2026-06-06T12:00:00Z"
}

Clinical AI Serving (fastapi_healthcare.ai)

Unified ML inference client for healthcare models deployed on Vertex AI (GCP), SageMaker (AWS), and Azure ML. PHI is automatically scrubbed from outbound payloads before they cross the service boundary, and every inference call is audit-logged.

Install the provider you need:

pip install fastapi-healthcare[vertex]     # Google Cloud Vertex AI
pip install fastapi-healthcare[sagemaker]  # Amazon SageMaker
pip install fastapi-healthcare[azure-ml]   # Azure Machine Learning

Vertex AI example:

from fastapi_healthcare.ai import (
    VertexAIClient,
    MLClientConfig,
    MLProvider,
    PredictionRequest,
)
from fastapi_healthcare.audit import FileAuditStorage

config = MLClientConfig(
    provider=MLProvider.VERTEX_AI,
    endpoint_name="clinical-risk-model",
    timeout=10.0,
    phi_scrub=True,  # scrub PHI before sending to Vertex AI
)

storage = FileAuditStorage("./audit_logs")
client = VertexAIClient(
    config=config,
    project="my-gcp-project",
    location="us-central1",
    audit_storage=storage,
)

request = PredictionRequest(
    endpoint_name="clinical-risk-model",
    instances=[{"age": 65, "bp_systolic": 140, "patient_ssn": "123-45-6789"}],
    metadata={"actor_id": "dr-smith", "department": "cardiology"},
)

response = await client.predict(request)
# response.predictions -> [{"risk_score": 0.87, "risk_level": "high"}]
# response.latency_ms -> 45.2
# response.status -> PredictionStatus.SUCCESS
# The SSN was scrubbed before it left your service.
# An audit event was logged with actor_id, endpoint, and latency.

SageMaker and Azure ML follow the same pattern:

from fastapi_healthcare.ai import SageMakerClient, AzureMLClient

# Amazon SageMaker
sagemaker_client = SageMakerClient(
    config=MLClientConfig(provider=MLProvider.SAGEMAKER, endpoint_name="my-endpoint"),
    region_name="us-east-1",
    audit_storage=storage,
)

# Azure Machine Learning
azure_client = AzureMLClient(
    config=MLClientConfig(provider=MLProvider.AZURE_ML, endpoint_name="my-endpoint"),
    subscription_id="00000000-0000-0000-0000-000000000000",
    resource_group="my-rg",
    workspace_name="my-workspace",
    audit_storage=storage,
)

Custom ML providers -- subclass ClinicalMLClient to integrate any provider while keeping PHI scrubbing, audit logging, and timeout handling:

from fastapi_healthcare.ai import ClinicalMLClient, PredictionRequest, PredictionResponse

class MyCustomClient(ClinicalMLClient):
    async def _predict(self, request: PredictionRequest) -> PredictionResponse:
        # Your provider-specific inference call here.
        # PHI is already scrubbed from request.instances if phi_scrub=True.
        ...

    async def close(self) -> None:
        ...

Request body middleware -- scrub PHI from JSON request bodies at the middleware level before they reach downstream AI endpoints:

from fastapi_healthcare.ai import ClinicalMLMiddleware

app.add_middleware(
    ClinicalMLMiddleware,
    scrub_paths=["/api/*/predict*", "/api/*/infer*"],
)
# POST /api/v1/predict with {"patient_ssn": "123-45-6789", "age": 65}
# becomes {"patient_ssn": "***-**-6789", "age": 65} before your endpoint sees it.

Error handling -- timeouts and provider errors raise RFC 9457 ProblemDetails:

from fastapi_healthcare.errors import ProblemDetails

try:
    response = await client.predict(request)
except ProblemDetails as exc:
    # exc.status == 504 for timeouts
    # exc.status == 503 for provider errors
    # exc.detail describes the failure
    ...

Key Design Decisions

• Zero runtime overhead when unused -- each module is independent; import only what you need
• Pydantic v2 native -- all models use Pydantic v2 with model_dump(), model_construct(), and model_validate_json()
• Async-first -- audit storage, health checks, and middleware are all async/await
• Type-safe -- strict mypy, PEP 561 py.typed marker, all public APIs fully annotated
• Standards-based -- RFC 9457 errors, HIPAA audit fields, Kubernetes probe conventions

Contributing

Development Setup

git clone https://github.com/abelpech/fastapi-healthcare.git
cd fastapi-healthcare
pip install -e ".[dev,all]"

Running Tests

# Full test suite with coverage
bash scripts/test.sh

# Quick test run
pytest tests/ -v

Linting and Formatting

# Check lint and formatting
bash scripts/lint.sh

# Auto-fix
bash scripts/format.sh

Building the Package

The package uses pdm-backend as its build system (configured in pyproject.toml). The version is read dynamically from fastapi_healthcare/__init__.py.

# Build sdist (.tar.gz) and wheel (.whl) into dist/
uv build

# Or with standard pip tooling
python -m build

This produces:

dist/
  fastapi_healthcare-0.1.0.tar.gz       # source distribution
  fastapi_healthcare-0.1.0-py3-none-any.whl  # wheel (pure Python)

To publish to PyPI:

# Test PyPI first
uv publish --publish-url https://test.pypi.org/legacy/

# Production PyPI
uv publish

Quality Gates

All contributions must pass:

Gate	Command	Threshold
Tests	pytest tests/ --strict-config --strict-markers	98 tests, 0 failures
Lint	ruff check fastapi_healthcare tests	0 errors
Format	ruff format --check fastapi_healthcare tests	0 reformats
Types	mypy fastapi_healthcare	0 errors (strict mode)
Build	uv build	sdist + wheel

Project Conventions

This project follows FastAPI's conventions:

• Build system: pdm-backend with dynamic version from __init__.py
• Style: ruff for linting/formatting (E, W, F, I, B, C4, UP rule sets)
• Types: mypy strict mode with pydantic plugin
• Imports: X | None unions (PEP 604), collections.abc for Callable/Awaitable/Sequence, X as X re-exports in __init__.py
• Tests: pytest + pytest-asyncio, filterwarnings = ["error"]
• Pre-commit: ruff + pre-commit-hooks

Areas for Contribution

High-impact additions:

• FHIR R4 Models -- Pydantic models for core FHIR resources (Patient, Observation, Encounter, Bundle) with built-in validation
• Cloud audit storage backends -- Google Cloud Logging, AWS CloudWatch, Azure Monitor adapters for AuditStorage
• Database health checks -- Production-ready connection pool monitoring for SQLAlchemy, asyncpg, Motor
• OAuth2/SMART on FHIR scopes -- Scope-based access control integrated with PHI classification levels
• Rate limiting with clinical context -- Differentiated rate limits for emergency vs. routine access patterns
• Model registry integration -- Version tracking and model metadata for ClinicalMLClient predictions

Incremental improvements:

• Expand test coverage for edge cases (malformed PHI patterns, concurrent encryption, storage rotation)
• Add benchmarks for PHI redaction at scale (10K+ records)
• OpenTelemetry integration for audit event export
• Structured logging integration (structlog adapter for PHISafeLogger)
• Additional ML provider clients (Hugging Face Inference Endpoints, custom gRPC serving)

License

MIT -- see LICENSE for details.