← FastAPI

Database Integration & SQLAlchemy

Performance & Optimization

← Database Integration & SQLAlchemy

Select a question from the sidebar

15 questions in this topic

Question 1 / 15

How do you set up SQLAlchemy with FastAPI?

Setting up SQLAlchemy with FastAPI involves creating an engine, a session factory, and a declarative base. Unlike Flask-SQLAlchemy, FastAPI uses plain SQLAlchemy directly -- there is no official FastAPI-SQLAlchemy extension.

Steps:

Create the engine with create_engine() pointing to your database URL
Create a session factory with sessionmaker() bound to the engine
Define a declarative base using DeclarativeBase (SQLAlchemy 2.0 style)
Create a dependency that yields a session per request
Create tables during application startup using lifespan

SQLAlchemy 2.0 style:

SQLAlchemy 2.0 uses DeclarativeBase instead of declarative_base(), and Mapped/mapped_column instead of Column. This gives you better type hints and IDE support.

Pseudo Code

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase): pass

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

@app.get("/items")
def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()

Code

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, String
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column,
    sessionmaker, Session
)
from contextlib import asynccontextmanager
from typing import Annotated

# Database setup
DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite only
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# SQLAlchemy 2.0 declarative base
class Base(DeclarativeBase):
    pass

# Model
class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)

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

DbDep = Annotated[Session, Depends(get_db)]

# Lifespan to create tables on startup
@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)
    yield

app = FastAPI(lifespan=lifespan)

@app.get("/users")
def list_users(db: DbDep):
    return db.query(User).all()

Key Points

•FastAPI uses plain SQLAlchemy -- no wrapper extension needed
•Use DeclarativeBase and Mapped/mapped_column for SQLAlchemy 2.0 style
•check_same_thread=False is required only for SQLite
•Use lifespan context manager (not deprecated @app.on_event) to create tables at startup

Gotchas

•Forgetting check_same_thread=False for SQLite causes threading errors
•autoflush=True (the default) can cause unexpected queries -- set autoflush=False explicitly
•create_all() does not update existing tables -- use Alembic for migrations
•The session must always be closed, even on exceptions -- yield in a try/finally handles this

Create an engine, a sessionmaker, and a DeclarativeBase. Provide a get_db dependency that yields a session per request and closes it in finally. Use lifespan to create tables at startup.

•Plain SQLAlchemy with sessionmaker and DeclarativeBase
•Yield-based dependency ensures session cleanup

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session
from fastapi import Depends
from typing import Annotated

engine = create_engine("sqlite:///./app.db",
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase): pass

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

DbDep = Annotated[Session, Depends(get_db)]

check_same_thread=False is needed only for SQLite -- omit it for PostgreSQL/MySQL

Question 2 / 15

How do you create a database session dependency using yield?

FastAPI's yield dependencies work like context managers. The code before yield runs before the endpoint, and the code after yield runs as cleanup -- even if the endpoint raises an exception.

How it works:

FastAPI calls get_db(), which creates a session
The yield pauses the generator and hands the session to the endpoint
After the endpoint returns (or raises), execution resumes after yield
The finally block closes the session

Why yield instead of return:

Guaranteed cleanup: finally always runs, preventing connection leaks
Transaction management: You can commit or rollback in the cleanup phase
Resource lifecycle: Matches the request lifecycle perfectly

Annotated pattern:

Using Annotated[Session, Depends(get_db)] as a type alias avoids repeating Depends() everywhere and improves readability.

Pseudo Code

def get_db():
    db = create_session()
    try:
        yield db        # endpoint uses the session
        db.commit()     # auto-commit if no error
    except:
        db.rollback()   # rollback on error
    finally:
        db.close()      # always close

Code

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, String
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column,
    sessionmaker, Session
)
from typing import Annotated

engine = create_engine("sqlite:///./app.db",
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase): pass

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

# Yield dependency with cleanup
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Type alias -- use this everywhere instead of repeating Depends(get_db)
DbDep = Annotated[Session, Depends(get_db)]

app = FastAPI()

@app.post("/items")
def create_item(name: str, db: DbDep):
    item = Item(name=name)
    db.add(item)
    db.commit()
    db.refresh(item)  # reload from DB to get generated id
    return {"id": item.id, "name": item.name}

@app.get("/items/{item_id}")
def read_item(item_id: int, db: DbDep):
    item = db.get(Item, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"id": item.id, "name": item.name}

# Advanced: dependency with auto-commit/rollback
def get_db_transactional():
    db = SessionLocal()
    try:
        yield db
        db.commit()  # commit if endpoint succeeds
    except Exception:
        db.rollback()  # rollback if endpoint raises
        raise
    finally:
        db.close()

Key Points

•yield pauses the generator, handing the session to the endpoint
•Code after yield runs as cleanup, like __exit__ in a context manager
•finally block guarantees the session is closed even on exceptions
•Annotated[Session, Depends(get_db)] creates a reusable type alias

Gotchas

•Yield dependencies cannot catch HTTPException -- FastAPI re-raises it after cleanup
•If you commit in the dependency, db.refresh() in the endpoint may fail after commit
•Only one yield is allowed per dependency function
•Background tasks run after the response is sent, so the yielded session is already closed

A yield dependency creates a session before yield, hands it to the endpoint, then closes it in finally. This guarantees cleanup even if the endpoint raises an exception.

•Code after yield is cleanup, guaranteed by finally

from typing import Annotated
from fastapi import Depends
from sqlalchemy.orm import Session

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

DbDep = Annotated[Session, Depends(get_db)]

@app.get("/items")
def list_items(db: DbDep):
    return db.query(Item).all()

Background tasks run after the response, so the yielded session is already closed

Question 3 / 15

How do you define SQLAlchemy models for use with FastAPI?

SQLAlchemy 2.0 introduces mapped columns with full type annotation support. Models inherit from a DeclarativeBase subclass and use Mapped[type] with mapped_column() for column definitions.

Key concepts:

Mapped[type]: Declares the Python type; also implies NOT NULL
Mapped[Optional[type]]: Makes the column nullable
mapped_column(): Configures column-level details (primary key, unique, default, etc.)
relationship(): Defines ORM relationships with back_populates

Naming conventions:

SQLAlchemy models represent database tables (use plural: users, posts)
Pydantic schemas represent API data shapes (use singular: User, UserCreate)
Keep them in separate files to avoid confusion

Pseudo Code

class Base(DeclarativeBase): pass

class ModelName(Base):
    __tablename__ = "table_name"
    id: Mapped[int] = mapped_column(primary_key=True)
    required_field: Mapped[str] = mapped_column(String(255))
    optional_field: Mapped[Optional[str]] = mapped_column(default=None)
    children: Mapped[list["Child"]] = relationship(back_populates="parent")

Code

from datetime import datetime
from typing import Optional
from sqlalchemy import String, Text, ForeignKey, func
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship
)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    bio: Mapped[Optional[str]] = mapped_column(Text, default=None)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    # One-to-many relationship
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

    def __repr__(self) -> str:
        return f"<User(id={self.id}, username='{self.username}')>"

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column(Text)
    is_published: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    # Foreign key
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    # Many-to-one relationship
    author: Mapped["User"] = relationship(back_populates="posts")

    # Many-to-many through association table
    tags: Mapped[list["Tag"]] = relationship(
        secondary="post_tags", back_populates="posts"
    )

# Association table for many-to-many
from sqlalchemy import Table, Column, Integer
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", Integer, ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    posts: Mapped[list["Post"]] = relationship(
        secondary="post_tags", back_populates="tags"
    )

