FastAPI PostgreSQL: Build a Production-Ready API

FastAPI and PostgreSQL make a strong foundation for production APIs: FastAPI handles HTTP requests, validation, and OpenAPI documentation, while PostgreSQL provides transactions, constraints, indexing, and dependable relational storage. The combination suits SaaS products, internal platforms, data services, and AI applications that need a clear path from prototype to scale.

This guide uses modern Python patterns and focuses on decisions that matter in production. The examples use SQLAlchemy 2.x with the asynchronous asyncpg driver, Pydantic settings, and Alembic migrations. If you are designing a larger service estate, pair this foundation with guidance on building scalable microservices with FastAPI and keep service boundaries driven by domain ownership rather than framework fashion.

Why use FastAPI with PostgreSQL?

FastAPI is built around Python type hints. Request and response models make API contracts explicit, validate input before it reaches business logic, and generate OpenAPI documentation automatically. Its ASGI architecture supports asynchronous endpoints and integrates well with background workers, streaming responses, and modern observability tools.

PostgreSQL complements those strengths with:

Transactional integrity: Use atomic commits and rollbacks for operations that change multiple records.
Powerful querying: Common table expressions, window functions, full-text search, JSONB, and extensions cover more than basic CRUD.
Strong constraints: Primary keys, foreign keys, unique constraints, checks, and non-null rules protect data at the database boundary.
Operational maturity: Backups, replication, managed services, monitoring, and connection pooling are widely available.

For AI builders in India, PostgreSQL is often the system of record behind model registries, evaluation results, user entitlements, prompt metadata, and workflow state. Store large objects such as model weights or documents in object storage; keep searchable metadata, permissions, and lifecycle state in PostgreSQL.

Recommended project setup

Create an isolated environment and install the core dependencies:

python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn sqlalchemy asyncpg alembic pydantic-settings

Use psycopg instead of asyncpg if your team standardises on Psycopg 3. The important choice is consistency: select one driver and configure SQLAlchemy for that driver rather than mixing synchronous and asynchronous sessions casually.

Keep credentials outside source control. A local .env file might contain:

DATABASE_URL=postgresql+asyncpg://app_user:password@localhost:5432/app_db

In staging and production, supply the value through the deployment platform’s secret manager. Do not commit passwords, tokens, or full connection strings to Git.

Configure the async database layer

Create a settings object and an async engine:

# db.py
from collections.abc import AsyncIterator
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy.ext.asyncio import (
    AsyncSession, async_sessionmaker, create_async_engine
)

class Settings(BaseSettings):
    database_url: str
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

settings = Settings()
engine = create_async_engine(
    settings.database_url,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=10,
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncIterator[AsyncSession]:
    async with session_factory() as session:
        yield session

pool_pre_ping helps detect stale connections, but it is not a substitute for sensible pool sizing. A database has finite connection capacity. Calculate the total possible connections across all application replicas, workers, and administrative tools before increasing pool_size.

For simple low-traffic services, synchronous SQLAlchemy is also valid. The failure mode to avoid is running blocking database calls inside an async endpoint, which can occupy the event loop and reduce throughput under load.

Define models and API schemas separately

SQLAlchemy models describe persistence; Pydantic schemas describe the public API. Keeping them separate prevents internal columns from being exposed accidentally and lets the API evolve without rewriting the database schema.

# models.py
from sqlalchemy import String, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120), index=True)
    description: Mapped[str | None] = mapped_column(Text())

# schemas.py
from pydantic import BaseModel, ConfigDict, Field

class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=120)
    description: str | None = None

class ItemRead(ItemCreate):
    id: int
    model_config = ConfigDict(from_attributes=True)

Use database constraints as the final line of defence. Application validation improves error messages, but it cannot prevent race conditions between two requests; a unique constraint and correct transaction handling can.

Build CRUD endpoints with transactions

# main.py
from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from db import get_db
from models import Item
from schemas import ItemCreate, ItemRead

app = FastAPI(title="Items API")

@app.post("/items", response_model=ItemRead, status_code=status.HTTP_201_CREATED)
async def create_item(payload: ItemCreate, db: AsyncSession = Depends(get_db)):
    item = Item(**payload.model_dump())
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

@app.get("/items/{item_id}", response_model=ItemRead)
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Item).where(Item.id == item_id))
    item = result.scalar_one_or_none()
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

