Transactions and Migrations

Transactions and migrations solve two different but connected problems in a FastAPI application. A transaction makes one application operation atomic: either every database change for the operation is committed, or none of it is. A migration changes the database structure in a reviewed, repeatable way so the code and schema keep matching as the service evolves.

In this course section, the practical outcome is a database layer where request handlers do not scatter commits, rollback happens predictably, and schema changes are captured as source-controlled Alembic revisions. That matters because FastAPI can handle many requests concurrently. Without clear transaction ownership, two endpoints can leave partial records behind or run against different assumptions about the same table.

How Database Transactions Work

A relational database transaction groups statements on a single connection. The database keeps the changes private until COMMIT. If the application sends ROLLBACK, or the connection fails before committing, those changes are discarded. SQLAlchemy exposes this through a Session or AsyncSession. The session tracks Python objects, translates changes into SQL, flushes statements when needed, and manages the transaction boundary.

The important terms are precise. flush sends pending SQL to the database but does not make it durable. commit flushes, ends the transaction, and makes the changes visible according to the database isolation rules. rollback ends the transaction and discards uncommitted changes. refresh reloads server-generated values such as IDs or timestamps into an object. isolation level controls how much one transaction can see of another transaction’s work.

FastAPI dependencies are a natural place to own the session lifecycle. The endpoint asks for a session, service code performs work, and the dependency commits or rolls back at the boundary. This keeps transaction policy in one place. It also makes tests easier because test code can override the dependency and wrap each test in a rollback.

Session Dependency Anatomy

A common async setup creates one engine for the process, one session factory, and one session per request. The dependency yields the session to the endpoint. Code after yield runs after the endpoint returns or raises, so it can decide whether to commit or rollback.

from collections.abc import AsyncIterator

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import Integer, String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

DATABASE_URL = "sqlite+aiosqlite:///./shop.db"

engine = create_async_engine(DATABASE_URL)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = "accounts"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
    balance_cents: Mapped[int] = mapped_column(Integer, default=0)

class AccountIn(BaseModel):
    email: str

async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

app = FastAPI()

@app.post("/accounts", status_code=201)
async def create_account(payload: AccountIn, session: AsyncSession = Depends(get_session)):
    account = Account(email=payload.email, balance_cents=0)
    session.add(account)
    await session.flush()
    return {"id": account.id, "email": account.email, "balance_cents": account.balance_cents}

Expected behavior is deterministic after the table exists. A valid POST /accounts inserts one row and returns its generated ID. If the unique email constraint fails, SQLAlchemy raises during flush or commit, the dependency rolls back, and no half-created account remains. The endpoint uses flush because it needs the generated ID before returning; it does not call commit because the dependency owns that decision.

Worked Example: Atomic Transfer

The next example shows why one transaction often spans several statements. A transfer must debit one account and credit another. Committing after the debit would be a bug because a later failure could lose money. The service function locks the relevant rows, checks the invariant, mutates both objects, and lets the outer transaction commit.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def transfer_cents(session: AsyncSession, from_id: int, to_id: int, amount_cents: int) -> dict[str, int]:
    if amount_cents <= 0:
        raise ValueError("transfer amount must be positive")

    result = await session.execute(
        select(Account).where(Account.id.in_([from_id, to_id])).with_for_update()
    )
    accounts = {account.id: account for account in result.scalars()}

    source = accounts.get(from_id)
    target = accounts.get(to_id)
    if source is None or target is None:
        raise LookupError("account not found")
    if source.balance_cents < amount_cents:
        raise ValueError("insufficient funds")

    source.balance_cents -= amount_cents
    target.balance_cents += amount_cents
    await session.flush()
    return {"from_balance_cents": source.balance_cents, "to_balance_cents": target.balance_cents}

If account 1 has 5000 cents and account 2 has 1000 cents, calling transfer_cents(session, 1, 2, 1200) returns {"from_balance_cents": 3800, "to_balance_cents": 2200}. If the source has only 500 cents, the function raises before changing either balance. The with_for_update request asks databases that support row locks to prevent concurrent transfers from reading and updating the same balances at the same time. Some development databases have weaker locking behavior, so concurrency-sensitive code should be tested against the same database family used in production.

Migration Anatomy

Transactions protect data changes at runtime. Migrations protect structural changes over time. Alembic records each schema change as a revision file with identifiers, dependencies, and two functions: upgrade and downgrade. upgrade moves the schema forward. downgrade describes how to reverse the change when reversal is practical.

A migration is not just a dump of model definitions. It is an ordered operation against existing data. Adding a nullable column is usually simple. Adding a non-null column to a populated table requires a default, a backfill, or a multi-step rollout. Renaming a column may be safe for the database but unsafe for running application versions that still read the old name.

"""add account balance

Revision ID: 20260906_add_account_balance
Revises: 20260901_create_accounts
"""

from alembic import op
import sqlalchemy as sa

revision = "20260906_add_account_balance"
down_revision = "20260901_create_accounts"
branch_labels = None
depends_on = None

def upgrade() -> None:
    op.add_column("accounts", sa.Column("balance_cents", sa.Integer(), nullable=True))
    op.execute("UPDATE accounts SET balance_cents = 0 WHERE balance_cents IS NULL")
    op.alter_column("accounts", "balance_cents", nullable=False)

def downgrade() -> None:
    op.drop_column("accounts", "balance_cents")

This revision uses a three-step pattern. First it adds the column as nullable so existing rows remain valid. Then it backfills deterministic data. Finally it tightens the constraint. The expected result is that every existing and future account has a non-null balance. The downgrade drops the column, which is structurally reversible but loses balance data; that trade-off should be explicit in review.

