Lesson 14 · FastAPI deep dive

FastAPI Tutorial #14: Databases with SQLAlchemy

Video fourteen of the eighteen-part series: persisting real data instead of an in-memory dict. Models, sessions, dependency-injected database access, and a…

⬇ Download notebook

CourseFastAPI deep dive
Lesson14 of 18
FormatJupyter notebook · 10 code cells

SQLAlchemyFastAPI Pydantic

What you'll learn

engine and declarativebase
Defining a Model - a Real Table as a Real Class
createall - Actually Building the Tables
sessionmaker - the Unit of Work
Create - add, commit, refresh
Read - Querying Rows Back
Update - Modify and Commit Again
Delete - Remove and Commit

Data

No separate download needed — the notebook creates or downloads everything it uses.

📓 Full notebook

Download .ipynb

FastAPI Deep-Dive, Video 14: Databases with SQLAlchemy#

Video fourteen of the eighteen-part series: persisting real data instead of an in-memory dict.
Models, sessions, dependency-injected database access, and a real full CRUD flow.
Let's get into it.

Part 1: engine and declarative_base#

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy.pool import StaticPool
engine = create_engine(
    'sqlite:///:memory:',
    connect_args={'check_same_thread': False},
    poolclass=StaticPool,
)
Base = declarative_base()
print(type(engine))

<class 'sqlalchemy.engine.base.Engine'>

Part 2: Defining a Model - a Real Table as a Real Class#

from sqlalchemy import Column, Integer, String, Float
class ProductModel(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    price = Column(Float, nullable=False)
print(ProductModel.__tablename__)
print([c.name for c in ProductModel.__table__.columns])

products
['id', 'name', 'price']

Part 3: create_all - Actually Building the Tables#

Base.metadata.create_all(bind=engine)
from sqlalchemy import inspect
inspector = inspect(engine)
print(inspector.get_table_names())

['products']

Part 4: sessionmaker - the Unit of Work#

from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(bind=engine)
db = SessionLocal()
print(type(db))

<class 'sqlalchemy.orm.session.Session'>

Part 5: Create - add, commit, refresh#

mug = ProductModel(name='Mug', price=9.99)
print(mug.id)
db.add(mug)
db.commit()
db.refresh(mug)
print(mug.id)

None
1

Part 6: Read - Querying Rows Back#

plate = ProductModel(name='Plate', price=6.5)
db.add(plate)
db.commit()
print(db.get(ProductModel, mug.id).name)
print([p.name for p in db.query(ProductModel).all()])
print([p.name for p in db.query(ProductModel).filter(ProductModel.price < 8).all()])

Mug
['Mug', 'Plate']
['Plate']

Part 7: Update - Modify and Commit Again#

tracked_mug = db.get(ProductModel, mug.id)
tracked_mug.price = 11.99
db.commit()
db.refresh(tracked_mug)
print(tracked_mug.price)

11.99

Part 8: Delete - Remove and Commit#

db.delete(plate)
db.commit()
print(db.get(ProductModel, plate.id))
print(db.query(ProductModel).count())

None
1

Part 9: get_db - a Dependency-Injected Session#

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
def get_db():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()
app = FastAPI()
@app.get('/products/count')
def count_products(db: Session = Depends(get_db)):
    return {'count': db.query(ProductModel).count()}
from fastapi.testclient import TestClient
client = TestClient(app)
print(client.get('/products/count').json())

{'count': 1}

Part 10: A Real Pattern - a Small, Complete CRUD API#

from pydantic import BaseModel, ConfigDict
crud_engine = create_engine('sqlite:///:memory:', connect_args={'check_same_thread': False}, poolclass=StaticPool)
CrudBase = declarative_base()
class TaskModel(CrudBase):
    __tablename__ = 'tasks'
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    done = Column(Integer, default=0)
CrudBase.metadata.create_all(bind=crud_engine)
CrudSessionLocal = sessionmaker(bind=crud_engine)
def get_crud_db():
    session = CrudSessionLocal()
    try:
        yield session
    finally:
        session.close()
class TaskOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    done: int
crud_app = FastAPI()
@crud_app.post('/tasks', response_model=TaskOut)
def create_task(title: str, db: Session = Depends(get_crud_db)):
    task = TaskModel(title=title)
    db.add(task)
    db.commit()
    db.refresh(task)
    return task
@crud_app.get('/tasks/{task_id}', response_model=TaskOut)
def get_task(task_id: int, db: Session = Depends(get_crud_db)):
    task = db.get(TaskModel, task_id)
    if not task:
        raise HTTPException(status_code=404, detail='task not found')
    return task
@crud_app.delete('/tasks/{task_id}')
def delete_task(task_id: int, db: Session = Depends(get_crud_db)):
    task = db.get(TaskModel, task_id)
    if not task:
        raise HTTPException(status_code=404, detail='task not found')
    db.delete(task)
    db.commit()
    return {'deleted': task_id}
crud_client = TestClient(crud_app)
created = crud_client.post('/tasks', params={'title': 'write lesson'}).json()
print(created)
print(crud_client.get(f'/tasks/{created["id"]}').json())
print(crud_client.delete(f'/tasks/{created["id"]}').json())
print(crud_client.get(f'/tasks/{created["id"]}').status_code)

{'id': 1, 'title': 'write lesson', 'done': 0}
{'id': 1, 'title': 'write lesson', 'done': 0}
{'deleted': 1}
404

Wrap-Up: What You Learned#

create_engine represents the connection; declarative_base is the base every model inherits from.
A model class declares tablename and Column fields, mapping a Python class to a table.
Base.metadata.create_all actually issues the CREATE TABLE statements for every defined model.
sessionmaker builds a factory for Session objects, each tracking its own pending changes.
db.add stages an object, db.commit writes it, and db.refresh reloads database-generated values.
db.get fetches by primary key directly; db.query with filter narrows a broader search.
A tracked object's attributes can be changed directly in Python, then written back with commit.
db.delete stages a removal exactly like add stages an addition, until commit makes it permanent.
A yield-based get_db dependency opens one session per request and always closes it afterward.
That wraps up databases with SQLAlchemy. Next up: Async Endpoints and Background Tasks.

Found this useful?

All lessons, notebooks and datasets here are free. If they helped you, a coffee keeps new lessons coming.

☕ Buy me a coffee Other ways to help