# Verify model creation
print(User.__tablename__)
print([c.name for c in User.__table__.columns])

users
['id', 'username', 'email', 'bio', 'is_active', 'created_at']

Key Points

•Mapped[type] makes columns NOT NULL; Mapped[Optional[type]] makes them nullable
•mapped_column() replaces Column() with better type inference
•Use relationship() with back_populates for bidirectional relationships
•Many-to-many uses a Table() association table with secondary parameter

Gotchas

•String columns require a length argument in MySQL -- PostgreSQL accepts both String() without length and Text without issue
•server_default uses SQL expressions (runs in DB), while default uses Python values (runs in app)
•Forward references in relationship type hints require quotes: Mapped["Post"]
•Forgetting back_populates on both sides means the relationship is one-directional in Python

Use SQLAlchemy 2.0 style: Mapped[type] for NOT NULL columns, Mapped[Optional[type]] for nullable, mapped_column() for configuration, and relationship() for ORM relations.

•Mapped[type] = NOT NULL, Mapped[Optional[type]] = nullable
•relationship() with back_populates for bidirectional links

from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from typing import Optional

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[Optional[str]] = mapped_column(default=None)
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped["User"] = relationship(back_populates="posts")

String columns need a length for MySQL/SQLite -- PostgreSQL allows unbounded Text

Question 4 / 15

How do you implement CRUD operations in FastAPI?

CRUD (Create, Read, Update, Delete) operations in FastAPI combine Pydantic schemas for validation with SQLAlchemy for database access. The typical pattern separates concerns into:

Layers:

Pydantic schemas: Validate incoming data and shape outgoing responses
CRUD functions: Pure database operations (can be reused)
Route handlers: Orchestrate validation, CRUD, and HTTP responses

Pydantic v2 patterns:

Use model_config = ConfigDict(from_attributes=True) instead of class Config: orm_mode = True
Use model_dump() instead of .dict()
Use model_dump(exclude_unset=True) for partial updates (PATCH)

Pseudo Code

# schemas
class ItemCreate(BaseModel): name: str
class ItemRead(BaseModel): id: int; name: str
    model_config = ConfigDict(from_attributes=True)

# crud
def create_item(db, data): ...
def get_item(db, id): ...
def update_item(db, id, data): ...
def delete_item(db, id): ...

# routes
@app.post("/items", response_model=ItemRead)
def create(data: ItemCreate, db: DbDep): ...

Code

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy import create_engine, String
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column,
    sessionmaker, Session
)
from typing import Annotated, Optional

# --- Database setup ---
engine = create_engine("sqlite:///./app.db",
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase): pass

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    description: Mapped[Optional[str]] = mapped_column(String(500), default=None)
    price: Mapped[float] = mapped_column(default=0.0)

# --- Pydantic schemas (Pydantic v2) ---
class ItemCreate(BaseModel):
    name: str
    description: str | None = None
    price: float = 0.0

class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None

class ItemRead(BaseModel):
    id: int
    name: str
    description: str | None = None
    price: float

    model_config = ConfigDict(from_attributes=True)  # Pydantic v2 ORM mode

# --- CRUD functions ---
def create_item(db: Session, data: ItemCreate) -> Item:
    item = Item(**data.model_dump())
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

def get_item(db: Session, item_id: int) -> Item | None:
    return db.get(Item, item_id)

def get_items(db: Session, skip: int = 0, limit: int = 100) -> list[Item]:
    return db.query(Item).offset(skip).limit(limit).all()

def update_item(db: Session, item_id: int, data: ItemUpdate) -> Item | None:
    item = db.get(Item, item_id)
    if not item:
        return None
    update_data = data.model_dump(exclude_unset=True)  # only fields that were sent
    for field, value in update_data.items():
        setattr(item, field, value)
    db.commit()
    db.refresh(item)
    return item

def delete_item(db: Session, item_id: int) -> bool:
    item = db.get(Item, item_id)
    if not item:
        return False
    db.delete(item)
    db.commit()
    return True

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

DbDep = Annotated[Session, Depends(get_db)]

# --- Routes ---
app = FastAPI()

@app.post("/items", response_model=ItemRead, status_code=201)
def create(data: ItemCreate, db: DbDep):
    return create_item(db, data)

@app.get("/items", response_model=list[ItemRead])
def read_all(db: DbDep, skip: int = 0, limit: int = 100):
    return get_items(db, skip, limit)

@app.get("/items/{item_id}", response_model=ItemRead)
def read_one(item_id: int, db: DbDep):
    item = get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.patch("/items/{item_id}", response_model=ItemRead)
def update(item_id: int, data: ItemUpdate, db: DbDep):
    item = update_item(db, item_id, data)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.delete("/items/{item_id}", status_code=204)
def delete(item_id: int, db: DbDep):
    if not delete_item(db, item_id):
        raise HTTPException(status_code=404, detail="Item not found")

Key Points

•Separate Pydantic schemas for Create, Update, and Read operations
•model_dump(exclude_unset=True) enables partial updates (PATCH)
•ConfigDict(from_attributes=True) replaces old orm_mode for Pydantic v2
•CRUD functions are separated from routes for reusability and testability

Gotchas

•model_dump(exclude_unset=True) vs exclude_none=True: unset means not sent, None means explicitly sent as null
•db.refresh() after commit is needed to load auto-generated fields like id
•response_model validates the output -- if your SQLAlchemy model has extra fields they get filtered
•DELETE returning 204 must return no body -- return None or use Response(status_code=204)

Use separate Pydantic schemas for input (Create/Update) and output (Read). CRUD functions take a session and schema, handle DB operations, and return ORM objects. Use model_dump(exclude_unset=True) for PATCH updates.

•Separate schemas: Create for input, Read for output
•ConfigDict(from_attributes=True) enables ORM object serialization

from pydantic import BaseModel, ConfigDict

class ItemCreate(BaseModel):
    name: str
    price: float = 0.0

class ItemRead(BaseModel):
    id: int
    name: str
    price: float
    model_config = ConfigDict(from_attributes=True)

@app.post("/items", response_model=ItemRead, status_code=201)
def create(data: ItemCreate, db: DbDep):
    item = Item(**data.model_dump())
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

model_dump(exclude_unset=True) for PATCH -- exclude_none=True has different semantics

Question 5 / 15

How does async SQLAlchemy (AsyncSession) work with FastAPI?

Async SQLAlchemy uses create_async_engine and AsyncSession to perform non-blocking database I/O. This is beneficial for high-concurrency FastAPI apps because the event loop is not blocked during database queries.

Requirements:

SQLAlchemy 2.0+ with async support
An async driver: aiosqlite (SQLite), asyncpg (PostgreSQL), aiomysql (MySQL)
Async session factory: async_sessionmaker instead of sessionmaker

Key differences from sync:

Use await for all session operations: await db.execute(), await db.commit()
Use select() statements instead of db.query() (query-style is not async-compatible)
Relationship loading must be explicit (lazy loading triggers sync I/O, which fails)
Use AsyncSession.get() instead of Session.get() with await

Pseudo Code

async_engine = create_async_engine(ASYNC_DB_URL)
AsyncSessionLocal = async_sessionmaker(bind=async_engine)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/items")
async def read_items(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Item))
    return result.scalars().all()