Worked Example: Testing Rollback

Tests should prove that an exception leaves the database unchanged. The example below uses a fake repository so the transaction idea is visible without a live database. In a real integration test, the same shape applies: arrange data, perform an operation that fails, then query again with a fresh session.

class FakeUnitOfWork:
    def __init__(self) -> None:
        self.items: list[str] = []
        self.committed = False
        self.rolled_back = False

    def add(self, value: str) -> None:
        self.items.append(value)

    def commit(self) -> None:
        self.committed = True

    def rollback(self) -> None:
        self.items.clear()
        self.rolled_back = True

def create_two_items(uow: FakeUnitOfWork, fail_after_first: bool) -> None:
    try:
        uow.add("first")
        if fail_after_first:
            raise RuntimeError("simulated failure")
        uow.add("second")
        uow.commit()
    except Exception:
        uow.rollback()
        raise

def test_failure_rolls_back() -> None:
    uow = FakeUnitOfWork()
    try:
        create_two_items(uow, fail_after_first=True)
    except RuntimeError:
        pass
    assert uow.items == []
    assert uow.committed is False
    assert uow.rolled_back is True

The deterministic output is the final set of assertions passing. This is not a replacement for database tests, but it captures the application rule: partial work is not acceptable. Database integration tests then verify that the rule still holds when constraints, flush timing, and driver errors are involved.

Design Choices and Trade-offs

Commit once per request is easy to reason about and works well for ordinary CRUD operations. It becomes less appropriate for long workflows that call external services, stream responses, or hold locks while waiting on slow work. In those cases, move the database transaction around the smallest set of statements that must be atomic, and store an explicit workflow state instead of keeping a transaction open across network calls.

Using ORM models in endpoint code is convenient, but service functions should not depend on HTTP request objects. Keep business operations callable with a session and typed values. That makes background jobs, tests, and command-line maintenance scripts use the same transaction rules as API routes.

Autogenerated Alembic revisions are useful starting points, not finished migrations. Review them for destructive operations, constraint names, index cost, backfill strategy, and compatibility with deployed application versions. A large table backfill may need batching or a background job instead of one migration transaction.

Failure Modes and Troubleshooting

Symptom: an endpoint returns success but related rows are missing. Cause: code committed after the first insert and then swallowed a later exception. Diagnose: search for commit() calls outside the session dependency or unit-of-work boundary, and reproduce with a forced exception after the first write. Correct: move commits to one owner and let exceptions roll back the whole operation.

Symptom: tests pass locally but production reports duplicate or negative balances. Cause: concurrent requests updated rows without locks or database constraints. Diagnose: inspect isolation level, missing FOR UPDATE locks, and absent check constraints. Run two transfer requests at the same time against a production-like database. Correct: lock the rows, add constraints such as non-negative balances, and retry only operations known to be idempotent.

Symptom: deployment fails during migration with a non-null constraint error. Cause: the migration added a required column to a populated table without backfilling existing rows. Diagnose: run the migration against a recent database copy and count rows where the new column would be null. Correct: use the nullable, backfill, alter pattern or deploy the constraint in a later revision.

Symptom: Alembic says the target database is not up to date. Cause: a developer created a new revision before applying existing revisions, or two branches introduced competing heads. Diagnose: run alembic current and alembic heads. Correct: upgrade the database, merge heads intentionally, and avoid editing applied revision IDs.

Reliability and Performance

Transactions improve reliability only when they are short and well scoped. Holding a transaction open while waiting for email, payment, or object storage increases lock time and can block other requests. Prefer writing a durable row that records pending work, committing it, and letting a worker perform the external action with retry and idempotency keys.

Migrations are operational events. Adding an index can read the whole table. Backfilling data can create write pressure. Dropping a column can break an older application instance still running during a rolling deploy. For important databases, test migrations on realistic data volume, record expected duration, and prepare the rollback or forward-fix command before release.

Hands-on Lab

Prerequisites: a FastAPI project using SQLAlchemy async sessions, Alembic configured for the same database URL, and a disposable local database. Step 1: create or identify a table such as accounts. Step 2: implement a request-scoped session dependency that commits after yield and rolls back on exceptions. Step 3: add an endpoint or service operation that writes two related changes and raises a controlled exception between them in a test-only path. Step 4: verify with a fresh session that neither change persisted after the exception. Step 5: create an Alembic revision that adds a nullable column, backfills existing rows, then makes the column non-null. Step 6: run the migration locally and inspect the table definition and row values. Step 7: run the downgrade only on the disposable database and confirm the schema returns to the previous shape.

Verification should include one successful request, one failing request, alembic current showing the expected revision, and a direct query proving the backfilled column has no null values. Cleanup is to downgrade or recreate the disposable database, remove any test-only failure switch before review, and keep the migration file if it represents the intended schema change.

Assessment Exercises

• Given an endpoint that calls commit() three times, identify which failure points can leave partial data and rewrite the transaction boundary.
• Design a migration for adding a required status column to a table with one million existing rows. Explain how you would backfill and verify it.
• Explain why flush can be necessary before returning a response even though commit has not happened yet.
• A transfer endpoint occasionally loses updates under load. List the database and application evidence you would collect before changing the code.
• Decide whether a downgrade should delete a new table that may contain user data. Defend the operational trade-off.

Summary

FastAPI does not make database work atomic by itself. The application must define where sessions begin, where commits happen, and how exceptions trigger rollback. SQLAlchemy provides the transaction machinery, while Alembic provides a history of schema changes that can be reviewed and replayed. Treat runtime transactions and migrations as one persistence discipline: keep operations small, constraints explicit, revisions tested, and ownership unambiguous.