How to Integrate FastAPI for investment banking with PostgreSQL for startups
FastAPI gives you the HTTP layer for banking workflows; PostgreSQL gives you durable state for trades, clients, positions, and audit trails. Put them together and you get an AI agent backend that can accept requests, validate them fast, persist financial records safely, and serve low-latency reads for downstream models and internal tools.
Prerequisites
- •Python 3.10+
- •A running PostgreSQL instance
- •A FastAPI app scaffolded in a virtual environment
- •
pipinstalled packages:- •
fastapi - •
uvicorn - •
sqlalchemy - •
psycopg2-binary - •
pydantic
- •
- •A PostgreSQL database and user with permissions to create tables
- •Basic familiarity with REST endpoints and SQLAlchemy ORM patterns
Integration Steps
- •
Install the dependencies
Keep the stack simple. For startup systems, I prefer SQLAlchemy over raw drivers because you get connection pooling, transaction control, and cleaner model definitions.
pip install fastapi uvicorn sqlalchemy psycopg2-binary pydantic - •
Create the database connection layer
Use a dedicated module for your engine and session factory. This keeps your FastAPI routes clean and makes it easier to swap environments later.
# db.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base DATABASE_URL = "postgresql+psycopg2://bank_user:bank_pass@localhost:5432/investment_bank" engine = create_engine( DATABASE_URL, pool_pre_ping=True, pool_size=10, max_overflow=20, ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() - •
Define a persistent model for banking records
For an AI agent system, you usually need a transaction ledger or an order table. This example stores trade orders with status fields that your agent can update as it moves through review or execution.
# models.py from sqlalchemy import Column, Integer, String, Numeric, DateTime, func from db import Base class TradeOrder(Base): __tablename__ = "trade_orders" id = Column(Integer, primary_key=True, index=True) client_id = Column(String(64), nullable=False, index=True) symbol = Column(String(16), nullable=False) side = Column(String(4), nullable=False) # BUY / SELL quantity = Column(Integer, nullable=False) price = Column(Numeric(18, 4), nullable=False) status = Column(String(32), nullable=False, default="PENDING") created_at = Column(DateTime(timezone=True), server_default=func.now()) - •
Build the FastAPI service with PostgreSQL dependency injection
This is the part that matters in production: each request gets its own DB session, and every write is wrapped in a transaction boundary. If your AI agent decides to place an order or store a recommendation, this is where it lands.
# main.py from fastapi import FastAPI, Depends from pydantic import BaseModel from sqlalchemy.orm import Session from db import SessionLocal, engine from models import TradeOrder, Base Base.metadata.create_all(bind=engine) app = FastAPI(title="Investment Banking Agent API") def get_db(): db = SessionLocal() try: yield db finally: db.close() class TradeOrderCreate(BaseModel): client_id: str symbol: str side: str quantity: int price: float @app.post("/orders") def create_order(payload: TradeOrderCreate, db: Session = Depends(get_db)): order = TradeOrder( client_id=payload.client_id, symbol=payload.symbol.upper(), side=payload.side.upper(), quantity=payload.quantity, price=payload.price, status="PENDING", ) db.add(order) db.commit() db.refresh(order) return {"id": order.id, "status": order.status} @app.get("/orders/{order_id}") def get_order(order_id: int, db: Session = Depends(get_db)): order = db.query(TradeOrder).filter(TradeOrder.id == order_id).first() if not order: return {"error": "order_not_found"} return { "id": order.id, "client_id": order.client_id, "symbol": order.symbol, "side": order.side, "quantity": order.quantity, "price": float(order.price), "status": order.status, } - •
Run migrations or initialize tables before serving traffic
For startup teams moving quickly, people often skip migrations and regret it later. At minimum, create tables on startup in dev; in production use Alembic.
Testing the Integration
Start the app:
uvicorn main:app --reload
Then verify both write and read paths:
import requests
base_url = "http://127.0.0.1:8000"
create_resp = requests.post(
f"{base_url}/orders",
json={
"client_id": "CUST-1001",
"symbol": "AAPL",
"side": "BUY",
"quantity": 50,
"price": 192.55
},
)
print(create_resp.json())
order_id = create_resp.json()["id"]
get_resp = requests.get(f"{base_url}/orders/{order_id}")
print(get_resp.json())
Expected output:
{"id": 1,"status":"PENDING"}
{"id": 1,"client_id":"CUST-1001","symbol":"AAPL","side":"BUY","quantity":50,"price":192.55,"status":"PENDING"}
Real-World Use Cases
- •
AI-assisted trade intake
- •An agent receives deal data from analysts or clients.
- •FastAPI validates the payload.
- •PostgreSQL stores the canonical record for compliance and downstream processing.
- •
Portfolio monitoring APIs
- •Store positions and market snapshots in PostgreSQL.
- •Expose read endpoints through FastAPI for internal dashboards or LLM-powered assistants.
- •
Approval workflows for banking ops
- •Persist approval states like
PENDING,REVIEWED,APPROVED,REJECTED. - •Let an AI agent recommend actions while humans keep final control.
- •Persist approval states like
If you want this production-ready for a startup bank stack, add Alembic migrations, structured logging with request IDs, row-level security in PostgreSQL, and async background jobs for enrichment or risk checks.
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.
Want the complete 8-step roadmap?
Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.
Get the Starter Kit