Lesson 4: Relational Persistence

Connection management and pooling

Opening a database connection is expensive: a TCP connection, authentication, and session setup, typically a few milliseconds. Doing it per query is the most common database performance mistake.

# Wrong: a connection per call
async def get_document(doc_id: str) -> Document:
    conn = await asyncpg.connect(DATABASE_URL)
    row = await conn.fetchrow("SELECT * FROM documents WHERE id = $1", doc_id)
    await conn.close()
    return Document(**row)

Under concurrency this is worse than slow. A hundred concurrent requests open a hundred connections, and databases have a hard connection limit, typically in the low hundreds. Exceeding it means new connections are refused, and the failure looks like the database being down.

A connection pool keeps a set of open connections and lends them out.

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine(
    settings.database_url,           # postgresql+asyncpg://...
    pool_size=10,                    # connections kept open
    max_overflow=5,                  # extra allowed under burst
    pool_timeout=5.0,                # wait for a free connection, then fail
    pool_recycle=1800,               # replace connections older than 30 minutes
    pool_pre_ping=True,              # check liveness before handing one out
)

SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

[VOLATILE: SQLAlchemy 2.0.x is the current stable line with 2.1 in beta at time of writing. Verify parameter names and the async API before publishing.]

Each parameter earns its place. pool_size plus max_overflow is your maximum concurrent connections, and multiplied by the number of application instances it must stay under the database limit. pool_timeout is the one people omit, and Module 5 explains why: without it, waiting for a free connection is unbounded, which is a hang that looks nothing like a database problem. pool_recycle avoids connections silently dropped by an idle timeout somewhere in the network path. pool_pre_ping catches a dead connection before your query fails on it.

One engine per process, many short sessions. The engine holds the pool and is created once. A session is a unit of work: create one, use it, close it.

async def get_document(doc_id: str) -> Document | None:
    async with SessionFactory() as session:
        row = await session.get(DocumentRow, doc_id)
        return row.to_document() if row else None

The context manager returns the connection to the pool on exit, including when an exception is raised. A session held open across a long operation holds a pooled connection with it, which starves everything else, so keep sessions short and never store one on a long-lived object.

Pool sizing under async load. Async concurrency makes it easy to launch a thousand simultaneous queries against a pool of ten. Nine hundred and ninety of them wait, and with a pool timeout set they fail rather than hang, which is the correct behaviour and still a failure. The fix is bounded concurrency at the application level, which is the next module's subject, rather than an ever larger pool.

[IMAGE PROMPT M7-3
Purpose: Show how a connection pool serves many concurrent tasks from a small fixed set of connections, and what happens when demand exceeds the pool.
Visual type: Resource allocation diagram with a queue.
Prompt: A clean educational diagram in three vertical zones reading left to right. The left zone shows a column of twelve small boxes labelled "concurrent tasks". The middle zone shows a rounded container labelled "connection pool (size 10, overflow 5)" containing ten small connection tokens, three of which are marked "idle" and seven marked "in use", plus a small waiting queue drawn beneath the container labelled "waiting for a free connection" holding two task markers, with a timer symbol beside it labelled "pool_timeout: fail, do not hang". The right zone shows a single box labelled "database" with a note beneath reading "hard connection limit, typically low hundreds". Arrows run from tasks into the pool and from the pool to the database, with the arrows into the database numbering exactly ten to emphasise the fixed count.
Required elements: More tasks than connections, a pool container with in-use and idle tokens, a waiting queue with a timeout marker, a database box with the connection limit note, exactly ten arrows reaching the database.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector, consistent token sizing.
Layout: Three zones left to right, tasks stacked vertically on the left, pool centred, database on the right.
Text labels: "concurrent tasks", "connection pool (size 10, overflow 5)", "idle", "in use", "waiting for a free connection", "pool_timeout: fail, do not hang", "database", "hard connection limit, typically low hundreds".
Aspect ratio: 16:9
Accessibility: Mark idle and in-use tokens with text labels rather than colour alone, and count arrows visibly.
Avoid: Vendor logos, decorative icons, screenshots, tiny text, watermarks.
Alt text: Diagram showing twelve concurrent tasks served by a connection pool of ten connections with an overflow allowance, two tasks queued waiting with a pool timeout that fails rather than hanging, and a fixed number of connections reaching the database.
END IMAGE PROMPT]

