FastAPI PostgreSQL Tutorial: Build Production-Ready Python APIs

Building robust, scalable REST APIs is a cornerstone of modern software development. FastAPI has emerged as the go-to framework for Python developers seeking high performance and developer-friendly features. Combined with PostgreSQL's reliability and advanced features, this stack provides an excellent foundation for production-grade applications.

In this comprehensive guide, we'll walk through creating a complete API system from database design to deployment, covering authentication, testing, and performance optimization strategies that work in real-world scenarios.

Setting Up Your Development Environment

Before diving into code, let's establish a solid development foundation. We'll use Docker for consistent environments and modern Python dependency management.

First, create your project structure:

`bash mkdir fastapi-postgres-api cd fastapi-postgres-api mkdir app tests touch requirements.txt docker-compose.yml `

Your requirements.txt should include:

`txt fastapi==0.104.1 uvicorn[standard]==0.24.0 psycopg2-binary==2.9.7 sqlalchemy==2.0.23 alembic==1.12.1 pydantic[email]==2.5.0 python-jose[cryptography]==3.3.0 passlib[bcrypt]==1.7.4 pytest==7.4.3 httpx==0.25.2 `

Create a docker-compose.yml for local development:

`yaml version: '3.8' services: db: image: postgres:15 environment: POSTGRES_DB: apidb POSTGRES_USER: apiuser POSTGRES_PASSWORD: apipass ports: - "5432:5432" volumes: - postgres_data:/var/lib/postgresql/data

api: build: . ports: - "8000:8000" depends_on: - db environment: DATABASE_URL: postgresql://apiuser:apipass@db:5432/apidb volumes: - .:/app command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

volumes: postgres_data: `

This setup provides isolated, reproducible environments and eliminates "works on my machine" issues.

Database Design and SQLAlchemy Integration

Effective database design forms the backbone of any successful API. We'll implement a user management system with proper relationships and constraints.

Create app/database.py:

`python from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import os

Database URL from environment variable

engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Dependency for database sessions

Now, define your models in app/models.py:

`python from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey, Text from sqlalchemy.relationship import relationship from sqlalchemy.sql import func from .database import Base

class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) email = Column(String, unique=True, index=True, nullable=False) hashed_password = Column(String, nullable=False) full_name = Column(String, nullable=False) is_active = Column(Boolean, default=True) is_superuser = Column(Boolean, default=False) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), onupdate=func.now()) # Relationship to posts posts = relationship("Post", back_populates="author")

class Post(Base): __tablename__ = "posts" id = Column(Integer, primary_key=True, index=True) title = Column(String, nullable=False) content = Column(Text) published = Column(Boolean, default=False) author_id = Column(Integer, ForeignKey("users.id")) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), onupdate=func.now()) # Relationship to user author = relationship("User", back_populates="posts") `

This design includes proper indexing, relationships, and timestamp tracking—essential for production applications.

Building FastAPI Endpoints with Pydantic Models

FastAPI's integration with Pydantic provides automatic request validation, serialization, and API documentation. Let's create comprehensive schemas and endpoints.

Create app/schemas.py:

`python from pydantic import BaseModel, EmailStr, validator from datetime import datetime from typing import List, Optional

User schemas

class UserCreate(UserBase): password: str @validator('password') def validate_password(cls, v): if len(v) < 8: raise ValueError('Password must be at least 8 characters long') return v

class UserUpdate(BaseModel): email: Optional[EmailStr] = None full_name: Optional[str] = None is_active: Optional[bool] = None

class User(UserBase): id: int created_at: datetime updated_at: Optional[datetime] class Config: from_attributes = True

Post schemas

class PostCreate(PostBase): pass

class Post(PostBase): id: int author_id: int created_at: datetime author: User class Config: from_attributes = True

Authentication schemas

class TokenData(BaseModel): username: Optional[str] = None `

Now implement the main API in app/main.py:

`python from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from sqlalchemy.orm import Session from typing import List import uvicorn

from . import models, schemas, auth from .database import engine, get_db

Create database tables

app = FastAPI( title="FastAPI PostgreSQL API", description="Production-ready API with authentication and CRUD operations", version="1.0.0" )

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

Authentication endpoints

@app.post("/auth/login", response_model=schemas.Token) async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)): user = auth.authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token = auth.create_access_token(data={"sub": user.email}) return {"access_token": access_token, "token_type": "bearer"}

User endpoints

@app.get("/users/{user_id}", response_model=schemas.User) async def read_user(user_id: int, db: Session = Depends(get_db)): db_user = db.query(models.User).filter(models.User.id == user_id).first() if db_user is None: raise HTTPException(status_code=404, detail="User not found") return db_user

Post endpoints

@app.get("/posts/", response_model=List[schemas.Post]) async def read_posts( skip: int = 0, limit: int = 100, db: Session = Depends(get_db) ): posts = db.query(models.Post).filter(models.Post.published == True).offset(skip).limit(limit).all() return posts

@app.get("/posts/{post_id}", response_model=schemas.Post) async def read_post(post_id: int, db: Session = Depends(get_db)): post = db.query(models.Post).filter(models.Post.id == post_id).first() if post is None: raise HTTPException(status_code=404, detail="Post not found") return post

if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000) `