For updates involving several tables, use async with db.begin(): so the entire operation commits or rolls back as one unit. Catch expected integrity errors and return a useful 409 Conflict; do not expose raw database exception text to clients.

Migrations, indexes, and query design

Use Alembic for every schema change. Base.metadata.create_all() is acceptable for a throwaway local experiment, not for a shared environment. A migration provides a reviewable, repeatable history:

alembic init migrations
alembic revision --autogenerate -m "create items"
alembic upgrade head

Review autogenerated migrations manually. Add indexes based on real access patterns, such as a composite index for frequent tenant-and-created-time queries. Avoid indexing every column: indexes speed reads but consume storage and make writes more expensive.

For list endpoints, implement pagination from the beginning. Offset pagination is easy but becomes slower and less stable on changing datasets. Cursor or keyset pagination using an indexed, deterministic sort key is usually better for large tables. Select only the columns required by the response and inspect slow queries with EXPLAIN (ANALYZE, BUFFERS).

PostgreSQL JSONB is useful for flexible metadata, but do not use it to avoid modelling relationships or fields that require frequent filtering. If your application needs vector similarity search, evaluate a PostgreSQL extension such as pgvector alongside a clear retrieval and indexing strategy; database storage does not remove the need to measure recall, latency, and cost.

Security and reliability checklist

A production FastAPI PostgreSQL service should include:

Authentication and authorisation: Verify identity and enforce object-level permissions, not only endpoint-level access.
Parameterised queries: Use SQLAlchemy expressions or bound parameters; never concatenate user input into SQL.
CORS discipline: Allow only known origins and avoid wildcard credentials.
Request limits: Set body-size, timeout, pagination, and upload limits.
Safe errors: Log detailed exceptions internally while returning stable public error responses.
TLS and least privilege: Encrypt database connections and give the application role only the permissions it needs.
Backups and restore tests: A backup that has never been restored is an assumption, not a recovery plan.
Observability: Track request latency, error rates, database pool saturation, slow queries, and migration status.

FastAPI’s dependency system is useful for authentication, tenant resolution, and transaction-scoped sessions. For real-time workloads, also plan for connection lifecycle and backpressure; the real-time software guidance topic provides useful context for choosing between polling, server-sent events, and WebSockets.

Testing and deployment

Test business logic without requiring a live production database, then run integration tests against PostgreSQL itself. SQLite can be convenient for unit tests, but it does not faithfully reproduce PostgreSQL types, locking, query planning, or constraint behaviour.

A practical test stack includes:

FastAPI’s TestClient or HTTPX for endpoint tests.
A disposable PostgreSQL service in CI, commonly via Docker or a managed test database.
A clean migration run before tests.
Fixtures that roll back or truncate data safely between cases.
Tests for authentication, duplicate records, missing resources, pagination, and transaction rollback.

Run the service locally with:

uvicorn main:app --reload

Production deployment should run behind a process manager or container platform with health checks, structured logs, graceful shutdown, and a separate migration step. Do not run --reload in production. Set worker counts based on measured CPU, memory, database capacity, and workload; more workers are not automatically faster.

FAQ

Is FastAPI PostgreSQL integration difficult?

No. The basic connection is straightforward; production quality comes from session management, migrations, constraints, testing, and operational discipline.

Should I use async SQLAlchemy?

Use it when your service is predominantly asynchronous and performs concurrent I/O. A synchronous stack is simpler for some applications. Do not mix blocking database work into async endpoints without isolating it.

What is the best PostgreSQL driver for FastAPI?

asyncpg is a common choice for SQLAlchemy’s async engine. Psycopg 3 is another strong option, especially for teams already using Psycopg across synchronous and asynchronous services.

Can PostgreSQL support AI applications?

Yes. It works well for structured application state, metadata, permissions, evaluations, and moderate-scale retrieval workloads. Use object storage for large files and benchmark extensions such as pgvector against your actual data and latency requirements.

FastAPI with PostgreSQL is not merely a quick CRUD combination. With explicit schemas, controlled transactions, reviewed migrations, measured queries, and tested recovery procedures, it becomes a dependable base for Indian startups and engineering teams building APIs that must operate beyond the prototype stage.

Apply for AI Grants India