Code

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import String, select
from sqlalchemy.ext.asyncio import (
    create_async_engine, AsyncSession, async_sessionmaker
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from typing import Annotated
from contextlib import asynccontextmanager

# Async engine (note: aiosqlite driver for SQLite)
ASYNC_DATABASE_URL = "sqlite+aiosqlite:///./app.db"

async_engine = create_async_engine(ASYNC_DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

class Base(DeclarativeBase): pass

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

# Async dependency
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

AsyncDbDep = Annotated[AsyncSession, Depends(get_db)]

# Lifespan: create tables async
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await async_engine.dispose()

app = FastAPI(lifespan=lifespan)

@app.post("/items", status_code=201)
async def create_item(name: str, db: AsyncDbDep):
    item = Item(name=name)
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return {"id": item.id, "name": item.name}

@app.get("/items")
async def list_items(db: AsyncDbDep):
    result = await db.execute(select(Item))
    items = result.scalars().all()
    return [{"id": i.id, "name": i.name} for i in items]

@app.get("/items/{item_id}")
async def read_item(item_id: int, db: AsyncDbDep):
    item = await db.get(Item, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Not found")
    return {"id": item.id, "name": item.name}

Key Points

•Use create_async_engine and async_sessionmaker for non-blocking DB I/O
•All session operations require await: execute, commit, refresh, get
•Use select() statements -- db.query() is not async-compatible
•expire_on_commit=False prevents lazy loads after commit (which would fail in async)

Gotchas

•Lazy relationship loading causes errors in async -- use selectinload() or joinedload()
•You need an async driver: asyncpg for PostgreSQL, aiosqlite for SQLite
•run_sync() is needed for DDL operations like create_all in async context
•Mixing sync and async sessions in the same app causes subtle bugs

Use create_async_engine, async_sessionmaker, and AsyncSession with await on all operations. Replace db.query() with select() statements since the query interface is sync-only.

•await all session operations; use select() not query()

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select

async_engine = create_async_engine("sqlite+aiosqlite:///./app.db")
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/items")
async def list_items(db: AsyncDbDep):
    result = await db.execute(select(Item))
    return result.scalars().all()

Lazy loading fails in async -- use selectinload() or joinedload() explicitly

Question 6 / 15

How do you set up Alembic for database migrations with FastAPI?

Alembic is the migration tool for SQLAlchemy. It tracks schema changes in version-controlled migration scripts, so you can evolve your database without dropping tables.

Setup steps:

pip install alembic and run alembic init alembic
Edit alembic/env.py to import your models and set target_metadata
Set sqlalchemy.url in alembic.ini (or override in env.py from environment variables)
Run alembic revision --autogenerate -m "description" to generate a migration
Run alembic upgrade head to apply migrations

Autogenerate detects:

New tables, dropped tables
Added/removed columns
Changed column types (sometimes)
New indexes and constraints

Autogenerate does NOT detect:

Table or column renames (shows as drop + add)
Changes to column data
Custom CHECK constraints

Pseudo Code

# 1. Initialize
alembic init alembic

# 2. Edit env.py:
target_metadata = Base.metadata

# 3. Generate migration
alembic revision --autogenerate -m "add users table"

# 4. Apply
alembic upgrade head

# 5. Rollback
alembic downgrade -1

Code

# alembic/env.py (key modifications)
import os
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# Import your models so Alembic can detect them
from app.database import Base  # your DeclarativeBase
import app.models  # noqa: F401 -- ensures models are registered

config = context.config

# Override sqlalchemy.url from environment variable
db_url = os.environ.get("DATABASE_URL", "sqlite:///./app.db")
config.set_main_option("sqlalchemy.url", db_url)

fileConfig(config.config_file_name)
target_metadata = Base.metadata

def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )
        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

# --- Example migration file (generated by autogenerate) ---
# alembic/versions/001_add_users_table.py
"""
add users table

Revision ID: a1b2c3d4e5f6
Create Date: 2024-01-15 10:30:00.000000
"""
from alembic import op
import sqlalchemy as sa

revision = "a1b2c3d4e5f6"
down_revision = None

def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("username", sa.String(50), nullable=False, unique=True),
        sa.Column("email", sa.String(255), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now()),
    )
    op.create_index("ix_users_email", "users", ["email"])

def downgrade() -> None:
    op.drop_index("ix_users_email", table_name="users")
    op.drop_table("users")

Key Points

•alembic init creates the migration folder structure
•env.py must import your Base.metadata and all model modules
•autogenerate compares models to the DB and creates migration scripts
•Always review autogenerated migrations -- they can miss renames and constraints

Gotchas

•Models must be imported before autogenerate runs, or Alembic will not detect them
•Renames appear as drop + create -- you must manually edit the migration to use op.alter_column()
•SQLite does not support ALTER TABLE for most operations -- Alembic uses batch mode as a workaround
•Do not use Base.metadata.create_all() in production -- use Alembic exclusively

Run alembic init alembic, point env.py at your Base.metadata, then use alembic revision --autogenerate -m "msg" to generate migrations and alembic upgrade head to apply them.

•autogenerate compares your models to the DB schema
•Always review generated migrations before applying

# Terminal commands:
# alembic init alembic
# alembic revision --autogenerate -m "add users table"
# alembic upgrade head
# alembic downgrade -1

# Key line in alembic/env.py:
from app.database import Base
target_metadata = Base.metadata

Models must be imported in env.py or autogenerate will not detect them

Question 7 / 15

How does connection pooling work with SQLAlchemy in FastAPI?

SQLAlchemy's engine maintains a connection pool that reuses database connections instead of opening a new one for each request. This dramatically improves performance.

Pool types:

QueuePool (default for most DBs): Fixed-size pool with overflow
NullPool: No pooling -- new connection every time (useful for serverless)
StaticPool: Single connection shared by all threads (testing only)
AsyncAdaptedQueuePool: Default for async engines

Key parameters:

pool_size: Number of persistent connections (default: 5)
max_overflow: Extra connections allowed beyond pool_size (default: 10)
pool_timeout: Seconds to wait for a connection (default: 30)
pool_recycle: Seconds before a connection is recycled (prevents stale connections)
pool_pre_ping: Test connections before use (handles DB restarts)

Pseudo Code

engine = create_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=1800,
    pool_pre_ping=True
)

Code

from sqlalchemy import create_engine, event, text
from sqlalchemy.pool import QueuePool, NullPool

# Production configuration with connection pooling
engine = create_engine(
    "postgresql://user:pass@localhost/dbname",
    pool_size=10,          # 10 persistent connections
    max_overflow=20,       # up to 20 extra under load
    pool_timeout=30,       # wait 30s for a connection
    pool_recycle=1800,     # recycle connections after 30 min
    pool_pre_ping=True,    # test connection health before use
    poolclass=QueuePool,   # explicit (this is already the default)
)

# Check pool status
pool = engine.pool
print(f"Pool size: {pool.size()}")
print(f"Checked out: {pool.checkedout()}")
print(f"Overflow: {pool.overflow()}")
# Serverless (AWS Lambda, etc.) -- disable pooling
serverless_engine = create_engine(
    "postgresql://user:pass@localhost/dbname",
    poolclass=NullPool,  # new connection every time
)

# Testing -- single shared connection
from sqlalchemy.pool import StaticPool
test_engine = create_engine(
    "sqlite:///:memory:",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,  # one connection for all threads
)

# Event listener: log slow checkouts
@event.listens_for(engine, "checkout")
def receive_checkout(dbapi_connection, connection_record, connection_proxy):
    """Called when a connection is checked out from the pool."""
    pass  # add timing/logging here