Parameterized queries and SQL injection

# Never
query = f"SELECT * FROM documents WHERE source = '{source}'"

If source contains ' OR '1'='1, the query returns every row. If it contains '; DROP TABLE documents; --, the consequences are worse. The value came from a scraped page, an API request, or a model's tool call, and none of those are trustworthy.

Parameterized queries send the query and the values separately, so a value can never be interpreted as SQL.

from sqlalchemy import text

result = await session.execute(
    text("SELECT * FROM documents WHERE source = :source"),
    {"source": source},
)

The database receives the query structure once and the value as data. There is no parsing step in which the value could become syntax, which is why this is a structural fix rather than an escaping trick.

Escaping is not the answer. Attempting to sanitise input by removing quotes fails against encoding tricks, Unicode variants, and cases you did not anticipate. Parameterization removes the possibility rather than filtering for it.

What cannot be parameterized. Table names, column names, and sort directions are query structure, not values, so they cannot be bound as parameters. When one must be dynamic, validate it against an allowlist:

SORTABLE_COLUMNS = frozenset({"created_at", "title", "word_count"})


def build_sort(column: str, direction: str) -> str:
    if column not in SORTABLE_COLUMNS:
        raise ValidationError(f"cannot sort by {column!r}")
    if direction not in {"asc", "desc"}:
        raise ValidationError(f"invalid direction {direction!r}")
    return f"ORDER BY {column} {direction}"

This matters directly for tool calling. A tool letting a model choose a sort column is a place where a model-supplied string reaches query structure, and Module 6's principle applies unchanged: validate against a closed set before it gets anywhere near the query.

Transactions, commit, rollback, and atomicity

A transaction groups operations so that either all of them take effect or none do.

async def store_extraction(doc: Document, extraction: ExtractedRecipe) -> None:
    """Both writes succeed, or neither does."""
    async with SessionFactory() as session:
        async with session.begin():                  # transaction boundary
            session.add(DocumentRow.from_document(doc))
            session.add(ExtractionRow.from_extraction(extraction))
        # commits on clean exit, rolls back on exception

Without a transaction, a failure between the two writes leaves a document with no extraction, and nothing records that the second write was intended. Atomicity is the property that prevents this: the pair is indivisible.

session.begin() as a context manager commits on success and rolls back on any exception, which is the behaviour you want and removes the possibility of forgetting.

Keep transactions short. A transaction holds locks, and the longer it is open the more it blocks others. The specific mistake to avoid is doing slow work inside one:

# Wrong: an API call inside a transaction
async with session.begin():
    doc = await session.get(DocumentRow, doc_id)
    summary = await provider.generate(...)      # seconds, holding locks
    doc.summary = summary

That transaction is open for the duration of a model call, which under Module 5's timeout settings could be minutes. Do the slow work first, then open a short transaction to write the result.

Understand what your transaction actually guarantees. Isolation levels determine what one transaction sees of another's uncommitted work, and defaults differ by database. The practical advice at this level is to know that the default is usually read committed, which does not prevent the lost update problem covered next, and to look up the specifics when correctness under concurrency matters.

Optimistic versus pessimistic concurrency

Two workers update the same document. Both read version A, both modify it, both write. The second write overwrites the first, and the first update is gone with no error anywhere. This is the lost update problem.

Pessimistic concurrency locks the row on read so the second reader waits.

async with session.begin():
    result = await session.execute(
        select(DocumentRow).where(DocumentRow.id == doc_id).with_for_update()
    )
    row = result.scalar_one()
    row.status = "processed"

FOR UPDATE holds a lock until the transaction ends. It is correct and it serialises access, so it costs throughput and can deadlock when two transactions lock the same rows in different orders. Use it when conflicts are frequent and correctness matters more than concurrency, such as decrementing an inventory count.

