Build a Production REST API with FastAPI and JWT Authentication



Build a Production REST API with FastAPI, SQLAlchemy, and JWT Auth

A complete walkthrough — project structure, database models, JWT authentication, request validation, error handling, and deployment. Everything you need to ship a real API.

Contents

1. Why FastAPI Over Flask/Django
2. Project Structure
3. SQLAlchemy Models
4. Pydantic Schemas
5. JWT Authentication
6. CRUD Endpoints
7. Error Handling
8. Deploy with Docker

Why FastAPI Over Flask/Django?

Feature	FastAPI	Flask	Django REST
Performance	âš¡ Starlette/uvicorn (async)	Sync by default	Sync by default
Auto API Docs	✅ Swagger + ReDoc built-in	Extension needed	Extension needed
Data Validation	✅ Pydantic v2 built-in	Manual	Serializers
Type Safety	✅ Full Python type hints	Optional	Optional
Learning Curve	Low	Very Low	High

Project Structure

myapi/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI app, middleware, startup
│   ├── config.py        # Settings (pydantic BaseSettings)
│   ├── database.py      # SQLAlchemy engine + session
│   ├── models/
│   │   ├── user.py
│   │   └── post.py
│   ├── schemas/
│   │   ├── user.py
│   │   └── post.py
│   ├── routers/
│   │   ├── auth.py
│   │   ├── users.py
│   │   └── posts.py
│   ├── services/
│   │   ├── auth_service.py
│   │   └── post_service.py
│   └── dependencies.py  # get_db, get_current_user
├── tests/
├── Dockerfile
├── docker-compose.yml
└── requirements.txt

SQLAlchemy Models

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.config import settings

engine = create_engine(settings.DATABASE_URL, pool_pre_ping=True, pool_size=10)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, func
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id       = Column(Integer, primary_key=True, index=True)
    email    = Column(String(255), unique=True, index=True, nullable=False)
    username = Column(String(50), unique=True, index=True, nullable=False)
    password = Column(String(255), nullable=False)
    is_active= Column(Boolean, default=True)
    created  = Column(DateTime(timezone=True), server_default=func.now())
    updated  = Column(DateTime(timezone=True), onupdate=func.now())

Pydantic Schemas

# app/schemas/user.py
from pydantic import BaseModel, EmailStr, field_validator
from datetime import datetime

class UserCreate(BaseModel):
    email: EmailStr
    username: str
    password: str

    @field_validator("username")
    @classmethod
    def username_alphanumeric(cls, v: str) -> str:
        if not v.replace("_", "").isalnum():
            raise ValueError("username must be alphanumeric")
        if len(v) < 3 or len(v) > 50:
            raise ValueError("username must be 3-50 characters")
        return v.lower()

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("password must be at least 8 characters")
        return v

class UserResponse(BaseModel):
    id: int
    email: EmailStr
    username: str
    is_active: bool
    created: datetime

    model_config = {"from_attributes": True}

JWT Authentication

pip install python-jose[cryptography] passlib[bcrypt]

# app/services/auth_service.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import HTTPException, status
from app.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_minutes: int = 30) -> str:
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + timedelta(minutes=expires_minutes)
    return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")

def decode_token(token: str) -> dict:
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )

# app/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from app.database import get_db
from app.services.auth_service import decode_token
from app.models.user import User

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

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User:
    payload = decode_token(token)
    user = db.query(User).filter(User.id == payload.get("sub")).first()
    if not user or not user.is_active:
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user

CRUD Endpoints

# app/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserResponse
from app.services.auth_service import hash_password, verify_password, create_access_token

router = APIRouter(prefix="/auth", tags=["Authentication"])

@router.post("/register", response_model=UserResponse, status_code=201)
def register(data: UserCreate, db: Session = Depends(get_db)):
    if db.query(User).filter(User.email == data.email).first():
        raise HTTPException(status_code=400, detail="Email already registered")
    user = User(
        email=data.email,
        username=data.username,
        password=hash_password(data.password)
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

@router.post("/login")
def login(form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = db.query(User).filter(User.email == form.username).first()
    if not user or not verify_password(form.password, user.password):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    token = create_access_token({"sub": str(user.id)})
    return {"access_token": token, "token_type": "bearer"}

# app/routers/posts.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import get_db
from app.dependencies import get_current_user
from app.models.user import User

router = APIRouter(prefix="/posts", tags=["Posts"])

@router.get("/")
def list_posts(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    return db.query(Post).offset(skip).limit(limit).all()

@router.post("/", status_code=201)
def create_post(data: PostCreate, db: Session = Depends(get_db),
                current_user: User = Depends(get_current_user)):
    post = Post(**data.model_dump(), author_id=current_user.id)
    db.add(post); db.commit(); db.refresh(post)
    return post

@router.delete("/{post_id}", status_code=204)
def delete_post(post_id: int, db: Session = Depends(get_db),
                current_user: User = Depends(get_current_user)):
    post = db.query(Post).filter(Post.id == post_id, Post.author_id == current_user.id).first()
    if not post:
        raise HTTPException(404, "Post not found")
    db.delete(post); db.commit()

Global Error Handling

# app/main.py
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI(title="My API", version="1.0.0")

@app.exception_handler(RequestValidationError)
async def validation_error_handler(req: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body}
    )

@app.exception_handler(Exception)
async def global_error_handler(req: Request, exc: Exception):
    # Log to your monitoring system here
    return JSONResponse(status_code=500, content={"detail": "Internal server error"})

Deploy with Docker

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      DATABASE_URL: postgresql://user:pass@db:5432/mydb
      SECRET_KEY: your-256-bit-secret-key
    depends_on: [db]
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes: [pgdata:/var/lib/postgresql/data]
volumes:
  pgdata:

✅ Run it: docker-compose up -d — then visit http://localhost:8000/docs for interactive Swagger UI.

---

Key Takeaways: FastAPI gives you automatic docs, Pydantic validation, async support, and type safety out of the box. Combine with SQLAlchemy for the ORM, JWT for auth, and Docker for consistent deployment.

PythonFastAPIREST APISQLAlchemyJWTDockerBackend