# Dispose all connections (e.g., after fork)
# engine.dispose()

Pool size: 10
Checked out: 0
Overflow: 0

Key Points

•QueuePool is the default -- it maintains pool_size connections and can grow to pool_size + max_overflow
•pool_pre_ping=True prevents errors from stale/dropped connections
•pool_recycle prevents connections from going stale (important for MySQL which drops idle connections)
•Use NullPool for serverless and StaticPool for in-memory test databases

Gotchas

•SQLite in-memory databases default to StaticPool (single connection shared) -- file-based SQLite uses NullPool by default
•pool_recycle should be less than the DB server's wait_timeout (MySQL default: 8 hours)
•Connections are returned to the pool when session.close() is called -- always close sessions
•After process forking (e.g., gunicorn), call engine.dispose() to avoid sharing connections across processes

SQLAlchemy's engine maintains a QueuePool of reusable connections. Configure pool_size, max_overflow, pool_recycle, and pool_pre_ping=True for production. Use NullPool for serverless.

•pool_pre_ping=True handles DB restarts gracefully

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql://user:pass@localhost/dbname",
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    pool_pre_ping=True,
)

pool_recycle must be less than the DB server's connection timeout

Question 8 / 15

What is the repository pattern and how do you implement it?

The repository pattern abstracts database access behind a clean interface, separating business logic from data persistence. Each entity gets a repository class with methods like get, list, create, update, delete.

Benefits:

Testability: Mock the repository instead of the database
Swappability: Replace SQLAlchemy with another ORM or even an API without changing business logic
Reusability: Same repository methods used by routes, background tasks, CLI commands
Single Responsibility: Routes handle HTTP, repositories handle data

Implementation approach:

Create a base repository with generic CRUD methods
Create entity-specific repositories that extend the base
Inject repositories as FastAPI dependencies

Pseudo Code

class BaseRepository(Generic[T]):
    def __init__(self, db: Session, model: type[T]):
        self.db = db
        self.model = model

    def get(self, id: int) -> T | None: ...
    def list(self, skip, limit) -> list[T]: ...
    def create(self, data: dict) -> T: ...
    def delete(self, id: int) -> bool: ...

class UserRepository(BaseRepository[User]):
    def get_by_email(self, email: str) -> User | None: ...

Code

from typing import TypeVar, Generic, Annotated
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import select, String
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
)
from sqlalchemy import create_engine
from pydantic import BaseModel, ConfigDict

# --- Setup ---
engine = create_engine("sqlite:///./app.db",
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)

# --- Generic base repository ---
ModelT = TypeVar("ModelT", bound=Base)

class BaseRepository(Generic[ModelT]):
    def __init__(self, db: Session, model: type[ModelT]):
        self.db = db
        self.model = model

    def get(self, entity_id: int) -> ModelT | None:
        return self.db.get(self.model, entity_id)

    def list(self, skip: int = 0, limit: int = 100) -> list[ModelT]:
        stmt = select(self.model).offset(skip).limit(limit)
        return list(self.db.execute(stmt).scalars().all())

    def create(self, data: dict) -> ModelT:
        instance = self.model(**data)
        self.db.add(instance)
        self.db.commit()
        self.db.refresh(instance)
        return instance

    def delete(self, entity_id: int) -> bool:
        instance = self.get(entity_id)
        if not instance:
            return False
        self.db.delete(instance)
        self.db.commit()
        return True

# --- Entity-specific repository ---
class UserRepository(BaseRepository[User]):
    def __init__(self, db: Session):
        super().__init__(db, User)

    def get_by_email(self, email: str) -> User | None:
        stmt = select(User).where(User.email == email)
        return self.db.execute(stmt).scalar_one_or_none()

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

def get_user_repo(db: Annotated[Session, Depends(get_db)]) -> UserRepository:
    return UserRepository(db)

UserRepoDep = Annotated[UserRepository, Depends(get_user_repo)]

# --- Schemas ---
class UserCreate(BaseModel):
    name: str
    email: str

class UserRead(BaseModel):
    id: int
    name: str
    email: str
    model_config = ConfigDict(from_attributes=True)

# --- Routes ---
app = FastAPI()

@app.post("/users", response_model=UserRead, status_code=201)
def create_user(data: UserCreate, repo: UserRepoDep):
    existing = repo.get_by_email(data.email)
    if existing:
        raise HTTPException(status_code=409, detail="Email already registered")
    return repo.create(data.model_dump())

@app.get("/users/{user_id}", response_model=UserRead)
def read_user(user_id: int, repo: UserRepoDep):
    user = repo.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users", response_model=list[UserRead])
def list_users(repo: UserRepoDep, skip: int = 0, limit: int = 100):
    return repo.list(skip, limit)

Key Points

•A generic BaseRepository provides reusable CRUD for any model
•Entity-specific repositories add domain methods like get_by_email
•Repositories are injected as FastAPI dependencies via Depends()
•Routes contain only HTTP logic -- all DB access goes through the repository

Gotchas

•Over-engineering: for small apps, a simple CRUD module is often enough
•The repository should not know about HTTP (no HTTPException inside repositories)
•Transaction boundaries become tricky when multiple repositories need to share a transaction
•Generic type parameters are erased at runtime -- the model must be passed explicitly

The repository pattern wraps database operations in a class with methods like get, list, create. A generic base repository handles CRUD, and entity-specific repos add custom queries. Inject them as FastAPI dependencies.

•Repository abstracts DB access behind a clean interface
•Injected as a FastAPI dependency for testability

class UserRepository:
    def __init__(self, db: Session):
        self.db = db

    def get(self, user_id: int) -> User | None:
        return self.db.get(User, user_id)

    def get_by_email(self, email: str) -> User | None:
        return self.db.execute(
            select(User).where(User.email == email)
        ).scalar_one_or_none()

def get_user_repo(db: Annotated[Session, Depends(get_db)]):
    return UserRepository(db)

@app.get("/users/{user_id}")
def read_user(user_id: int, repo: Annotated[UserRepository, Depends(get_user_repo)]):
    user = repo.get(user_id)
    if not user:
        raise HTTPException(status_code=404)
    return user

Avoid putting HTTP concerns (HTTPException) inside the repository layer

Question 9 / 15

How do you handle database transactions in FastAPI?

A transaction groups multiple database operations into an atomic unit -- either all succeed or all are rolled back. SQLAlchemy sessions implicitly start a transaction on the first operation.

Default behavior:

session.commit() makes changes permanent
session.rollback() undoes all changes since the last commit
session.close() rolls back any uncommitted changes and returns the connection to the pool

Patterns:

Commit per operation: Each CRUD function commits. Simple but no cross-operation atomicity
Commit in endpoint: Endpoint calls multiple operations, then commits once
Commit in dependency: The yield dependency auto-commits on success, auto-rolls back on exception
Nested transactions: session.begin_nested() creates savepoints for partial rollback

Pseudo Code

# Pattern: commit in dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
        db.commit()      # auto-commit if no exception
    except Exception:
        db.rollback()    # auto-rollback on error
        raise
    finally:
        db.close()

Code

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, String, ForeignKey
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
)
from typing import Annotated

engine = create_engine("sqlite:///./app.db",
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase): pass

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    total: Mapped[float] = mapped_column(default=0.0)

