How to Integrate FastAPI for healthcare with PostgreSQL for production AI

FastAPI for healthcare gives you a clean way to expose clinical workflows as APIs, while PostgreSQL gives you durable, queryable storage for patient records, audit trails, and AI-generated outputs. Put them together and you can build production AI systems that ingest medical data, run inference, and persist results with the controls healthcare teams expect.

Prerequisites

• •Python 3.11+
• •PostgreSQL 14+
• •A running FastAPI application
• •A database user with CREATE TABLE, INSERT, SELECT, and UPDATE permissions
• •Python packages:
  • •fastapi
  • •uvicorn
  • •psycopg2-binary or psycopg
  • •sqlalchemy
  • •pydantic
• •A .env file or secrets manager for:
  • •DATABASE_URL=postgresql://user:password@localhost:5432/healthcare_ai

Integration Steps

1. •

   Create the FastAPI app and database connection

   Start by wiring FastAPI to PostgreSQL through SQLAlchemy. In production, use a connection pool and keep the database URL out of source control.

   from fastapi import FastAPI
   from sqlalchemy import create_engine
   from sqlalchemy.orm import sessionmaker, declarative_base
   
   DATABASE_URL = "postgresql+psycopg2://user:password@localhost:5432/healthcare_ai"
   
   engine = create_engine(
       DATABASE_URL,
       pool_size=10,
       max_overflow=20,
       pool_pre_ping=True,
   )
   
   SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
   Base = declarative_base()
   
   app = FastAPI(title="Healthcare AI API")
   

2. •

   Define a production-friendly patient model

   Keep the schema explicit. For healthcare systems, you want typed fields, timestamps, and room for AI outputs like risk scores or triage labels.

   from sqlalchemy import Column, Integer, String, DateTime, Float, Text
   from sqlalchemy.sql import func
   
   class PatientRecord(Base):
       __tablename__ = "patient_records"
   
       id = Column(Integer, primary_key=True, index=True)
       patient_id = Column(String(64), unique=True, nullable=False, index=True)
       symptoms = Column(Text, nullable=False)
       ai_assessment = Column(Text, nullable=True)
       risk_score = Column(Float, nullable=True)
       created_at = Column(DateTime(timezone=True), server_default=func.now())
       updated_at = Column(DateTime(timezone=True), onupdate=func.now())
   

3. •

   Create tables and add a request/response contract

   Use Pydantic models at the API boundary. That keeps your payloads stable and prevents malformed data from reaching PostgreSQL.

   from pydantic import BaseModel
   
   class PatientCreate(BaseModel):
       patient_id: str
       symptoms: str
   
   class PatientResponse(BaseModel):
       id: int
       patient_id: str
       symptoms: str
       ai_assessment: str | None
       risk_score: float | None
   
       class Config:
           from_attributes = True
   
   @app.on_event("startup")
   def startup():
       Base.metadata.create_all(bind=engine)
   

4. •

   Add an endpoint that writes to PostgreSQL and returns AI output

   This is where the integration becomes useful. Your API receives clinical text, stores it in PostgreSQL, runs your AI logic, then persists the result back to the same row.

   from fastapi import Depends, HTTPException
   from sqlalchemy.orm import Session
   
   def get_db():
       db = SessionLocal()
       try:
           yield db
       finally:
           db.close()
   
   def simple_triage(symptoms: str) -> tuple[str, float]:
       text = symptoms.lower()
       if "chest pain" in text or "shortness of breath" in text:
           return "Urgent review recommended", 0.95
       if "fever" in text or "cough" in text:
           return "Routine clinical review", 0.55
       return "Low acuity", 0.20
   
   @app.post("/patients", response_model=PatientResponse)
   def create_patient(record: PatientCreate, db: Session = Depends(get_db)):
       assessment, score = simple_triage(record.symptoms)
   
       db_record = PatientRecord(
           patient_id=record.patient_id,
           symptoms=record.symptoms,
           ai_assessment=assessment,
           risk_score=score,
       )
   
       db.add(db_record)
       db.commit()
       db.refresh(db_record)
   
       return db_record
   

5. •

   Add a read endpoint for downstream agent workflows

   Production AI systems usually need retrieval as much as write-paths. Agents can query prior assessments before generating recommendations or escalation messages.

    @app.get("/patients/{patient_id}", response_model=PatientResponse)
    def get_patient(patient_id: str, db: Session = Depends(get_db)):
        row = (
            db.query(PatientRecord)
            .filter(PatientRecord.patient_id == patient_id)
            .first()
        )
   
        if not row:
            raise HTTPException(status_code=404, detail="Patient not found")
   
        return row
   

Testing the Integration

Run the app:

uvicorn main:app --reload

Then test it with a real request:

import requests

payload = {
    "patient_id": "P-10001",
    "symptoms": "Patient reports chest pain and shortness of breath"
}

r = requests.post("http://127.0.0.1:8000/patients", json=payload)
print(r.status_code)
print(r.json())

r2 = requests.get("http://127.0.0.1:8000/patients/P-10001")
print(r2.status_code)
print(r2.json())

Expected output:

200
{
  "id": 1,
  "patient_id": "P-10001",
  "symptoms": "Patient reports chest pain and shortness of breath",
  "ai_assessment": "Urgent review recommended",
  "risk_score": 0.95
}
200
{
  "id": 1,
  "patient_id": "P-10001",
  "symptoms": "Patient reports chest pain and shortness of breath",
  "ai_assessment": "Urgent review recommended",
  "risk_score": 0.95
}

Real-World Use Cases

• •

  Clinical triage assistants

  • •FastAPI handles intake forms.
  • •PostgreSQL stores symptom history and model outputs for auditability.
• •

  Prior authorization support

  • •Agents extract medical context from requests.
  • •PostgreSQL stores decision traces and insurer-facing responses.
• •

  Care coordination workflows

  • •One service ingests provider notes through FastAPI.
  • •Another agent queries PostgreSQL to generate follow-up tasks or escalation flags.

If you’re building production AI in healthcare, this pattern is the baseline: API at the edge, Postgres as system of record, deterministic persistence around any model output that affects care or operations.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.