Chapter Outline

Chapter 11: Working With Relational Databases

In this chapter, you will learn:

How to use SQLAlchemy with FastAPI for relational data.
How to manage schema changes with Alembic (FastAPI) and migrations.
How to implement CRUD operations that persist to a database.
How to write tests for database-backed APIs.

By the end, our running projects will have:

FastAPI Todo API → switched from in-memory to SQLite with SQLAlchemy.
Django Blog App → persisting posts in DB (already implemented in Ch.9), expanded with queries.

11.1 FastAPI With SQLAlchemy

Let's replace our in-memmory database for our todo app, with an actual SQLite database. For that we're going to use SQLAlchemy.

11.1.1 Install Dependencies

bash
poetry add sqlalchemy aiosqlite alembic

11.1.2 Configure Database Connection

app/core/db.py
1from sqlalchemy import create_engine
2from sqlalchemy.orm import sessionmaker, declarative_base
3
4SQLALCHEMY_DATABASE_URL = "sqlite:///./db.sqlite3"
5
6engine = create_engine(
7    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
8)
9SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
10
11Base = declarative_base()

The engine is SQLAlchemy’s interface to the database. create_engine() is used to connect to the SQLite database db.sqlite3 located at the project root. The database will be created the first time it is accessed.

FastAPI uses asynchronous code and dependency injection, so multiple threads may touch the DB. However, SQLite limits access to the database to a single thread that created the connection. Therefore we set "check_same_thread": False as the connection arguments, so FastAPI can safely use SQLite. For PostgreSQL/MySQL, this would not be needed.

SessionLocal is a factory, not an actual session. A Session handles database conversations (unit of work). We will see later how it is used when FastAPI handles a request.

Finally, declarative_base() creates a base class used to define SQLAlchemy ORM models. We'll examine an ORM model next.

11.1.3 Define the App Models

Let's add a SQL model to our app for the todos. This model sould have all the fields found in TodoItemEntity. Every model must inherit from Base.

app/todos/models.py
1from sqlalchemy import Column, Integer, String, Boolean
2from app.core.db import Base
3
4class TodoBaseModel(Base):
5    __tablename__ = "todos"
6
7    id = Column(Integer, primary_key=True, index=True)
8    title = Column(String, index=True)
9    completed = Column(Boolean, default=False)

The model is represented by the table todos. The id field is being declared as the primary key for the table. title is declared as a string, and completed as a boolean.

11.1.4 Replace In-memory DB

Let's replace the in-memory database with the SQLite database.

app/core/db.py
1from sqlalchemy import create_engine
2from sqlalchemy.orm import sessionmaker, declarative_base, Session
3
4SQLALCHEMY_DATABASE_URL = "sqlite:///./db.sqlite3"
5
6engine = create_engine(
7    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
8)
9SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
10
11Base = declarative_base()
12
13# Create tables
14Base.metadata.create_all(bind=engine)
15
16def get_db():
17    """Dependency to provide DB session."""
18    db = SessionLocal()
19    try:
20        yield db
21    finally:
22        db.close()

11.1.5 Update Repository CRUD Operations

Let's update the TodoRepository so that it can query the SQL database for data, instead of the mock in-memory database.

app/todos/repository.py
1from typing import Protocol, Iterable
2from sqlalchemy.orm import Session
3
4from app.todos.models import TodoBaseModel
5
6
7class TodoRepository(TodoRepositoryProtocol):
8    def __init__(self, db: Session):
9        self.db = db
10
11    def list_todos(self):
12        """Return all todos."""
13        return self.db.query(TodoBaseModel).all()
14
15    def create_todo(self, title: str, completed: bool) -> TodoBaseModel :
16        """Adds a new TodoItem to the database."""
17        new_todo = TodoBaseModel (title=title, completed=completed)
18        self.db.add(new_todo)
19        self.db.commit()
20        self.db.refresh(new_todo)
21        return new_todo
22
23    def get_todo(self, id: int):
24        """Retrieve a Todo item by ID."""
25        found_todo = self.db.get(TodoBaseModel, id)
26        return found_todo
27
28    def update_todo(self, id: int, title: str, completed: bool) -> TodoBaseModel :
29        """Update a Todo item by ID."""
30        found_todo = self.get_todo(id)
31        found_todo.title = title
32        found_todo.completed = completed
33        self.db.add(found_todo)
34        self.db.commit()
35        self.db.refresh(found_todo)
36        return found_todo
37
38    def delete_todo(self, id: int) -> TodoBaseModel :
39        """Deletes an item by ID."""
40        found_todo = self.get_todo(id)
41        self.db.delete(found_todo)
42        self.db.commit()
43        self.db.refresh(found_todo)
44        return found_todo

The repository methods demonstrate four common SQLAlchemy Queries (ORM Style).

1. Select all:

python
found_todos = self.db.query(TodoBaseModel).all()

2. Get by primary key:

python
found_todo = self.db.get(TodoBaseModel, id)

3. Insert:

python
new_todo = TodoBaseModel (title=title, completed=completed)
self.db.add(new_todo)
self.db.commit()
self.db.refresh(new_todo)

4. Update

python
found_todo = self.db.get(TodoBaseModel, id)
found_todo.title = title
self.db.add(found_todo)
self.db.commit()

5. Delete

python
found_todo = self.get_todo(id)
self.db.delete(found_todo)
self.db.commit()

11.2 Alembic for DB Migration

We'll use Alembic for database migrations. This includes database setup and any schema changes.

11.2.1 Alembic Setup

Let's initialize Alembic for our project by running the following:

bash
poetry run alembic init alembic

11.2.2 Update Migration Setup

Now that Alembic has been setup, you will find the following directory setup along with the rest of our source:

bash
fastapi-todo/
├── alembic.ini
├── alembic/
|   ├── env.py  
|   ├── README  
|   ├── script.py.mako  
│   └── versions/
|        └── ... # This is where the migration script live

We need to update the configration files that is generated to ensure that Alembic uses our database and models correctly.

alembic.ini
# Update DB URL
sqlalchemy.url = sqlite:///./db.sqlite3

Now let's ensure that our app models are imported and used correctly by Alembic.

alembic/env.py
1from logging.config import fileConfig
2
3from sqlalchemy import engine_from_config
4from sqlalchemy import pool
5from app.core.db import engine, Base
6
7from alembic import context
8
9# this is the Alembic Config object, which provides
10# access to the values within the .ini file in use.
11config = context.config
12
13# Interpret the config file for Python logging.
14# This line sets up loggers basically.
15if config.config_file_name is not None:
16    fileConfig(config.config_file_name)
17
18# add your model's MetaData object here
19# for 'autogenerate' support
20from app.todos import models
21target_metadata = Base.metadata
22
23# other values from the config, defined by the needs of env.py,
24# can be acquired:
25# my_important_option = config.get_main_option("my_important_option")
26# ... etc.
27
28
29def run_migrations_offline() -> None:
30    """Run migrations in 'offline' mode.
31
32    This configures the context with just a URL
33    and not an Engine, though an Engine is acceptable
34    here as well.  By skipping the Engine creation
35    we don't even need a DBAPI to be available.
36
37    Calls to context.execute() here emit the given string to the
38    script output.
39
40    """
41    url = config.get_main_option("sqlalchemy.url")
42    context.configure(
43        url=url,
44        target_metadata=target_metadata,
45        literal_binds=True,
46        dialect_opts={"paramstyle": "named"},
47    )
48
49    with context.begin_transaction():
50        context.run_migrations()
51
52
53def run_migrations_online() -> None:
54    """Run migrations in 'online' mode.
55
56    In this scenario we need to create an Engine
57    and associate a connection with the context.
58
59    """
60    connectable = engine
61
62    with connectable.connect() as connection:
63        context.configure(
64            connection=connection, target_metadata=target_metadata
65        )
66
67        with context.begin_transaction():
68            context.run_migrations()
69
70
71if context.is_offline_mode():
72    run_migrations_offline()
73else:
74    run_migrations_online()

11.2.3 Run Migration

Let's generate our first migration script, and run it with the upgrade head command.

bash
poetry run alembic revision --autogenerate -m "init"
poetry run alembic upgrade head

All subsequent migrations are run like a git commit. upgrade head is used to apply the changes into the database.

bash
poetry run alembic revision -m "Add a column"
poetry run alembic upgrade head

This should create the databse with the new todos table.

11.3 Testing Database Operations

We are going to use a mock database for our tests. The mock DB is going to be file based.

11.3.1 Test Utils

Let's create some utility functions to setup the mock database, and the test client that uses it. This is done so that the same client can be used across all test suites:

tests/test_utils.py
1from sqlalchemy import create_engine
2from sqlalchemy.orm import sessionmaker
3from fastapi.testclient import TestClient
4
5from app.core.db import Base, get_db
6from app.main import app
7
8TEST_DB_URL = "sqlite:///./test.db"
9
10engine = create_engine(
11    TEST_DB_URL,
12    connect_args={"check_same_thread": False},
13)
14
15TestingSessionLocal = sessionmaker(
16    autocommit=False, autoflush=False, bind=engine
17)
18
19def override_get_db():
20    db = TestingSessionLocal()
21    try:
22        yield db
23    finally:
24        db.close()
25
26def create_test_app():
27    """Factory to return a fresh TestClient with DB overrides."""
28    Base.metadata.drop_all(bind=engine)
29    Base.metadata.create_all(bind=engine)
30
31    # This swaps the get_db to the override
32    app.dependency_overrides[get_db] = override_get_db
33    return TestClient(app)

11.3.2 Test Fixtures

We can now declare the test client and to create test fixtures which are injected into our tests by pytest.

tests/conftest.py
1import pytest
2from tests.test_utils import create_test_app
3
4@pytest.fixture(scope="session")
5def client():
6    client = create_test_app()
7    return client

11.3.3 Using Test Fixtures

Now the new client with the test database can be used for all tests as follows:

tests/conftest.py
1import pytest
2
3from ..conftest import client, auth_token
4
5def test_create_todo(client, auth_token) -> None:
6    """Test the addition of a todo item."""
7    headers = {"Authorization": f"Bearer {auth_token}"}
8    response = client.post(
9      "/api/todos",
10      json={"title": "Learn FastAPI"}, headers=headers
11    )
12    assert response.status_code == 200
13    data = response.json()
14    assert data["title"] == "Learn FastAPI"
15    assert data["completed"] is False

11.4 SQLAlchemy & FastAPI Flow

flowchart TD R[Request /todos] --> V[FastAPI Route] V --> D[SessionLocal] D --> M[SQLAlchemy Model] M --> DB[(SQLite/Postgres)] DB --> M M --> V V --> Rsp[JSON Response]

11.5 Chapter Assignment

Create a SQLAlchemy model for the users app.
Update the UserRepository to use the new model.
Seed the users table with data from the existing mock DB.
Test login.

11.6 Further Reading

Check your understanding

Test your knowledge of Working with Relational Databases

Python

Web Dev