class OrderItem(Base):
    __tablename__ = "order_items"
    id: Mapped[int] = mapped_column(primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"))
    product: Mapped[str] = mapped_column(String(100))
    price: Mapped[float]

# Pattern 1: Auto-commit dependency
def get_db_transactional():
    db = SessionLocal()
    try:
        yield db
        db.commit()  # commit if endpoint succeeds
    except Exception:
        db.rollback()  # rollback if endpoint raises
        raise
    finally:
        db.close()

DbDep = Annotated[Session, Depends(get_db_transactional)]

app = FastAPI()

# Pattern 2: Multiple operations in one transaction
@app.post("/orders")
def create_order(items: list[dict], db: DbDep):
    order = Order()
    db.add(order)
    db.flush()  # gets order.id without committing

    total = 0.0
    for item_data in items:
        order_item = OrderItem(
            order_id=order.id,
            product=item_data["product"],
            price=item_data["price"],
        )
        db.add(order_item)
        total += item_data["price"]

    order.total = total
    # commit happens in the dependency after this returns
    return {"order_id": order.id, "total": total}

# Pattern 3: Savepoints for partial rollback
@app.post("/orders/{order_id}/items")
def add_item_to_order(order_id: int, product: str, price: float, db: DbDep):
    order = db.get(Order, order_id)
    if not order:
        raise HTTPException(status_code=404, detail="Order not found")

    # Savepoint: if adding the item fails, only this part rolls back
    savepoint = db.begin_nested()
    try:
        item = OrderItem(order_id=order_id, product=product, price=price)
        db.add(item)
        db.flush()
        order.total += price
    except Exception:
        savepoint.rollback()
        raise HTTPException(status_code=400, detail="Failed to add item")

    return {"item_id": item.id, "order_total": order.total}

Key Points

•SQLAlchemy sessions implicitly start a transaction on first operation
•flush() sends SQL to the DB without committing -- useful for getting auto-generated IDs
•begin_nested() creates a savepoint for partial rollback within a transaction
•A yield dependency can auto-commit on success and auto-rollback on exception

Gotchas

•flush() is NOT a commit -- data is not visible to other sessions and will be lost on rollback
•HTTPException raised by the endpoint is caught by the dependency's except block, triggering rollback
•Nested transactions (savepoints) require database support -- SQLite has limited support
•autoflush=True (default) can trigger unexpected SQL -- set to False for explicit control

SQLAlchemy sessions auto-start a transaction. Use commit() to persist, rollback() to undo, and flush() to send SQL without committing. A yield dependency can auto-commit/rollback based on whether the endpoint raises.

•flush() gets auto-generated IDs without committing
•Yield dependency auto-commits on success, rolls back on error

def get_db_transactional():
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

@app.post("/orders")
def create_order(items: list[dict], db: DbDep):
    order = Order()
    db.add(order)
    db.flush()  # get order.id without committing
    for item in items:
        db.add(OrderItem(order_id=order.id, **item))
    # auto-commits via dependency

flush() data is not visible to other sessions and is lost on rollback

Question 10 / 15

How do you implement pagination for database queries?

Pagination avoids loading entire tables into memory. The two main approaches are:

1. Offset/Limit pagination:

Simple: SELECT * FROM items OFFSET 20 LIMIT 10
Problem: Slow for deep pages because the DB still scans skipped rows
Best for: Small datasets, admin panels, when users rarely go past page 5

2. Cursor-based (keyset) pagination:

Uses the last seen ID/timestamp: WHERE id > last_id LIMIT 10
Advantage: Consistent performance regardless of page depth
Best for: Feeds, infinite scroll, large datasets, real-time data

FastAPI implementation:

Use Query parameters with defaults and validation
Return pagination metadata alongside results
Create a reusable Pydantic response model for paginated results

Pseudo Code

# Offset-based
@app.get("/items")
def list_items(skip: int = 0, limit: int = 20):
    items = db.query(Item).offset(skip).limit(limit).all()
    total = db.query(func.count(Item.id)).scalar()
    return {"items": items, "total": total}

# Cursor-based
@app.get("/items")
def list_items(after: int | None = None, limit: int = 20):
    query = select(Item).order_by(Item.id)
    if after:
        query = query.where(Item.id > after)
    items = db.execute(query.limit(limit + 1))  # fetch one extra
    has_more = len(items) > limit

Code

from fastapi import FastAPI, Depends, Query
from pydantic import BaseModel, ConfigDict
from sqlalchemy import create_engine, String, func, select
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
)
from typing import Annotated, Generic, TypeVar

engine = create_engine("sqlite:///./app.db",
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase): pass

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

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

DbDep = Annotated[Session, Depends(get_db)]

# --- Reusable paginated response model ---
T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    page_size: int
    total_pages: int

class ItemRead(BaseModel):
    id: int
    name: str
    model_config = ConfigDict(from_attributes=True)

app = FastAPI()

# --- Offset-based pagination ---
@app.get("/items", response_model=PaginatedResponse[ItemRead])
def list_items(
    db: DbDep,
    page: Annotated[int, Query(ge=1)] = 1,
    page_size: Annotated[int, Query(ge=1, le=100)] = 20,
):
    total = db.execute(select(func.count(Item.id))).scalar_one()
    offset = (page - 1) * page_size
    items = db.execute(
        select(Item).offset(offset).limit(page_size)
    ).scalars().all()

    return PaginatedResponse(
        items=items,
        total=total,
        page=page,
        page_size=page_size,
        total_pages=(total + page_size - 1) // page_size,
    )

