Database Integration in Fast APIs

Why Database?

So far data lives in a Python list — when server restarts, everything is gone. Real applications need permanent storage.

In this stage we connect FastAPI to a real database using SQLAlchemy — the most popular Python ORM.

---

What is an ORM?

ORM = Object Relational Mapper

Instead of writing raw SQL:

SELECT * FROM users WHERE id = 1;

You write Python:

db.query(User).filter(User.id == 1).first()

ORM translates Python code to SQL automatically. You already know this concept — TypeORM in NestJS is the same idea.

---

Tools We'll Use

• SQLite — database (comes with Python, zero setup, perfect for learning)
• SQLAlchemy — ORM (translates Python to SQL)
• Alembic — database migrations (we'll keep this simple)

Later you can swap SQLite for PostgreSQL with just one line change — same code works.

---

Installing Dependencies

pip install sqlalchemy

That's it for SQLite. When you switch to PostgreSQL later:

pip install psycopg2-binary

---

Project Structure

Before writing code — let's organize properly. This is real project structure:

fastapi-learning/
├── venv/
├── main.py
├── database.py        ← database connection setup
├── models.py          ← SQLAlchemy table models
├── schemas.py         ← Pydantic models (request/response)
├── crud.py            ← database operations (create, read, update, delete)
└── requirements.txt

This separation is standard practice. Each file has one clear responsibility — same principle as NestJS modules.

---

Step 1 — database.py — Database Connection

    # database.py
    from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    # SQLite database URL
    # This creates a file called 'app.db' in your project folder
    DATABASE_URL = "sqlite:///./app.db"
    # For PostgreSQL later — just change this one line:
    # DATABASE_URL = "postgresql://username:password@localhost/dbname"
    # Create engine — the core connection to database
    engine = create_engine(
        DATABASE_URL,
        connect_args={"check_same_thread": False}  # needed only for SQLite
    )
    # SessionLocal — each request gets its own database session
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    # Base — all your database models will inherit from this
    Base = declarative_base()

    # Dependency — gives a database session to each route
    # Automatically closes session when request is done
    def get_db():
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()

Think of get_db as middleware that gives each request its own database connection and closes it cleanly when done.

---

Step 2 — models.py — Database Tables

SQLAlchemy models define your database tables:

    # models.py
    from sqlalchemy import Column, Integer, String, Boolean, Float, DateTime, ForeignKey, Text
    from sqlalchemy.orm import relationship
    from sqlalchemy.sql import func
    from database import Base

    class User(Base):
        __tablename__ = "users"
        id = Column(Integer, primary_key=True, index=True)
        name = Column(String(50), nullable=False)
        email = Column(String(100), unique=True, nullable=False, index=True)
        age = Column(Integer, nullable=False)
        city = Column(String(50), default="Unknown")
        is_active = Column(Boolean, default=True)
        created_at = Column(DateTime, server_default=func.now())
        # Relationship — one user has many posts
        posts = relationship("Post", back_populates="author")
    class Post(Base):
        __tablename__ = "posts"
        id = Column(Integer, primary_key=True, index=True)
        title = Column(String(100), nullable=False)
        content = Column(Text, nullable=False)
        published = Column(Boolean, default=False)
        created_at = Column(DateTime, server_default=func.now())
        # Foreign key — connects post to user
        author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
        # Relationship — each post belongs to one user
        author = relationship("User", back_populates="posts")

This is very similar to TypeORM entities you write in NestJS. Column types, constraints, relationships — same concepts.

Key column options:

Option	Meaning
primary_key=True	This is the primary key
nullable=False	Required field, cannot be NULL
unique=True	No duplicate values allowed
index=True	Creates database index for faster queries
default=	Default value in Python
server_default=	Default value set by database

---

Step 3 — schemas.py — Pydantic Models

These are the same Pydantic models from Stage 3 — request and response shapes:

    # schemas.py
    from pydantic import BaseModel, Field, field_validator
    from datetime import datetime

    # ── User Schemas ─────────────────────────────────
    class UserCreate(BaseModel):
        name: str = Field(min_length=2, max_length=50)
        email: str = Field(examples=["gagan@email.com"])
        age: int = Field(ge=0, le=150)
        city: str = Field(default="Unknown")
        @field_validator("email")
        @classmethod
        def normalize_email(cls, value: str) -> str:
            return value.lower().strip()

    class UserUpdate(BaseModel):
        name: str | None = Field(default=None, min_length=2)
        email: str | None = None
        age: int | None = Field(default=None, ge=0, le=150)
        city: str | None = None

    class UserResponse(BaseModel):
        id: int
        name: str
        email: str
        age: int
        city: str
        is_active: bool
        created_at: datetime
        model_config = {"from_attributes": True}    # allows reading SQLAlchemy objects
    # ── Post Schemas ─────────────────────────────────
    class PostCreate(BaseModel):
        title: str = Field(min_length=5, max_length=100)
        content: str = Field(min_length=20)
        author_id: int

    class PostResponse(BaseModel):
        id: int
        title: str
        content: str
        published: bool
        author_id: int
        created_at: datetime
        model_config = {"from_attributes": True}

    class PostWithAuthor(BaseModel):
        id: int
        title: str
        content: str
        published: bool
        created_at: datetime
        author: UserResponse        # nested user response
        model_config = {"from_attributes": True}

---

Step 4 — crud.py — Database Operations

This is where all database queries live. Separating them here keeps routes clean:

    # crud.py
    from sqlalchemy.orm import Session
    from sqlalchemy import or_
    import models
    import schemas

    # ── User CRUD ─────────────────────────────────────
    def get_user(db: Session, user_id: int):
        return db.query(models.User).filter(models.User.id == user_id).first()

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

    def get_users(db: Session, skip: int = 0, limit: int = 10):
        return db.query(models.User).offset(skip).limit(limit).all()

    def search_users(db: Session, name: str | None = None, city: str | None = None):
        query = db.query(models.User)
        if name:
            query = query.filter(models.User.name.ilike(f"%{name}%"))
        if city:
            query = query.filter(models.User.city.ilike(f"%{city}%"))
        return query.all()

    def create_user(db: Session, user: schemas.UserCreate):
        db_user = models.User(**user.model_dump())
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user

    def update_user(db: Session, user_id: int, user: schemas.UserUpdate):
        db_user = get_user(db, user_id)
        if not db_user:
            return None
        update_data = user.model_dump(exclude_none=True)
        for field, value in update_data.items():
            setattr(db_user, field, value)
        db.commit()
        db.refresh(db_user)
        return db_user

    def delete_user(db: Session, user_id: int):
        db_user = get_user(db, user_id)
        if not db_user:
            return None
        db.delete(db_user)
        db.commit()
        return db_user

    # ── Post CRUD ─────────────────────────────────────
    def get_post(db: Session, post_id: int):
        return db.query(models.Post).filter(models.Post.id == post_id).first()

    def get_posts(
        db: Session,
        skip: int = 0,
        limit: int = 10,
        published: bool | None = None,
        author_id: int | None = None
    ):
        query = db.query(models.Post)
        if published is not None:
            query = query.filter(models.Post.published == published)
        if author_id is not None:
            query = query.filter(models.Post.author_id == author_id)
        return query.offset(skip).limit(limit).all()

    def create_post(db: Session, post: schemas.PostCreate):
        db_post = models.Post(**post.model_dump())
        db.add(db_post)
        db.commit()
        db.refresh(db_post)
        return db_post

    def publish_post(db: Session, post_id: int):
        db_post = get_post(db, post_id)
        if not db_post:
            return None
        db_post.published = True
        db.commit()
        db.refresh(db_post)
        return db_post

    def delete_post(db: Session, post_id: int):
        db_post = get_post(db, post_id)
        if not db_post:
            return None
        db.delete(db_post)
        db.commit()
        return db_post

Understanding the key operations:

Query pattern:

db.query(Model)                    # start query on this table
  .filter(Model.field == value)    # WHERE clause
  .offset(skip)                    # skip N rows (pagination)
  .limit(limit)                    # take only N rows
  .all()                           # fetch all results as list
  .first()                         # fetch first result or None

Create pattern:

db_object = Model(**data)    # create model instance
db.add(db_object)            # stage for insertion
db.commit()                  # actually insert to database
db.refresh(db_object)        # reload from database (gets auto-generated id, created_at etc.)
return db_object

Update pattern:

setattr(db_object, field, value)    # update field value
db.commit()                          # save changes
db.refresh(db_object)                # reload updated data

---

Step 5 — main.py — Routes

Now routes are clean — they just call crud functions:

    # main.py
    from fastapi import FastAPI, HTTPException, Depends, status
    from sqlalchemy.orm import Session
    import models
    import schemas
    import crud
    from database import engine, get_db
    # Create all tables in database automatically
    models.Base.metadata.create_all(bind=engine)
    app = FastAPI(title="Users & Posts API", version="3.0.0")

    # ── User Routes ───────────────────────────────────
    @app.get("/")
    def root():
        return {"message": "API is running"}

    @app.post("/users", response_model=schemas.UserResponse, status_code=201)
    def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
        # Check duplicate email
        existing = crud.get_user_by_email(db, user.email)
        if existing:
            raise HTTPException(status_code=400, detail="Email already registered")
        return crud.create_user(db, user)

    @app.get("/users", response_model=list[schemas.UserResponse])
    def get_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
        return crud.get_users(db, skip, limit)

    @app.get("/users/search", response_model=list[schemas.UserResponse])
    def search_users(
        name: str | None = None,
        city: str | None = None,
        db: Session = Depends(get_db)
    ):
        return crud.search_users(db, name, city)

    @app.get("/users/{user_id}", response_model=schemas.UserResponse)
    def get_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

    @app.patch("/users/{user_id}", response_model=schemas.UserResponse)
    def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
        updated = crud.update_user(db, user_id, user)
        if not updated:
            raise HTTPException(status_code=404, detail="User not found")
        return updated

    @app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
    def delete_user(user_id: int, db: Session = Depends(get_db)):
        deleted = crud.delete_user(db, user_id)
        if not deleted:
            raise HTTPException(status_code=404, detail="User not found")

    # ── Post Routes ───────────────────────────────────
    @app.post("/posts", response_model=schemas.PostResponse, status_code=201)
    def create_post(post: schemas.PostCreate, db: Session = Depends(get_db)):
        # Check author exists
        author = crud.get_user(db, post.author_id)
        if not author:
            raise HTTPException(status_code=404, detail="Author not found")
        return crud.create_post(db, post)

    @app.get("/posts", response_model=list[schemas.PostResponse])
    def get_posts(
        skip: int = 0,
        limit: int = 10,
        published: bool | None = None,
        author_id: int | None = None,
        db: Session = Depends(get_db)
    ):
        return crud.get_posts(db, skip, limit, published, author_id)

    @app.get("/posts/{post_id}", response_model=schemas.PostWithAuthor)
    def get_post(post_id: int, db: Session = Depends(get_db)):
        post = crud.get_post(db, post_id)
        if not post:
            raise HTTPException(status_code=404, detail="Post not found")
        return post

    @app.post("/posts/{post_id}/publish", response_model=schemas.PostResponse)
    def publish_post(post_id: int, db: Session = Depends(get_db)):
        post = crud.publish_post(db, post_id)
        if not post:
            raise HTTPException(status_code=404, detail="Post not found")
        return post

    @app.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
    def delete_post(post_id: int, db: Session = Depends(get_db)):
        deleted = crud.delete_post(db, post_id)
        if not deleted:
            raise HTTPException(status_code=404, detail="Post not found")

---

Understanding Depends(get_db)

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

Depends(get_db) is FastAPI's Dependency Injection. Every time this route is called:

1. FastAPI calls get_db()
2. get_db() creates a database session
3. Passes that session as db to your function
4. After function finishes — get_db() automatically closes the session

You already know this pattern from NestJS — it's constructor injection. Same concept, FastAPI style.

---

Running and Testing

uvicorn main:app --reload

When you run this the first time — SQLAlchemy automatically creates app.db file and all tables inside it. No manual SQL needed.

Check /docs — test this flow:

1. Create a user:

POST /users
{
    "name": "Gagan Singh",
    "email": "gagan@email.com",
    "age": 22,
    "city": "Delhi"
}

2. Create a post for that user:

POST /posts
{
    "title": "Learning FastAPI",
    "content": "FastAPI is an amazing framework for building APIs with Python",
    "author_id": 1
}

3. Get post with author details:

GET /posts/1

Response includes full author object nested inside:

{
    "id": 1,
    "title": "Learning FastAPI",
    "content": "...",
    "published": false,
    "created_at": "2025-03-03T10:30:00",
    "author": {
        "id": 1,
        "name": "Gagan Singh",
        "email": "gagan@email.com",
        "age": 22,
        "city": "Delhi",
        "is_active": true,
        "created_at": "2025-03-03T10:29:00"
    }
}

4. Publish the post:

POST /posts/1/publish

5. Restart server — data is still there. Unlike the in-memory list — SQLite persists everything.

---

Switching to PostgreSQL

When you're ready for production — just change one line in database.py:

# SQLite (development)
DATABASE_URL = "sqlite:///./app.db"

# PostgreSQL (production) — same as your current NestJS setup
DATABASE_URL = "postgresql://username:password@localhost/dbname"

And install the driver:

pip install psycopg2-binary

Everything else — models, crud, routes — stays exactly the same.

---

Viewing Your Database

Since it's SQLite you can view the app.db file using:

• DB Browser for SQLite — free GUI tool — sqlitebrowser.org
• VS Code SQLite extension — search "SQLite Viewer" in extensions

This lets you see your tables and data visually — like pgAdmin for PostgreSQL.

---

Common SQLAlchemy Queries — Reference

# Get all
db.query(Model).all()

# Get by id
db.query(Model).filter(Model.id == id).first()

# Filter with multiple conditions
db.query(Model).filter(
    Model.age > 18,
    Model.city == "Delhi"
).all()

# OR condition
from sqlalchemy import or_
db.query(Model).filter(
    or_(Model.name == "Gagan", Model.name == "Rahul")
).all()

# Case insensitive search (like ILIKE in PostgreSQL)
db.query(Model).filter(Model.name.ilike("%gagan%")).all()

# Order by
db.query(Model).order_by(Model.created_at.desc()).all()

# Count
db.query(Model).filter(Model.is_active == True).count()

# Pagination
db.query(Model).offset(skip).limit(limit).all()

# Create
obj = Model(field1=val1, field2=val2)
db.add(obj)
db.commit()
db.refresh(obj)

# Update
obj.field = new_value
db.commit()

# Delete
db.delete(obj)
db.commit()

---

Exercise 🏋️

Extend the current project — add a Products feature:

Add to models.py:

class Category(Base):
    __tablename__ = "categories"
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    products = relationship("Product", back_populates="category")

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    price = Column(Float, nullable=False)
    stock = Column(Integer, default=0)
    category_id = Column(Integer, ForeignKey("categories.id"))
    category = relationship("Category", back_populates="products")

Build these routes:

• POST /categories — create category
• GET /categories — list all categories
• POST /products — create product (validate category exists)
• GET /products — list products with filters: category_id, min_price, max_price, in_stock
• GET /products/{id} — get product with category details nested
• PATCH /products/{id} — update price or stock
• DELETE /products/{id} — delete product

Add to schemas.py:

• CategoryCreate, CategoryResponse
• ProductCreate, ProductResponse, ProductWithCategory