Optimistic concurrency assumes conflicts are rare, detects them, and retries.

class DocumentRow(Base):
    __tablename__ = "documents"
    id: Mapped[str] = mapped_column(primary_key=True)
    status: Mapped[str]
    version: Mapped[int] = mapped_column(default=0)


async def update_status(doc_id: str, status: str) -> None:
    """Update only if nobody else has, otherwise raise a conflict."""
    async with SessionFactory() as session, session.begin():
        row = await session.get(DocumentRow, doc_id)
        if row is None:
            raise RetrievalError(f"document {doc_id} not found")

        result = await session.execute(
            update(DocumentRow)
            .where(DocumentRow.id == doc_id, DocumentRow.version == row.version)
            .values(status=status, version=row.version + 1)
        )
        if result.rowcount == 0:
            raise ConflictError(f"document {doc_id} was modified concurrently")

The version check in the WHERE clause is the whole mechanism. If another writer incremented the version between the read and the write, zero rows match and the update did nothing, which rowcount reveals. The caller retries with fresh data.

Optimistic concurrency suits pipelines well, because conflicts are genuinely rare and the retry is cheap. It also fails loudly, which is the property that matters: a silent lost update is invisible until someone notices missing data weeks later.

[IMAGE PROMPT M7-4
Purpose: Show how a lost update occurs and how optimistic version checking detects it.
Visual type: Two-panel sequence diagram over a shared timeline.
Prompt: A clean educational comparison with two stacked panels, each showing two horizontal lanes labelled "Worker 1" and "Worker 2" running left to right over a shared time axis, with a third lane beneath labelled "Row state". The upper panel is headed "Lost update" and shows: Worker 1 reads at t1 marked "reads status=pending"; Worker 2 reads at t2 marked "reads status=pending"; Worker 1 writes at t3 marked "writes status=processed"; Worker 2 writes at t4 marked "writes status=rejected". The row state lane shows "pending" then "processed" then "rejected", with an annotation beneath t4 reading "Worker 1's update silently gone, no error". The lower panel is headed "Optimistic version check" and shows the same four events, but the reads are marked "reads status=pending, version=3", Worker 1's write is marked "writes with WHERE version=3, succeeds, version=4", and Worker 2's write is marked "writes with WHERE version=3, matches 0 rows". An annotation beneath reads "conflict detected, caller retries with fresh data".
Required elements: Two worker lanes plus a row state lane in each panel, four ordered events per panel, version numbers in the lower panel only, the two contrasting annotations.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector, clear time ordering markers.
Layout: Two panels stacked vertically, each reading left to right on a shared time axis.
Text labels: "Lost update", "Optimistic version check", "Worker 1", "Worker 2", "Row state", "reads status=pending", "reads status=pending, version=3", "writes status=processed", "writes status=rejected", "writes with WHERE version=3, succeeds, version=4", "writes with WHERE version=3, matches 0 rows", "pending", "processed", "rejected", "Worker 1's update silently gone, no error", "conflict detected, caller retries with fresh data".
Aspect ratio: 16:9
Accessibility: Convey ordering through left to right position and explicit event text rather than colour.
Avoid: Code screenshots, decorative icons, tiny text, logos, watermarks.
Alt text: Two sequence diagrams showing two workers reading the same row and both writing, with the first update silently lost in the top panel, and a version check in the where clause detecting the conflict in the bottom panel.
END IMAGE PROMPT]

Pagination and keyset pagination

Returning two million rows at once exhausts memory. Offset pagination is the obvious approach and degrades badly.

SELECT * FROM documents ORDER BY created_at LIMIT 50 OFFSET 100000;

The database must produce and discard 100,000 rows to return 50. Page one is fast and page two thousand is slow, and the cost grows with the page number. There is a correctness problem too: if a row is inserted while a user pages through results, the offset shifts and a row is either skipped or shown twice.

Keyset pagination, also called cursor pagination, remembers where the last page ended.