# --- Cursor-based pagination ---
class CursorResponse(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: int | None = None
    has_more: bool

@app.get("/items/feed", response_model=CursorResponse[ItemRead])
def list_items_cursor(
    db: DbDep,
    after: int | None = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
    query = select(Item).order_by(Item.id)
    if after is not None:
        query = query.where(Item.id > after)

    # Fetch one extra to check if there are more
    results = db.execute(query.limit(limit + 1)).scalars().all()
    has_more = len(results) > limit
    items = results[:limit]

    return CursorResponse(
        items=items,
        next_cursor=items[-1].id if items and has_more else None,
        has_more=has_more,
    )

Key Points

•Offset/limit is simple but slows down on deep pages
•Cursor-based pagination has consistent performance regardless of depth
•Fetch limit+1 rows to determine if there are more pages without a separate COUNT query
•Validate page/limit with Query(ge=1, le=100) to prevent abuse

Gotchas

•Offset pagination can skip or duplicate rows if data is inserted between page requests
•COUNT queries on large tables can be slow -- consider caching or approximate counts
•Cursor pagination requires a unique, ordered column (usually id or created_at)
•Generic Pydantic models (BaseModel with Generic[T]) work on Python 3.8+ with Pydantic v2 -- use typing_extensions if needed for older Python versions

Use offset/limit for simple pagination or cursor-based (WHERE id > last_id) for large datasets. Validate page parameters with Query(ge=1, le=100) and return metadata (total, page, has_more) alongside results.

•Offset/limit is simple; cursor-based scales better

@app.get("/items")
def list_items(
    db: DbDep,
    page: Annotated[int, Query(ge=1)] = 1,
    page_size: Annotated[int, Query(ge=1, le=100)] = 20,
):
    total = db.execute(select(func.count(Item.id))).scalar_one()
    offset = (page - 1) * page_size
    items = db.execute(
        select(Item).offset(offset).limit(page_size)
    ).scalars().all()
    return {"items": items, "total": total, "page": page}

Offset pagination can miss or duplicate rows during concurrent inserts

Question 11 / 15

How do you use Pydantic models alongside SQLAlchemy models?

FastAPI apps use two model layers: SQLAlchemy models for the database and Pydantic models (schemas) for API validation and serialization. They serve different purposes and should be kept separate.

Why two model layers:

SQLAlchemy models: Define tables, columns, relationships, constraints. They interact with the DB
Pydantic schemas: Define what the API accepts (input) and returns (output). They validate and serialize

Common schema patterns:

Create schema: Fields for creating (no id, no timestamps)
Update schema: All fields optional (for PATCH)
Read schema: All fields including id, timestamps, computed values
Base schema: Shared fields to avoid repetition (inheritance)

Pydantic v2 changes:

from_attributes=True replaces orm_mode = True
model_dump() replaces .dict()
model_dump(exclude_unset=True) for partial updates
ConfigDict replaces class Config

Pseudo Code

class ItemBase(BaseModel):
    name: str
    price: float

class ItemCreate(ItemBase): pass

class ItemUpdate(BaseModel):
    name: str | None = None
    price: float | None = None

class ItemRead(ItemBase):
    id: int
    created_at: datetime
    model_config = ConfigDict(from_attributes=True)

Code

from datetime import datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict, field_validator
from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# --- SQLAlchemy model (database layer) ---
class Base(DeclarativeBase): pass

class UserDB(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(255))
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

# --- Pydantic schemas (API layer) ---
class UserBase(BaseModel):
    """Shared fields between Create and Read."""
    username: str
    email: str

class UserCreate(UserBase):
    """Fields for creating a user (includes password, excludes id)."""
    password: str  # plain text, will be hashed before saving

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

class UserUpdate(BaseModel):
    """All optional for PATCH updates."""
    username: str | None = None
    email: str | None = None
    is_active: bool | None = None

class UserRead(UserBase):
    """Fields returned to the client (no password)."""
    id: int
    is_active: bool
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

# --- Usage ---
# Creating from Pydantic to SQLAlchemy
create_data = UserCreate(
    username="alice", email="alice@example.com", password="securepass123"
)
print(create_data.model_dump())
# pass to SQLAlchemy: UserDB(**create_data.model_dump(exclude={"password"}))

# Reading from SQLAlchemy to Pydantic (simulated)
class FakeUser:
    id = 1
    username = "alice"
    email = "alice@example.com"
    is_active = True
    created_at = datetime(2024, 1, 15)

user_read = UserRead.model_validate(FakeUser())
print(user_read.model_dump())

# Partial update
update_data = UserUpdate(email="newemail@example.com")
print(update_data.model_dump(exclude_unset=True))  # only email

{'username': 'alice', 'email': 'alice@example.com', 'password': 'securepass123'}
{'username': 'alice', 'email': 'alice@example.com', 'id': 1, 'is_active': True, 'created_at': datetime.datetime(2024, 1, 15, 0, 0)}
{'email': 'newemail@example.com'}

Key Points

•SQLAlchemy models handle the DB; Pydantic schemas handle API validation and serialization
•Use schema inheritance (UserBase -> UserCreate, UserRead) to avoid field duplication
•from_attributes=True (Pydantic v2) enables creating schemas from ORM objects
•model_dump(exclude_unset=True) gives only the fields the client actually sent

Gotchas

•Never expose hashed_password in UserRead -- keep sensitive fields out of response schemas
•model_validate() (not model_validate_json()) is used for ORM objects
•exclude_unset vs exclude_none: unset means not in the request body, None means explicitly null
•Pydantic v2 does not auto-convert lazy-loaded relationships -- access them before serialization

Use separate Pydantic schemas (Create, Update, Read) alongside SQLAlchemy models. from_attributes=True enables ORM-to-Pydantic conversion. Use model_dump(exclude_unset=True) for PATCH updates.

•Separate schemas for Create (input), Read (output), Update (partial)
•from_attributes=True replaces the old orm_mode = True

from pydantic import BaseModel, ConfigDict

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

class UserRead(BaseModel):
    id: int
    username: str
    email: str
    model_config = ConfigDict(from_attributes=True)

# ORM object -> Pydantic: UserRead.model_validate(db_user)
# Pydantic -> dict: data.model_dump(exclude_unset=True)

Never include sensitive fields like hashed_password in response schemas

Question 12 / 15

How do you handle database errors and rollbacks?

Database operations can fail due to constraint violations, connection issues, or data integrity problems. Proper error handling involves catching SQLAlchemy exceptions, rolling back the transaction, and returning meaningful HTTP responses.

Common SQLAlchemy exceptions:

IntegrityError: Unique constraint, foreign key, or NOT NULL violation
OperationalError: Connection lost, table does not exist, timeout
DataError: Value too long, invalid type
NoResultFound: scalar_one() with no matching row
MultipleResultsFound: scalar_one() with more than one match

Strategies:

Try/except in endpoints: Catch specific errors and raise HTTPException
Exception handlers: Global handler for all IntegrityError occurrences
Dependency rollback: Auto-rollback in the yield dependency

Pseudo Code

# Strategy 1: try/except in endpoint
try:
    db.add(item)
    db.commit()
except IntegrityError:
    db.rollback()
    raise HTTPException(409, "Duplicate entry")

# Strategy 2: global exception handler
@app.exception_handler(IntegrityError)
def handle_integrity_error(request, exc):
    return JSONResponse(status_code=409, ...)

# Strategy 3: dependency auto-rollback
def get_db():
    try: yield db; db.commit()
    except: db.rollback(); raise
    finally: db.close()

Code

from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.responses import JSONResponse
from sqlalchemy import create_engine, String
from sqlalchemy.exc import IntegrityError, OperationalError
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
)
from typing import Annotated

engine = create_engine("sqlite:///./app.db",
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)

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

DbDep = Annotated[Session, Depends(get_db)]

app = FastAPI()

# --- Strategy 1: Catch in endpoint ---
@app.post("/users")
def create_user(email: str, db: DbDep):
    user = User(email=email)
    db.add(user)
    try:
        db.commit()
        db.refresh(user)
    except IntegrityError:
        db.rollback()
        raise HTTPException(
            status_code=409,
            detail=f"A user with email '{email}' already exists",
        )
    return {"id": user.id, "email": user.email}

# --- Strategy 2: Global exception handler ---
@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError):
    return JSONResponse(
        status_code=409,
        content={"detail": "Database integrity error: duplicate or invalid reference"},
    )

@app.exception_handler(OperationalError)
async def operational_error_handler(request: Request, exc: OperationalError):
    return JSONResponse(
        status_code=503,
        content={"detail": "Database is temporarily unavailable"},
    )

# --- Strategy 3: Auto-rollback dependency ---
def get_db_safe():
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except IntegrityError as e:
        db.rollback()
        # Extract useful info from the original error
        raise HTTPException(status_code=409, detail=str(e.orig))
    except OperationalError:
        db.rollback()
        raise HTTPException(status_code=503, detail="Database unavailable")
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

# Demonstrating rollback behavior
print("Rollback ensures the session is clean after an error")
print("Uncommitted changes are discarded")

Rollback ensures the session is clean after an error
Uncommitted changes are discarded

Key Points

•Always rollback before raising an exception -- the session is in a broken state after an error
•IntegrityError covers unique violations, FK violations, and NOT NULL violations
•Global exception handlers catch errors that escape the endpoint
•The yield dependency pattern centralizes error handling in one place

Gotchas

•After an IntegrityError, the session MUST be rolled back before any further operations
•Global exception handlers catch raw SQLAlchemy errors -- the session is not rolled back automatically
•str(e.orig) gives the underlying DB error message, which may leak schema details to the client
•OperationalError on connection loss may not be recoverable -- consider pool_pre_ping=True

