Isolating persistence behind an interface

Database code spreading through an application is the same problem as provider SDK code spreading through it. The repository pattern applies the Lesson 3 solution to data.

Define what your application needs, in your vocabulary:

from typing import Protocol


class DocumentRepository(Protocol):
    """Storage for documents, independent of how they are stored."""

    async def get(self, doc_id: str) -> Document | None: ...

    async def get_by_hash(self, content_hash: str) -> Document | None: ...

    async def save(self, document: Document) -> None: ...

    async def save_many(self, documents: list[Document]) -> None: ...

    async def list_by_domain(
        self, domain: str, *, after: str | None = None, limit: int = 50
    ) -> list[Document]: ...

Three properties make this an interface rather than a thin wrapper.

It speaks in domain types. It returns Document, the record from Module 2, not a database row object. A caller never touches a DocumentRow.

It exposes intent, not queries. get_by_hash names a purpose. A repository with a query(sql) method has abstracted nothing.

It is a Protocol. Implementations do not import it, and a fake implementation for tests is any class with these five methods.

An implementation:

class SqlDocumentRepository:
    """Stores documents in a relational database."""

    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
        self._session_factory = session_factory

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

    async def get_by_hash(self, content_hash: str) -> Document | None:
        async with self._session_factory() as session:
            result = await session.execute(
                select(DocumentRow).where(DocumentRow.content_hash == content_hash)
            )
            row = result.scalar_one_or_none()
            return row.to_document() if row else None

    async def save(self, document: Document) -> None:
        async with self._session_factory() as session, session.begin():
            await session.merge(DocumentRow.from_document(document))

to_document and from_document are the translation between the storage shape and the domain shape. Keeping them as methods on the row class means the mapping lives in one place, and the domain record stays free of database concerns.

[IMAGE PROMPT M7-5
Purpose: Show the repository interface isolating business logic from storage implementations, and the dependency direction that makes swapping possible.
Visual type: Layered architecture diagram with multiple implementations.
Prompt: A clean educational layered diagram reading top to bottom. The top layer is a wide box labelled "ExtractionService (business logic)" with a sub-label "imports no database driver". A downward arrow labelled "depends on" leads to a narrower box in the middle labelled "DocumentRepository (Protocol)" with a sub-label "get, save, list_by_domain". Beneath it, three boxes are arranged side by side labelled "SqlDocumentRepository", "InMemoryDocumentRepository", and "SqliteDocumentRepository", each connected upward to the Protocol box by a dashed arrow labelled "satisfies structurally". Beneath those three, three corresponding small boxes read "Postgres", "dict in memory", and "SQLite file". A vertical bracket on the left spans the top two layers labelled "your domain: Document records only". A second bracket spans the bottom two layers labelled "infrastructure: rows, drivers, SQL".
Required elements: Four layers in the stated order, three implementations side by side with their backing stores, downward dependency arrow from service to protocol, dashed upward satisfaction arrows from implementations, two labelled brackets separating domain from infrastructure.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector, generous whitespace.
Layout: Vertical stack with the implementation layer spread horizontally into three columns.
Text labels: "ExtractionService (business logic)", "imports no database driver", "depends on", "DocumentRepository (Protocol)", "get, save, list_by_domain", "SqlDocumentRepository", "InMemoryDocumentRepository", "SqliteDocumentRepository", "Postgres", "dict in memory", "SQLite file", "satisfies structurally", "your domain: Document records only", "infrastructure: rows, drivers, SQL".
Aspect ratio: 16:9
Accessibility: Distinguish dependency and satisfaction arrows by line style and text label rather than colour.
Avoid: Vendor logos, decorative icons, screenshots, tiny text, watermarks.
Alt text: Layered diagram showing business logic depending on a document repository protocol, satisfied by SQL, in-memory, and SQLite implementations backed by Postgres, a dictionary, and a file, with brackets separating the domain layer from infrastructure.
END IMAGE PROMPT]

SQLite in tests, Postgres in production

The payoff is that swapping storage is a configuration change.

def build_repository(settings: Settings) -> DocumentRepository:
    """Choose a repository implementation from configuration."""
    if settings.database_url.startswith("sqlite"):
        engine = create_async_engine(settings.database_url)
    else:
        engine = create_async_engine(
            settings.database_url,
            pool_size=settings.db_pool_size,
            max_overflow=5,
            pool_timeout=5.0,
            pool_pre_ping=True,
        )
    return SqlDocumentRepository(async_sessionmaker(engine, expire_on_commit=False))

No service-layer code changes. ExtractionService receives something satisfying DocumentRepository and never learns which.

For unit tests, skip the database entirely:

class InMemoryDocumentRepository:
    """A repository backed by a dictionary. For tests."""

    def __init__(self) -> None:
        self._documents: dict[str, Document] = {}

    async def get(self, doc_id: str) -> Document | None:
        return self._documents.get(doc_id)

    async def get_by_hash(self, content_hash: str) -> Document | None:
        return next(
            (d for d in self._documents.values() if d.content_hash == content_hash),
            None,
        )

    async def save(self, document: Document) -> None:
        self._documents[document.id] = document

Tests of business logic now run in milliseconds with no database at all.

A caution worth stating. SQLite and Postgres are not identical, and a test suite passing on SQLite can fail in production. They differ on type strictness, concurrency behaviour, some SQL syntax, and specific features such as JSON operators and full text search. Run the fast suite on the in-memory or SQLite implementation for speed, and run an integration suite against real Postgres before shipping. The repository pattern makes both cheap, which is the point, but it does not make the databases equivalent.

Why business logic never imports a database driver

The rule is simple to state and worth being strict about.

# src/recipe_extractor/services/extraction.py

from recipe_extractor.domain import Document, ExtractedRecipe      # yes
from recipe_extractor.ports import DocumentRepository, LLMProvider # yes
import sqlalchemy                                                   # no
import asyncpg                                                      # no

Four reasons.

Testability. Business logic importing a driver needs a database to test, which turns millisecond tests into second tests and makes a fast suite impossible.

Swappability. The Postgres to SQLite change above only works because nothing above the repository knows the difference.

Comprehensibility. A service function containing queries is doing two jobs, and the business rules are interleaved with mechanics. Reading it means separating them mentally every time.

Correct dependency direction. Business rules are the stable part of a system and infrastructure is the changeable part. If the stable part depends on the changeable part, every infrastructure change reaches into your rules.

Making the rule enforceable. State it in a package layout, so that domain and services sit apart from infrastructure, and it becomes visible in review. A linting rule restricting imports per package can enforce it automatically, which is worth setting up once for a codebase more than one person touches.

Concept check. Your ExtractionService has a method that fetches documents needing extraction, calls the provider for each, and saves results. Where does the transaction boundary belong, and why is it awkward?

Answer

It is awkward precisely because a transaction is a database concept and the service is not supposed to know about databases. Three workable answers.

Give the repository a method expressing the whole unit of work, such as save_extraction_results(results), which opens one transaction internally. The service asks for an outcome and the repository decides how to make it atomic. This is usually the cleanest.

Introduce a unit-of-work abstraction the service can use, which is a Protocol with begin and commit that the SQL implementation maps onto a transaction and the in-memory one implements trivially. More machinery, and worth it when several repositories must commit together.

Accept per-document atomicity, saving each result in its own transaction. Often correct here, since the documents are independent and a partial run is resumable, which is exactly what Module 4's checkpointing was built for.

What you should not do is open a transaction spanning the provider calls, since Lesson 4 explained why holding locks across a network call is a serious problem.