Backend Development2026-08-1615 min read

Share this article:

How to Build a Production REST API with Python and FastAPI

FastAPI has become the standard Python framework for modern backend APIs. Built on top of Starlette and Pydantic, it combines Python type hints with asynchronous I/O to deliver high performance, automatic data validation, and interactive OpenAPI documentation right out of the box.

However, moving from a single-file prototype (app = FastAPI()) to a production-ready web service requires a structured approach. You need a scalable directory architecture, asynchronous database session handling, modular routing with APIRouter, request and response serialization, and production process management.

This guide walks through building a complete, production-grade REST API in Python using FastAPI, Pydantic v2, and async SQLAlchemy 2.0.

Why FastAPI for Modern Python Backends

Traditional Python web frameworks like Flask and Django were designed around synchronous WSGI (Web Server Gateway Interface) standards. While mature and feature-rich, handling concurrent I/O-bound requests (such as database queries, external API calls, and microservice communication) often required multi-threading or multi-processing with noticeable memory overhead.

FastAPI builds on the ASGI (Asynchronous Server Gateway Interface) standard and modern Python type hints:

Native Asynchronous Concurrency: Routes can be defined with async def, allowing the server's event loop to handle other incoming requests while waiting for database queries or network I/O to resolve.
Pydantic v2 Validation: Request payloads, query parameters, and headers are parsed and validated automatically against declared schemas with descriptive error responses returned on invalid input.
Automatic OpenAPI / Swagger Documentation: Interactive API documentation is generated at /docs (Swagger UI) and /redoc (ReDoc) directly from your type annotations and route definitions.
Dependency Injection System: A hierarchical dependency injection system (Depends) manages database connections, authentication, and configuration lifecycles cleanly without global state.

Recommended Project Structure

A clean, modular directory layout separates routing, business logic, data models, schemas, and configuration:

fastapi-production-api/
├── app/
│   ├── __init__.py
│   ├── main.py                  # FastAPI app initialization and middleware
│   ├── core/
│   │   ├── __init__.py
│   │   └── config.py            # Environment settings with pydantic-settings
│   ├── db/
│   │   ├── __init__.py
│   │   ├── session.py           # Async SQLAlchemy engine & session factory
│   │   └── base.py              # DeclarativeBase model registry
│   ├── models/
│   │   ├── __init__.py
│   │   └── product.py           # SQLAlchemy ORM database models
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── product.py           # Pydantic v2 request/response schemas
│   └── api/
│       ├── __init__.py
│       └── v1/
│           ├── __init__.py
│           ├── router.py        # Central API router aggregator
│           └── endpoints/
│               ├── __init__.py
│               └── products.py  # Product CRUD route handlers
├── .env.example
├── pyproject.toml
└── requirements.txt

Step 1: Environment & Dependency Setup

Create and activate a virtual environment, then install the required production packages:

python3 -m venv .venv
source .venv/bin/activate

# Install FastAPI, Uvicorn, Pydantic, and Async Database Drivers
pip install "fastapi>=0.111.0" "uvicorn[standard]>=0.30.0" "pydantic>=2.7.0" "pydantic-settings>=2.2.0" "sqlalchemy>=2.0.30" "asyncpg>=0.29.0" "greenlet>=3.0.0"

Note on greenlet: When using SQLAlchemy's async ORM features with asynchronous drivers (such as asyncpg for PostgreSQL or aiosqlite for SQLite), greenlet is required by SQLAlchemy's async concurrency layer.

Step 2: Typed Configuration with `pydantic-settings`

Use pydantic-settings to load and validate environment variables into a strongly typed configuration object:

# app/core/config.py
from typing import List
from pydantic import AnyHttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    PROJECT_NAME: str = "FastAPI Production API"
    VERSION: str = "1.0.0"
    API_V1_STR: str = "/api/v1"
    
    # Database Configuration
    DATABASE_URL: str = "postgresql+asyncpg://app_user:secure_password@localhost:5432/production_db"
    
    # CORS Origins
    BACKEND_CORS_ORIGINS: List[str] = ["http://localhost:3000", "https://thedailydevs.com"]

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=True
    )

settings = Settings()

Step 3: Async Database Engine and Session Lifecycle

Configure an asynchronous SQLAlchemy engine and create a dependency generator (get_db) to manage database sessions per request:

# app/db/session.py
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.core.config import settings

# Create async engine with connection pool sizing
engine = create_async_engine(
    settings.DATABASE_URL,
    echo=False,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True, # Validates connection health before issuing queries
)

# Async session factory
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False,
)

# Dependency to yield an independent database session per request
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Define the declarative base class for your database models:

# app/db/base.py
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

Step 4: SQLAlchemy Database Models

Define database tables using SQLAlchemy 2.0's type-annotated Mapped and mapped_column syntax:

# app/models/product.py
from datetime import datetime, timezone
from typing import Optional
from sqlalchemy import String, Numeric, DateTime, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base

class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True, index=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(150), nullable=False, index=True)
    slug: Mapped[str] = mapped_column(String(160), unique=True, index=True, nullable=False)
    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    price: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False)
    in_stock: Mapped[bool] = mapped_column(default=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
        nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
        onupdate=lambda: datetime.now(timezone.utc),
        nullable=False
    )

Step 5: Pydantic v2 Request & Response Schemas

In Pydantic v2, schemas separate the validation rules for incoming user data from the serialization format of outgoing responses:

# app/schemas/product.py
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, ConfigDict

# Base schema with shared validation fields
class ProductBase(BaseModel):
    title: str = Field(..., min_length=2, max_length=150, description="Product display name")
    description: Optional[str] = Field(None, max_length=1000)
    price: float = Field(..., gt=0, description="Price in USD, must be greater than zero")
    in_stock: bool = Field(default=True)

# Schema for creating a product (requires slug)
class ProductCreate(ProductBase):
    slug: str = Field(..., min_length=2, max_length=160, pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")

# Schema for updating a product (all fields optional)
class ProductUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=2, max_length=150)
    description: Optional[str] = Field(None, max_length=1000)
    price: Optional[float] = Field(None, gt=0)
    in_stock: Optional[bool] = None

# Schema for outgoing API responses
class ProductResponse(ProductBase):
    id: int
    slug: str
    created_at: datetime
    updated_at: datetime

    # Pydantic v2 ORM compatibility (replaces orm_mode = True)
    model_config = ConfigDict(from_attributes=True)

Step 6: Implementing CRUD Endpoints with `APIRouter`

Create RESTful route handlers using standard HTTP methods and status codes:

# app/api/v1/endpoints/products.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from app.db.session import get_db
from app.models.product import Product
from app.schemas.product import ProductCreate, ProductUpdate, ProductResponse

router = APIRouter(prefix="/products", tags=["Products"])

# 1. GET /products (List with Pagination)
@router.get("/", response_model=List[ProductResponse])
async def list_products(
    skip: int = Query(0, ge=0, description="Number of records to skip"),
    limit: int = Query(20, ge=1, le=100, description="Max records to return"),
    db: AsyncSession = Depends(get_db)
):
    query = select(Product).offset(skip).limit(limit).order_by(Product.id.desc())
    result = await db.execute(query)
    return result.scalars().all()

# 2. POST /products (Create new product)
@router.post("/", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
async def create_product(
    payload: ProductCreate,
    db: AsyncSession = Depends(get_db)
):
    # Check if slug already exists
    existing = await db.execute(select(Product).where(Product.slug == payload.slug))
    if existing.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"A product with slug '{payload.slug}' already exists."
        )

    product = Product(**payload.model_dump())
    db.add(product)
    await db.commit()
    await db.refresh(product)
    return product

# 3. GET /products/{product_id} (Fetch single product)
@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(
    product_id: int,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Product).where(Product.id == product_id))
    product = result.scalar_one_or_none()
    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found."
        )
    return product

# 4. PUT /products/{product_id} (Partial or Full Update)
@router.put("/{product_id}", response_model=ProductResponse)
async def update_product(
    product_id: int,
    payload: ProductUpdate,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Product).where(Product.id == product_id))
    product = result.scalar_one_or_none()
    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found."
        )

    # Exclude unset fields from the payload
    update_data = payload.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(product, field, value)

    await db.commit()
    await db.refresh(product)
    return product

# 5. DELETE /products/{product_id} (Remove product)
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(
    product_id: int,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Product).where(Product.id == product_id))
    product = result.scalar_one_or_none()
    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found."
        )

    await db.delete(product)
    await db.commit()
    return None

Aggregate your API routers under a central router:

# app/api/v1/router.py
from fastapi import APIRouter
from app.api.v1.endpoints import products

api_router = APIRouter()
api_router.include_router(products.router)

Step 7: Application Entrypoint & Middleware

Configure the root FastAPI instance with CORS, lifespan events, and error handling:

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
from app.api.v1.router import api_router
from app.db.session import engine
from app.db.base import Base

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Ensure database tables are created (or run migrations via Alembic)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown: Dispose database connection pool
    await engine.dispose()

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.VERSION,
    openapi_url=f"{settings.API_V1_STR}/openapi.json",
    lifespan=lifespan,
)

# Configure Cross-Origin Resource Sharing (CORS)
if settings.BACKEND_CORS_ORIGINS:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.BACKEND_CORS_ORIGINS,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

# Include API Router
app.include_router(api_router, prefix=settings.API_V1_STR)

# Health Check Route
@app.get("/health", tags=["Health"])
async def health_check():
    return {"status": "healthy", "version": settings.VERSION}

Step 8: Reusable Authentication via Dependency Injection

FastAPI's dependency injection system makes it straightforward to protect endpoints using API keys or JWT tokens:

# app/core/security.py
from fastapi import Security, HTTPException, status
from fastapi.security import APIKeyHeader

API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
VALID_API_KEY = "my_secure_api_token"

async def verify_api_key(api_key: str = Security(API_KEY_HEADER)):
    if not api_key or api_key != VALID_API_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing API Key"
        )
    return api_key

To protect a specific endpoint, pass the dependency to the route:

@router.post("/", dependencies=[Depends(verify_api_key)])
async def create_secure_product(...):
    ...

Production Deployment & Process Management

In development, you run the application with hot-reloading:

uvicorn app.main:app --reload --host 127.0.0.1 --port 8000

Production ASGI Execution with Uvicorn

In production, run Uvicorn behind a reverse proxy (like Nginx) without reload mode. Use multiple worker processes to utilize available CPU cores:

# Run with 4 worker processes on an internal socket or port
uvicorn app.main:app --workers 4 --host 127.0.0.1 --port 8000 --proxy-headers

Systemd Process Supervisor Configuration

To ensure your FastAPI application restarts automatically on server reboot, configure a Systemd service file on your Linux VPS:

# /etc/systemd/system/fastapi-app.service
[Unit]
Description=FastAPI Production Application
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/var/www/fastapi-production-api
Environment="PATH=/var/www/fastapi-production-api/.venv/bin"
ExecStart=/var/www/fastapi-production-api/.venv/bin/uvicorn app.main:app --workers 4 --host 127.0.0.1 --port 8000 --proxy-headers

Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Enable and start the service:

sudo systemctl daemon-reload
sudo systemctl enable --now fastapi-app
sudo systemctl status fastapi-app

Nginx Reverse Proxy Configuration

Deploying FastAPI behind Nginx provides SSL termination, static file serving, and request buffering. For a complete guide to server setups, read our walkthrough on Hosting Node.js/Express with Nginx on a VPS, which follows the exact same reverse proxy principles:

# /etc/nginx/sites-available/fastapi.example.com
server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # WebSocket and Streaming support
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

Secure the server by restricting unnecessary inbound ports with a Linux UFW firewall.

Summary Checklist for Production FastAPI Services

Use Pydantic v2 model_config = ConfigDict(from_attributes=True) for clean ORM serialization.
Handle database sessions with async with and dependency injection to guarantee connections are released after request completion.
Validate input strictly using Pydantic Field() constraints (e.g., gt=0, regex patterns).
Use lifespan handlers instead of deprecated @app.on_event("startup") and @app.on_event("shutdown") decorators.
Run multiple Uvicorn workers in production behind an Nginx reverse proxy with --proxy-headers enabled.

Keep Learning

Connecting PostgreSQL to your backend? Read our practical guide on How to Connect PostgreSQL to Node.js Using pg and Prisma.
Deploying to a Linux server? Follow our step-by-step tutorial on How to Host Web Applications on a VPS with Nginx.
Hardening server security? Check our Linux UFW Firewall Setup Guide.

Written by Akshay Singh

Full-stack engineer and technical writer at TheDailyDevs. Specializing in practical, production-ready guides for JavaScript, React, Next.js, Node.js, and DevOps infrastructure.

About the Author →Explore All Topics →