Catch IntegrityError for constraint violations and OperationalError for connection issues. Always call db.rollback() before raising an HTTPException. Use global exception handlers or yield-dependency error handling.

•IntegrityError = constraint violation; OperationalError = connection issue

from sqlalchemy.exc import IntegrityError

@app.post("/users")
def create_user(email: str, db: DbDep):
    user = User(email=email)
    db.add(user)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(status_code=409, detail="Email already exists")
    return {"id": user.id}

Session must be rolled back after any error before further DB operations

Question 13 / 15

How do you implement relationship loading (eager vs lazy)?

Relationship loading determines when and how SQLAlchemy fetches related objects. The wrong strategy causes the N+1 query problem: 1 query to load N parents, then N queries to load each parent's children.

Loading strategies:

Lazy loading (default): Related objects loaded on first access. Simple but causes N+1
Eager loading: Related objects loaded in the initial query:
- joinedload(): Uses SQL JOIN. Good for one-to-one and many-to-one
- selectinload(): Uses a second SELECT with IN clause. Good for one-to-many and many-to-many
- subqueryload(): Uses a subquery. Similar to selectinload but different SQL
Explicit loading: raiseload() raises an error on access (forces explicit loading)
No loading: noload() sets the attribute to None/empty

Async note:

Lazy loading does not work with async SQLAlchemy because it triggers synchronous I/O. You must use eager loading or explicit await session.execute() with joined/select loads.

Pseudo Code

# N+1 problem (BAD)
users = db.query(User).all()  # 1 query
for u in users:
    print(u.posts)  # N queries (one per user)

# Eager loading (GOOD)
users = db.query(User).options(
    selectinload(User.posts)  # 2 queries total
).all()

Code

from sqlalchemy import create_engine, String, ForeignKey, select
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship,
    Session, sessionmaker, joinedload, selectinload, raiseload,
)

engine = create_engine("sqlite:///./app.db", echo=True,
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase): pass

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    books: Mapped[list["Book"]] = relationship(back_populates="author")

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    author: Mapped["Author"] = relationship(back_populates="books")

Base.metadata.create_all(engine)

# Seed some data
with SessionLocal() as db:
    a1 = Author(name="Author 1", books=[Book(title="Book A"), Book(title="Book B")])
    a2 = Author(name="Author 2", books=[Book(title="Book C")])
    db.add_all([a1, a2])
    db.commit()

# --- Lazy loading (default) -- causes N+1 ---
with SessionLocal() as db:
    authors = db.execute(select(Author)).scalars().all()  # 1 query
    for author in authors:
        print(f"{author.name}: {[b.title for b in author.books]}")  # N queries

# --- joinedload -- single JOIN query ---
with SessionLocal() as db:
    stmt = select(Author).options(joinedload(Author.books))
    authors = db.execute(stmt).unique().scalars().all()  # 1 query with JOIN
    for author in authors:
        print(f"{author.name}: {[b.title for b in author.books]}")

# --- selectinload -- 2 queries (best for collections) ---
with SessionLocal() as db:
    stmt = select(Author).options(selectinload(Author.books))
    authors = db.execute(stmt).scalars().all()  # 2 queries
    for author in authors:
        print(f"{author.name}: {[b.title for b in author.books]}")

# --- raiseload -- prevents accidental lazy loading ---
with SessionLocal() as db:
    stmt = select(Author).options(raiseload(Author.books))
    author = db.execute(stmt).scalars().first()
    # author.books  # Would raise: lazy load not permitted

print("Loading strategies demonstrated")

Author 1: ['Book A', 'Book B']
Author 2: ['Book C']
Author 1: ['Book A', 'Book B']
Author 2: ['Book C']
Author 1: ['Book A', 'Book B']
Author 2: ['Book C']
Loading strategies demonstrated

Key Points

•Default lazy loading causes N+1 queries -- one per related object access
•joinedload() uses a JOIN (best for to-one relationships)
•selectinload() uses a separate IN query (best for to-many relationships)
•raiseload() prevents accidental lazy loads by raising an error

Gotchas

•joinedload() with collections requires .unique() to deduplicate results from the JOIN
•Async SQLAlchemy does not support lazy loading at all -- you must eagerly load or use await
•Eager loading in the relationship definition (lazy='selectin') applies globally, which may over-fetch
•Nested eager loading: use selectinload(Author.books).selectinload(Book.reviews) for deep relations

Default lazy loading causes N+1 queries. Use joinedload() for to-one and selectinload() for to-many relationships. Async SQLAlchemy requires eager loading -- lazy loading raises errors.

•selectinload for collections, joinedload for single objects

from sqlalchemy.orm import selectinload, joinedload

# Eager load books when fetching authors (2 queries instead of N+1)
stmt = select(Author).options(selectinload(Author.books))
authors = db.execute(stmt).scalars().all()

# JOIN load for to-one (single query)
stmt = select(Book).options(joinedload(Book.author))
books = db.execute(stmt).unique().scalars().all()

joinedload on collections requires .unique() to avoid duplicate parent rows

Question 14 / 15

How do you set up multiple database connections?

Some applications need multiple databases: a primary for writes, a read replica for queries, or separate databases for different domains (e.g., users DB and analytics DB). FastAPI's dependency injection makes this clean.

Common scenarios:

Primary + read replica: Write to primary, read from replica
Multi-tenant: Each tenant gets their own database
Domain separation: Different services with their own databases

Implementation:

Create separate engines and session factories for each database
Create separate dependencies for each
Use different DeclarativeBase subclasses if schemas differ
Inject the appropriate session via Depends()

Pseudo Code

primary_engine = create_engine(PRIMARY_URL)
replica_engine = create_engine(REPLICA_URL)

PrimarySession = sessionmaker(bind=primary_engine)
ReplicaSession = sessionmaker(bind=replica_engine)

def get_primary_db(): yield PrimarySession()
def get_replica_db(): yield ReplicaSession()

@app.post("/users")
def create(db: Depends(get_primary_db)): ...  # writes

@app.get("/users")
def list(db: Depends(get_replica_db)): ...  # reads

Code

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
)
from typing import Annotated

# --- Primary database (writes) ---
primary_engine = create_engine(
    "sqlite:///./primary.db",
    connect_args={"check_same_thread": False},
)
PrimarySession = sessionmaker(bind=primary_engine)

# --- Read replica (reads) ---
replica_engine = create_engine(
    "sqlite:///./replica.db",
    connect_args={"check_same_thread": False},
)
ReplicaSession = sessionmaker(bind=replica_engine)

# --- Analytics database (separate schema) ---
analytics_engine = create_engine(
    "sqlite:///./analytics.db",
    connect_args={"check_same_thread": False},
)
AnalyticsSession = sessionmaker(bind=analytics_engine)

# Shared base for primary/replica (same schema)
class Base(DeclarativeBase): pass