async def list_documents(
    session: AsyncSession,
    *,
    after: tuple[datetime, str] | None = None,
    limit: int = 50,
) -> list[DocumentRow]:
    """Page by keyset. `after` is the (created_at, id) of the last row seen."""
    query = select(DocumentRow).order_by(DocumentRow.created_at, DocumentRow.id).limit(limit)
    if after is not None:
        last_created, last_id = after
        query = query.where(
            tuple_(DocumentRow.created_at, DocumentRow.id) > (last_created, last_id)
        )
    result = await session.execute(query)
    return list(result.scalars())

The database seeks directly to the position using the index and reads 50 rows, so page two thousand costs the same as page one. Rows inserted during paging do not shift anything, because the position is a value rather than a count.

Two requirements. The sort key must be unique, which is why id is included alongside created_at: two documents sharing a timestamp would otherwise produce an ambiguous boundary. And the ordering columns must be indexed together, or you have moved the cost rather than removed it.

The tradeoff is that keyset pagination cannot jump to an arbitrary page number, since it only knows how to continue from a position. For a "next page" interface it is strictly better. For a numbered page interface it does not apply, which is one reason numbered pagination is disappearing from large datasets.

Indexes: enough to recognise a slow query

An index is a separate structure letting the database find rows without scanning the table. Without one, WHERE content_hash = ... reads every row.

class DocumentRow(Base):
    __tablename__ = "documents"

    id: Mapped[str] = mapped_column(primary_key=True)
    content_hash: Mapped[str] = mapped_column(index=True)
    source_domain: Mapped[str]
    created_at: Mapped[datetime]

    __table_args__ = (
        Index("ix_documents_domain_created", "source_domain", "created_at"),
    )

Five things worth knowing.

Primary keys are indexed automatically. Foreign keys usually are not, which surprises people, and an unindexed foreign key makes joins slow.

Composite index order matters. An index on (source_domain, created_at) helps a query filtering by domain, or by domain and date, but not one filtering by date alone. The leading column must be used.

Indexes cost writes. Every insert and update maintains every index on the table, so indexing every column makes writes noticeably slower for no benefit.

Wrapping a column in a function usually defeats its index. WHERE lower(title) = 'x' cannot use a plain index on title, and the fix is either an index on the expression or storing the normalised value.

EXPLAIN tells you what actually happened. Prefixing a query with EXPLAIN or EXPLAIN ANALYZE shows whether an index was used or the table was scanned. When a query is slow, this is the first thing to run, and reading it is the single most useful database skill at this level.

SQLAlchemy Core versus ORM

SQLAlchemy offers two ways to work, and choosing deliberately is better than drifting into one.

Core is SQL expressed in Python. You write queries, you get rows back.

result = await session.execute(
    select(documents.c.id, documents.c.title)
    .where(documents.c.source_domain == domain)
    .order_by(documents.c.created_at.desc())
    .limit(50)
)
rows = result.all()

ORM maps rows to objects and tracks changes to them.

result = await session.execute(
    select(DocumentRow).where(DocumentRow.source_domain == domain).limit(50)
)
docs = result.scalars().all()
docs[0].status = "processed"          # tracked, written on flush

The ORM is convenient for record-oriented work, meaning loading an entity, changing it, and saving it. It costs you a layer of behaviour you must understand: identity mapping, flush timing, lazy loading, and cascades all do useful things and all surprise people who have not read about them.

The specific hazard in async code is lazy loading. Accessing a relationship the ORM has not loaded triggers a query, and in async that raises rather than working silently, which is arguably a kindness. The fix is to load relationships explicitly, and expire_on_commit=False in the session factory above exists to prevent a related problem where attributes are expired after commit and re-fetched on next access.

A reasonable default for this course. Use the ORM for entities your application owns and modifies. Use Core for bulk operations, reporting queries, and anything where you want to see exactly what SQL runs. Mixing them in one codebase is normal and fine.

Whichever you choose, the next lesson keeps it out of your business logic entirely.