This implementation provides comprehensive CRUD operations with proper error handling and response models.

Implementing JWT Authentication and Security

Security is paramount in production APIs. We'll implement JWT-based authentication with proper password hashing and token management.

Create app/auth.py:

`python from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from sqlalchemy.orm import Session

from . import models, schemas from .database import get_db

Security configuration

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password): return pwd_context.hash(password)

def get_user_by_email(db: Session, email: str): return db.query(models.User).filter(models.User.email == email).first()

def authenticate_user(db: Session, email: str, password: str): user = get_user_by_email(db, email) if not user: return False if not verify_password(password, user.hashed_password): return False return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception token_data = schemas.TokenData(username=username) except JWTError: raise credentials_exception user = get_user_by_email(db, email=token_data.username) if user is None: raise credentials_exception return user

async def get_current_active_user(current_user: models.User = Depends(get_current_user)): if not current_user.is_active: raise HTTPException(status_code=400, detail="Inactive user") return current_user `

This authentication system provides secure password handling, JWT token generation, and user session management.

Testing Your API with Pytest

Comprehensive testing ensures your API works correctly and helps prevent regressions. Let's implement a robust test suite.

Create tests/conftest.py:

`python import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker

from app.main import app from app.database import get_db, Base from app import models

Test database setup

engine = create_engine(SQLALCHEMY_DATABASE_URL) TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="session") def db(): Base.metadata.create_all(bind=engine) yield Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function") def db_session(db): connection = engine.connect() transaction = connection.begin() session = TestingSessionLocal(bind=connection) yield session session.close() transaction.rollback() connection.close()

@pytest.fixture(scope="function") def client(db_session): def override_get_db(): try: yield db_session finally: db_session.close() app.dependency_overrides[get_db] = override_get_db yield TestClient(app) app.dependency_overrides.clear() `

Now create tests/test_auth.py:

`python def test_register_user(client): response = client.post( "/auth/register", json={ "email": "test@example.com", "password": "testpass123", "full_name": "Test User" } ) assert response.status_code == 200 data = response.json() assert data["email"] == "test@example.com" assert "hashed_password" not in data

def test_login_user(client): # First register a user client.post( "/auth/register", json={ "email": "test@example.com", "password": "testpass123", "full_name": "Test User" } ) # Then login response = client.post( "/auth/login", data={ "username": "test@example.com", "password": "testpass123" } ) assert response.status_code == 200 data = response.json() assert "access_token" in data assert data["token_type"] == "bearer" `

Run your tests with:

`bash pytest tests/ -v `

This testing setup provides database isolation and comprehensive coverage of your API endpoints.

Production Deployment and Performance Optimization

Deploying FastAPI applications requires attention to performance, security, and scalability. Here are proven strategies for production environments.

Database Connection Pooling

Modify your database.py for production:

`python from sqlalchemy import create_engine from sqlalchemy.pool import QueuePool

engine = create_engine( DATABASE_URL, poolclass=QueuePool, pool_size=20, max_overflow=0, pool_recycle=3600, pool_pre_ping=True ) `

Environment Configuration

Create app/config.py:

`python from pydantic import BaseSettings

class Settings(BaseSettings): database_url: str secret_key: str algorithm: str = "HS256" access_token_expire_minutes: int = 30 class Config: env_file = ".env"

settings = Settings() `

Docker Production Setup

Create a production Dockerfile:

`dockerfile FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt

COPY ./app ./app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"] `

Performance Monitoring

Add middleware for request timing:

`python import time from fastapi import Request

@app.middleware("http") async def add_process_time_header(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) return response `

Common Pitfalls and Troubleshooting

Avoid these frequent mistakes that can cause production issues:

1. Database Connection Leaks: Always use dependency injection for database sessions 2. Blocking Operations: Use async/await for I/O operations 3. Missing Validation: Implement comprehensive input validation with Pydantic 4. Security Oversights: Never store secrets in code; use environment variables 5. Poor Error Handling: Implement global exception handlers for consistent error responses

For debugging database issues, enable SQLAlchemy logging:

`python import logging logging.basicConfig() logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO) `

Next Steps and Best Practices

To further enhance your FastAPI PostgreSQL application:

1. Implement Rate Limiting: Use slowapi or similar libraries 2. Add Request Validation: Implement comprehensive input sanitization 3. Set Up Monitoring: Integrate with tools like Prometheus and Grafana 4. Database Migrations: Use Alembic for schema version control 5. API Documentation: Leverage FastAPI's automatic OpenAPI documentation 6. Caching Strategy: Implement Redis for frequently accessed data

Building production-ready APIs requires attention to detail, but FastAPI and PostgreSQL provide an excellent foundation. Focus on security, testing, and performance from the start, and your application will scale effectively as your user base grows.

Remember to monitor your application in production, implement proper logging, and maintain comprehensive test coverage as you add new features. The patterns shown in this guide will serve as a solid foundation for any FastAPI project.