# Separate base for analytics
class AnalyticsBase(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

class PageView(AnalyticsBase):
    __tablename__ = "page_views"
    id: Mapped[int] = mapped_column(primary_key=True)
    path: Mapped[str] = mapped_column(String(500))
    count: Mapped[int] = mapped_column(default=0)

# --- Dependencies ---
def get_primary_db():
    db = PrimarySession()
    try:
        yield db
    finally:
        db.close()

def get_replica_db():
    db = ReplicaSession()
    try:
        yield db
    finally:
        db.close()

def get_analytics_db():
    db = AnalyticsSession()
    try:
        yield db
    finally:
        db.close()

PrimaryDep = Annotated[Session, Depends(get_primary_db)]
ReplicaDep = Annotated[Session, Depends(get_replica_db)]
AnalyticsDep = Annotated[Session, Depends(get_analytics_db)]

app = FastAPI()

# Writes go to primary
@app.post("/users")
def create_user(name: str, db: PrimaryDep):
    user = User(name=name)
    db.add(user)
    db.commit()
    db.refresh(user)
    return {"id": user.id, "name": user.name}

# Reads go to replica
@app.get("/users")
def list_users(db: ReplicaDep):
    return db.execute(select(User)).scalars().all()

# Analytics uses a separate database
@app.get("/analytics/top-pages")
def top_pages(db: AnalyticsDep):
    stmt = select(PageView).order_by(PageView.count.desc()).limit(10)
    return db.execute(stmt).scalars().all()

Key Points

•Create separate engines and session factories for each database
•Use Annotated type aliases (PrimaryDep, ReplicaDep) for clean injection
•Different databases can use different DeclarativeBase subclasses
•The dependency determines which database an endpoint uses

Gotchas

•Read replicas have replication lag -- recently written data may not appear immediately
•Cross-database joins are not possible with separate engines
•Each engine has its own connection pool -- total connections = sum of all pool sizes
•Alembic needs separate configurations (multiple alembic.ini or env.py branches) for different schemas

Create separate engines, session factories, and yield dependencies for each database. Inject the appropriate one via Depends(). Use Annotated type aliases to keep endpoint signatures clean.

•Separate engines and dependencies for each database
•Annotated aliases keep endpoint code clean

primary_engine = create_engine("postgresql://host/primary")
replica_engine = create_engine("postgresql://host/replica")

PrimarySession = sessionmaker(bind=primary_engine)
ReplicaSession = sessionmaker(bind=replica_engine)

def get_primary_db():
    db = PrimarySession()
    try: yield db
    finally: db.close()

def get_replica_db():
    db = ReplicaSession()
    try: yield db
    finally: db.close()

PrimaryDep = Annotated[Session, Depends(get_primary_db)]
ReplicaDep = Annotated[Session, Depends(get_replica_db)]

Read replicas have replication lag -- newly written data may not be immediately visible

Question 15 / 15

How do you implement soft delete patterns?

Soft delete marks records as deleted instead of removing them from the database. This preserves data for auditing, recovery, and referential integrity.

Implementation:

Add a deleted_at timestamp column (NULL = active, non-NULL = deleted)
Filter out deleted records in all normal queries
Provide endpoints to soft-delete and optionally restore

Approaches:

Manual filtering: Add .where(Model.deleted_at.is_(None)) to every query
SQLAlchemy events: Use @event.listens_for to auto-filter
Hybrid property: Add an is_deleted property to the model
Custom query class: Override the default query to auto-exclude deleted rows

The manual + helper method approach is recommended because it is explicit and avoids hidden behavior.

Pseudo Code

class SoftDeleteMixin:
    deleted_at: Mapped[Optional[datetime]]

    def soft_delete(self):
        self.deleted_at = datetime.utcnow()

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

# Filter active records
stmt = select(Item).where(Item.deleted_at.is_(None))

Code

from datetime import datetime, timezone
from typing import Optional, Annotated
from fastapi import FastAPI, Depends, HTTPException, Query
from sqlalchemy import create_engine, String, select, event
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
)

engine = create_engine("sqlite:///./app.db",
                       connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase): pass

# --- Soft delete mixin ---
class SoftDeleteMixin:
    deleted_at: Mapped[Optional[datetime]] = mapped_column(default=None)

    def soft_delete(self) -> None:
        self.deleted_at = datetime.now(timezone.utc)

    def restore(self) -> None:
        self.deleted_at = None

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

# --- Model using the mixin ---
class Article(SoftDeleteMixin, Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column(String(5000))

Base.metadata.create_all(engine)

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

DbDep = Annotated[Session, Depends(get_db)]

app = FastAPI()

# Helper: base query that excludes soft-deleted records
def active_articles():
    return select(Article).where(Article.deleted_at.is_(None))

# List only active articles
@app.get("/articles")
def list_articles(
    db: DbDep,
    include_deleted: Annotated[bool, Query()] = False,
):
    if include_deleted:
        stmt = select(Article)
    else:
        stmt = active_articles()
    return db.execute(stmt).scalars().all()

# Get a single article (only if active)
@app.get("/articles/{article_id}")
def get_article(article_id: int, db: DbDep):
    article = db.get(Article, article_id)
    if not article or article.is_deleted:
        raise HTTPException(status_code=404, detail="Article not found")
    return article

# Soft delete
@app.delete("/articles/{article_id}", status_code=204)
def delete_article(article_id: int, db: DbDep):
    article = db.get(Article, article_id)
    if not article or article.is_deleted:
        raise HTTPException(status_code=404, detail="Article not found")
    article.soft_delete()
    db.commit()

# Restore
@app.post("/articles/{article_id}/restore")
def restore_article(article_id: int, db: DbDep):
    article = db.get(Article, article_id)
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")
    if not article.is_deleted:
        raise HTTPException(status_code=400, detail="Article is not deleted")
    article.restore()
    db.commit()
    return {"id": article.id, "title": article.title, "restored": True}

# Hard delete (admin only -- permanently removes the record)
@app.delete("/admin/articles/{article_id}/purge", status_code=204)
def purge_article(article_id: int, db: DbDep):
    article = db.get(Article, article_id)
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")
    db.delete(article)
    db.commit()

# Demonstrate
with SessionLocal() as db:
    a = Article(title="Test", body="Content")
    db.add(a)
    db.commit()
    print(f"Active: is_deleted={a.is_deleted}")
    a.soft_delete()
    db.commit()
    print(f"Soft deleted: is_deleted={a.is_deleted}")
    a.restore()
    db.commit()
    print(f"Restored: is_deleted={a.is_deleted}")

Active: is_deleted=False
Soft deleted: is_deleted=True
Restored: is_deleted=False

Key Points

•Use a deleted_at timestamp (NULL = active) rather than a boolean flag for auditability
•Create a mixin class with soft_delete(), restore(), and is_deleted for reuse across models
•Filter with .where(Model.deleted_at.is_(None)) in all normal queries
•Provide separate endpoints for soft delete, restore, and hard delete (purge)

Gotchas

•Unique constraints still apply to soft-deleted rows -- consider partial unique indexes
•Foreign keys to soft-deleted rows remain valid, which can cause unexpected joins
•Forgetting to filter deleted rows in even one query leaks deleted data
•Soft-deleted rows accumulate over time -- schedule periodic hard-deletes or archiving

Add a deleted_at column (NULL = active). Create a mixin with soft_delete() and restore() methods. Filter queries with .where(Model.deleted_at.is_(None)) to exclude deleted records.

•deleted_at timestamp for auditability, not just a boolean

class SoftDeleteMixin:
    deleted_at: Mapped[Optional[datetime]] = mapped_column(default=None)

    def soft_delete(self):
        self.deleted_at = datetime.now(timezone.utc)

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

class Article(SoftDeleteMixin, Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))

# Always filter in queries
stmt = select(Article).where(Article.deleted_at.is_(None))

Unique constraints still apply to soft-deleted rows -- use partial indexes if needed

Built by onllm.dev

Home GitHub Support awesomePrep