SQLAlchemy with FastAPI (Build Production API Step-by-Step)

 

Introduction

In this blog, we’ll build a production-ready REST API using:

• FastAPI → High-performance web framework
• SQLAlchemy → Database ORM
• (Optional) Alembic → Migrations.

By the end, you’ll have a complete backend structure used in real-world projects.

What We’re Building

A simple User Management API:

• Create user
• Get all users
• Get single user
• Update user
• Delete user

Step 1: Install Dependencies

pip install fastapi uvicorn sqlalchemy

Optional (for production):

pip install psycopg2 alembic

Step 2: Project Structure

project/
│
├── app/
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   ├── schemas.py
│   └── crud.py
│
├── alembic/
└── alembic.ini

Step 3: Database Setup

database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)

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

Base = declarative_base()


Step 4: Define Models
models.py
from sqlalchemy import Column, Integer, String
from .database import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    email = Column(String, unique=True, index=True)

Step 5: Create Schemas (Pydantic)

schemas.py

from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        orm_mode = True

Step 6: CRUD Operations

crud.py




from sqlalchemy.orm import Session

from . import models, schemas

def create_user(db: Session, user: schemas.UserCreate):

    db_user = models.User(name=user.name, email=user.email)

    db.add(db_user)

    db.commit()

    db.refresh(db_user)

    return db_user

def get_users(db: Session):

    return db.query(models.User).all()

def get_user(db: Session, user_id: int):

    return db.query(models.User).filter(models.User.id == user_id).first()

def update_user(db: Session, user_id: int, user_data):

    user = get_user(db, user_id)

    if user:

        user.name = user_data.name

        user.email = user_data.email

        db.commit()

        db.refresh(user)

    return user

def delete_user(db: Session, user_id: int):

    user = get_user(db, user_id)

    if user:

        db.delete(user)

        db.commit()

    return user

Step 7: FastAPI App

main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

from . import models, schemas, crud
from .database import engine, SessionLocal

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

Dependency (DB Session)

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

API Endpoints
Create User
@app.post("/users", response_model=schemas.UserResponse)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    return crud.create_user(db, user)
Get All Users
@app.get("/users", response_model=list[schemas.UserResponse])
def read_users(db: Session = Depends(get_db)):
    return crud.get_users(db)
Get Single User
@app.get("/users/{user_id}", response_model=schemas.UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = crud.get_user(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user
Update User
@app.put("/users/{user_id}", response_model=schemas.UserResponse)
def update_user(user_id: int, user: schemas.UserCreate, db: Session = Depends(get_db)):
    updated_user = crud.update_user(db, user_id, user)
    if not updated_user:
        raise HTTPException(status_code=404, detail="User not found")
    return updated_user
Delete User
@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = crud.delete_user(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"message": "User deleted"}

Step 8: Run the Application

uvicorn app.main:app --reload

Open:

http://127.0.0.1:8000/docs

You’ll see Swagger UI automatically!

Production Improvements

Use PostgreSQL

DATABASE_URL = "postgresql://user:password@localhost/dbname"

Add Alembic

For migrations (Blog 6)

Add Validation

• Email validation
• Password hashing (bcrypt)

Async Support (Advanced)

Use:

from sqlalchemy.ext.asyncio import create_async_engine

Common Mistakes

• Not closing DB session
• Using global session
• No validation
• No error handling
• Direct DB logic inside routes

Real-World Architecture

Routes → CRUD Layer → Models → Database

Clean, scalable, maintainable

• FastAPI + SQLAlchemy = powerful backend stack
• Use layered architecture (models, schemas, CRUD)
• Dependency injection for DB sessions
• Ready for production with minor upgrades