Related Links

• Part 2 — Cognitive Layer. LLM Integration
• Part 3 — External Integration. Users Engagement
• Part 4 — Publication Lifecycle. Editorial Review
• Repository — main branch (this part)
• Pull Request #1 — iteration-02 → main (Part 2)
• Pull Request #2 — iteration-03 → main (Part 3)
• Pull Request #3 — iteration-04 → main (Part 4)

1. Foundations

Domain-Driven Design was introduced by Eric Evans in Domain-Driven Design: Tackling Complexity in the Heart of Software (2003), also known as "the Blue Book."

I think that the Domain-Driven Design book describes the experience of Eric Evans and his way to build Stateful Systems. I understand it as another cognitive layer on top of f(state, input) -> (state, output).

Is it good? It's great! Am I using it as a manifest for building software? Absolutely no. Should we learn it? Absolutely yes.

I will develop the same project across three real features so the rule has to hold up under change, not just on the first commit.

This articles shows only one side of the DDD — the direction of dependencies.

1.1 Four Parts

The point is what the diff looks like in each case, not the features. External change: the domain barely moves. Internal change: the domain is exactly where the work lands. Dependency direction tells you which case you're in before you touch anything.

1.2 Core Concepts

Two short analogies for the same idea.

Plumbing. The domain is the plumbing layout of a house — where the pipes run, how water is supposed to flow. The infrastructure is what those pipes are made of: copper, PVC, PEX. The layout does not care. You can renovate the material one room at a time without redrawing the blueprint.

Music. The domain is the rhythm and chord progression of a song — the structure every player must fit. The infrastructure is the players: bass, guitar, keys, drums. Each plays in its own voice, but all conform to the same underlying pulse. The application layer is the arrangement that threads the players together over the rhythm. Change the rhythm and the song stops being itself; change a player and it is the same song in a different texture.

In code, the domain holds the business rules and the contracts. The infrastructure decides the material: SQLAlchemy, PostgreSQL, an in-memory dict, a CSV file, an AI provider, a third-party scraper. The domain never names the material.

1.3 REST Architecture

The REST (Representational state transfer) architecture was defined by Roy Fielding in 2000.

REST Fundamental principles:

1. Client/Server separation
2. Stateless Clients
3. Data caching
4. Uniform interface
5. Layered system

So, if I develop REST DDD Server API, I consider next information:

1. I develop HTTP Server
2. I provide an interface for HTTP Client
3. HTTP Server exposes "HTTP Resources" to claim the information or mutate the state of the Server

I like to separate structures that HTTP Client can "see". I call them "HTTP Contracts". HTTP Contract is just a meaning of a data structure, that represents the uniform interface of accessing the system.

Based on that information I have http/resources with HTTP endpoints and http/contracts with HTTP Request Bodies and Responses definitions.

2. Implementation & Examples

The full source for everything below lives in this project's Gitea repository. The snippets here are trimmed for the article — the repository carries the docstrings, the tests, and any plumbing that did not make the page.

2.1 Project Shape

➜  fastapi-ddd-blog git:(main) ✗ tree -I '__init__.py|.gitignore'
.
├── pyproject.toml
├── README.md
└── src
    ├── application
    │   └── articles.py
    ├── domain
    │   ├── articles
    │   │   ├── entities.py
    │   │   ├── policies.py
    │   │   └── repository.py
    │   ├── base
    │   │   └── models.py
    │   └── errors
    ├── http
    │   ├── _base.py
    │   ├── contracts
    │   │   └── articles.py
    │   └── resources
    │       └── articles.py
    ├── infrastructure
    │   ├── application
    │   │   ├── error_handlers.py
    │   │   └── factory.py
    │   └── database
    │       ├── dal.py
    │       ├── repositories
    │       │   └── articles.py
    │       ├── tables.py
    │       └── transaction.py
    ├── main.py
    └── tests
        ├── conftest.py
        ├── fakes
        │   └── articles.py
        ├── integration
        ├── smoke
        └── unit
            ├── conftest.py
            ├── fakes
            └── test_articles.py

20 directories, 21 files

Each folder has a single responsibility:

2.2 Dependency Direction

HTTP request
    |
    v
http/resources/articles.py  <-- input/output (from the HTTP interation perspective)
    |
    v
application/articles.py      <-- implementation, orchestration
    |
    v
domain/articles/repository.py        <-- abstraction
    ^
    |
infrastructure/database/repositories/articles.py     <-- implementation

Everything points at the domain. Nothing inside the domain points outward.

2.3 Domain entities

# src/domain/articles/entities.py

from src.domain.base import DomainModel
from src.domain.errors import ArticleContainsStopWord

from .policies import find_stop_word

class ArticleDraft(DomainModel):
    title: str
    slug: str
    summary: str
    body: str
    published_on: date

    def validate_policies(self) -> None:
        _check_stop_words(title=self.title, summary=self.summary, body=self.body)


class ArticleUpdate(DomainModel):
    # No `slug` — slug is the identifier, not a mutable attribute.
    title: str
    summary: str
    body: str
    published_on: date

    def validate_policies(self) -> None:
        _check_stop_words(title=self.title, summary=self.summary, body=self.body)


class Article(ArticleDraft):
    id: int


class ArticleSummary(DomainModel):
    # Same as Article minus the body.
    id: int
    title: str
    slug: str
    summary: str
    published_on: date


def _check_stop_words(**fields: str) -> None:
    for name, value in fields.items():
        word = find_stop_word(value)
        if word is not None:
            raise ArticleContainsStopWord(field=name, word=word)

Three things become visible at a glance:

1. ArticleUpdate drops slug because the slug is the identifier and is not mutable
2. ArticleSummary drops body so list endpoints do not ship every paragraph of every post
3. Only the two writeable shapes carry validate_policies(). The rule itself — a forbidden-words list — lives next to the entity:

# src/domain/articles/policies.py

STOP_WORDS: frozenset[str] = frozenset({"spam", "clickbait", "scam"})

def find_stop_word(text: str) -> str | None:
    lowered = text.lower()
    for word in STOP_WORDS:
        if word in lowered:
            return word

    return None

2.4 Domain Errors

I used to use Python built-in errors to define custom errors for the Domain Layer. These errors could be translated into errors, that are convenient for the HTTP interraction. Long story short, you just create an interpretation layer into proper status codes.

Some examples:

• ArticleNotFound --> HTTP 404
• ArticleContainsStopWord --> HTTP 409

Errors like ArticleNotFound could be ommited because they could be generalized to NotFound error. But for the best control flow I recommend to define any business-concept errors to be able to handle any edge case.

# src/domain/errors/__init__.py

class DomainError(Exception):
    pass


class ArticleNotFound(DomainError):
    def __init__(self, identifier: int | str) -> None:
        super().__init__(f"Article {identifier!r} was not found.")
        self.identifier = identifier


class ArticleContainsStopWord(DomainError):
    def __init__(self, *, field: str, word: str) -> None:
        super().__init__(
            f"Article {field!r} contains the forbidden word {word!r}."
        )
        self.field = field
        self.word = word


class ArticleSlugAlreadyExists(DomainError):
    def __init__(self, slug: str) -> None:
        super().__init__(f"Article with slug {slug!r} already exists.")
        self.slug = slug

2.5 Repository Pattern

According to the "Blue Book", the repository represents interraction through the Aggregate. So the ArticlesRepository is something, like a "Bookshelf" and methods should represent operations with Entities, that are aggregated into Aggregates.

The "Repository" pattern is not specific for DDD, it's documented in 2003 in Martin Fowler's book, called "Catalog of Patterns of Enterprise Application Architecture"

From my experience, using of "Repository" pattern without properly defined business language won't be beneficial. The point of using the "Repository" layer for me is Reducing Cognitive Pressure during building complex application systems

P.S. CQRS/CQS is not covered in this article, but in principle you can "break" the Repository into 2 pieces to have "Commands" & "Queries" separation.

# src/domain/articles/repository.py

import abc
import functools

from .entities import Article, ArticleDraft, ArticleSummary, ArticleUpdate


class BookshelfRepository(abc.ABC):
    """Interface to manage articles in the persistent storage

    NOTES:
    (1) Only a few methods are represented for this article
    (2) The `.article()` is polymorphic. The next pattern could be used.
        It allows to create a facade to access article and hide implementations.
    """

    def __init__(self) -> None:
        self._articles: list[ArticleSummary] = []
        self._last_loaded_article: Article | None = None

    @abc.abstractmethod
    async def load_articles(self) -> list[ArticleSummary]:
        """Query articles and preserve them in memory"""

    @abc.abstractmethod
    async def add_article(self, draft: ArticleDraft) -> Article:
        """Add a new article"""

    # Polymorphic. Based on the first type.
    @functools.singledispatchmethod
    async def article(self, identifier) -> Article:
        """Polymorphic article retrieval by ID or Slug"""
        raise NotImplementedError(
            f"Unsupported identifier type: {type(identifier).__name__}"
        )

    @article.register
    async def _(self, identifier: int) -> Article:
        return await self._article_by_id(identifier)

    @article.register
    async def _(self, identifier: str) -> Article:
        return await self._article_by_slug(identifier)

    @abc.abstractmethod
    async def _article_by_id(self, identifier: int) -> Article:
        """Get article by id"""

    @abc.abstractmethod
    async def _article_by_slug(self, identifier: str) -> Article:
        """Get article by slug"""

    # ...

The pattern: singledispatchmethod is the concrete dispatcher — it picks the implementation by argument type. The typed underscore methods are thin routers to the abstract _article_by_* / _delete_by_* operations. Subclasses implement the abstracts; the dispatch logic lives once, in the ABC.

2.6 Application layer

# src/application/articles.py

from src.domain.articles import (
    Article,
    ArticleDraft,
    ArticleSummary,
    ArticleUpdate,
    BookshelfRepository,
)


async def articles_list(repository: BookshelfRepository) -> list[ArticleSummary]:
    return await repository.load_articles()

async def publish_article(
    repository: BookshelfRepository, draft: ArticleDraft
) -> Article:
    draft.validate_policies()
    return await repository.add_article(draft)

async def delete_article(repository: BookshelfRepository, slug: str) -> None:
    await repository.delete_article(slug)

# ...

update_article does a slug → id hop on purpose: the repository's update_article is typed on the integer id (the row's stable identifier), but the URL slug is what the HTTP client knows. The translation lives in the application layer so the route stays slug-shaped and the repository stays id-shaped.

2.7 HTTP Layer

HTTP Layer represents the "Presentation" layer according to the "Blue Book".

HTTP Contracts

# src/http/contracts/articles.py

from src.http._base import PublicModel

class ArticleCreateRequest(PublicModel):
    title: str
    slug: str
    summary: str
    body: str
    published_on: date


class ArticleUpdateRequest(PublicModel):
    # `slug` is in the URL path, not the body.
    title: str
    summary: str
    body: str
    published_on: date


class ArticleSummaryPublic(PublicModel):
    id: int
    title: str
    slug: str
    summary: str
    published_on: date


class ArticlePublic(ArticleSummaryPublic):
    body: str

HTTP Resources

# src/http/resources/articles.py

from fastapi import APIRouter, status

from src.application import articles
from src.domain.articles import ArticleDraft
from src.http.contracts.articles import ArticleCreateRequest, ArticlePublic
from src.infrastructure.database.repositories.articles import (
    SqlAlchemyArticlesRepository,
)
from src.infrastructure.database.transaction import transactional


router = APIRouter(prefix="/articles", tags=["Articles"])


@router.get("/{slug}", status_code=status.HTTP_200_OK)
async def article_details(slug: str) -> ArticlePublic:
    repository = SqlAlchemyArticlesRepository()
    article = await articles.get_article(repository, slug)
    return ArticlePublic.model_validate(article)


@router.post("", status_code=status.HTTP_201_CREATED)
@transactional
async def article_create(body: ArticleCreateRequest) -> ArticlePublic:
    repository = SqlAlchemyArticlesRepository()
    candidate = ArticleDraft.model_validate(body, from_attributes=True)
    article = await articles.publish_article(repository, candidate)
    return ArticlePublic.model_validate(article)

Naming convention: when a route expects a request body, the parameter is called body — not schema, not data. It matches what the HTTP spec calls it and reads naturally on the call site.

The route does not take session: AsyncSession = Depends(get_session). The session is request-scoped through middleware (see §2.9), and SqlAlchemyArticlesRepository() picks it up from the ContextVar. The router stays slug-shaped and free of plumbing.

HTTP Request Visualization

POST /articles
ArticleCreateRequest { title, slug, summary, body, published_on }
  │
  ▼
http/resources/articles.py · article_create()
  │   @transactional opens a transaction
  │   ArticleCreateRequest → ArticleDraft (domain shape)
  ▼
application/articles.py · publish_article()
  │   draft.validate_policies()
  │      └── raises ArticleContainsStopWord on a forbidden word
  │          (caught later by the error mapper → 400)
  ▼
infrastructure/database/repositories/articles.py · SqlAlchemyArticlesRepository.add_article()
  │   session.add( ArticlesTable(**draft.model_dump()) )
  │   await self.flush()   ← id populated, visible to next call in this tx
  │   build Article from the flushed row
  ▼
http/resources/articles.py · article_create() (continued)
  │   Article → ArticlePublic
  │
  ▼   (transaction commits as @transactional exits)
HTTP 201 + JSON body

A read request is the same shape minus the validation step and minus the transaction. The route does not implement business logic and never writes status codes for domain errors — that is the next section.

2.8 Error Mapping

This is the part worth pausing on.

The domain raises specific errors (ArticleNotFound, ArticleContainsStopWord, and in later parts CognitiveOutputRefused, ExternalSourceUnreachable, ...).

The HTTP routes never decide what status code those errors deserve. The translation happens in exactly one file — infrastructure/application/error_handlers.py. That file is the mapper: a domain error goes in, a standardized REST response comes out.

# src/infrastructure/application/error_handlers.py

from fastapi import Request, status
from fastapi.responses import JSONResponse

from src.domain.errors import (
    ArticleContainsStopWord,
    ArticleNotFound,
    ArticleSlugAlreadyExists,
    DomainError,
)


async def article_not_found_handler(
    _: Request, exc: ArticleNotFound
) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={"detail": str(exc), "identifier": exc.identifier},
    )


async def article_contains_stop_word_handler(
    _: Request, exc: ArticleContainsStopWord
) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={"detail": str(exc), "field": exc.field, "word": exc.word},
    )


async def article_slug_already_exists_handler(
    _: Request, exc: ArticleSlugAlreadyExists
) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_409_CONFLICT,
        content={"detail": str(exc), "slug": exc.slug},
    )


ERROR_HANDLERS: tuple[tuple[type[DomainError], object], ...] = (
    (ArticleNotFound, article_not_found_handler),
    (ArticleContainsStopWord, article_contains_stop_word_handler),
    (ArticleSlugAlreadyExists, article_slug_already_exists_handler),
)

# src/infrastructure/application/factory.py

from collections.abc import Iterable
from fastapi import APIRouter, FastAPI

from .error_handlers import ERROR_HANDLERS


def create_app(*, rest_routers: Iterable[APIRouter]) -> FastAPI:
    app = FastAPI()

    for router in rest_routers:
        app.include_router(router)

    for exc_type, handler in ERROR_HANDLERS:
        app.add_exception_handler(exc_type, handler)

    return app

Two consequences fall out of this split:

• The domain never has to know what an HTTP status code is.
• The HTTP routes never have to know what business meaning hides behind a 404 or a 409.

A new domain error means one new handler in error_handlers.py and one extra entry in ERROR_HANDLERS. If the error originates in infrastructure-specific behavior — duplicate slug on insert, for example — the repository translates it into the domain error first, and the routes still do not change.

The path an error walks from the place it is raised to the place it becomes a status code is worth tracing once:

infrastructure/database/repositories/articles.py
   _fetch_row(slug)  finds no matching row
      └── raise ArticleNotFound(slug)
                    │
                    ▼  (uncaught — propagates)
application/articles.py · get_article()        no try/except
                    │
                    ▼  (uncaught — propagates)
http/resources/articles.py · article_details() no try/except
                    │
                    ▼
FastAPI exception dispatcher
   finds the mapper registered by create_app
                    │
                    ▼
infrastructure/application/error_handlers.py
   article_not_found_handler(request, exc)
                    │
                    ▼
HTTP 404
{ "detail": "Article with slug 'xyz' was not found." }

The repository, the use case, and the route are all written as though the happy path is the only path. The mapper is the one place in the codebase that translates those failure paths into HTTP responses — and it knows them as domain concepts, not as transport details.

2.9 Infrastructure Layer. SQLAlchemy Repository

The database is assumed to already exist. The application configures a connection, hands out one async session per request, and runs the repository against it.

# src/infrastructure/database/dal.py

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from contextvars import ContextVar

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)


DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+asyncpg://blog:blog@localhost:5432/blog",
)


_engine = create_async_engine(DATABASE_URL, future=True)
_session_factory = async_sessionmaker(_engine, expire_on_commit=False)

# Per-task pointer to the session bound to the current request.
_session_var: ContextVar[AsyncSession | None] = ContextVar(
    "_session_var", default=None
)


def current_session() -> AsyncSession:
    session = _session_var.get()
    if session is None:
        raise RuntimeError(
            "No active session. `current_session()` must run inside a "
            "request scoped by `session_scope()`."
        )
    return session


class SqlAlchemyDAL:
    def __init__(self, session: AsyncSession | None = None) -> None:
        self._session = session or current_session()

    @property
    def session(self) -> AsyncSession:
        return self._session

    async def flush(self) -> None:
        await self._session.flush()

The session is bound to a ContextVar, not threaded through every call. SqlAlchemyDAL() reads it through current_session(); tests can still pass an explicit session.

A small middleware seeds that ContextVar for every request, so the route handlers stay free of Depends(get_session):

# src/infrastructure/database/dal.py  (continued)

@asynccontextmanager
async def session_scope() -> AsyncIterator[AsyncSession]:
    """Open a session, bind it to the current task, and tear it down."""
    async with _session_factory() as session:
        token = _session_var.set(session)
        try:
            yield session
        finally:
            _session_var.reset(token)

# src/infrastructure/application/factory.py  (excerpt)

from starlette.middleware.base import BaseHTTPMiddleware
from src.infrastructure.database.dal import session_scope


class DatabaseSessionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        async with session_scope():
            return await call_next(request)


def create_app(*, rest_routers: Iterable[APIRouter]) -> FastAPI:
    app = FastAPI()
    app.add_middleware(DatabaseSessionMiddleware)
    for router in rest_routers:
        app.include_router(router)
    for exc_type, handler in ERROR_HANDLERS:
        app.add_exception_handler(exc_type, handler)
    return app

That is the full plumbing for "instantiate a repository without arguments." Routes don't see a session; workers and one-off scripts open session_scope() directly.

SqlAlchemyDAL is the base for every SQLAlchemy-backed repository. It owns the session and surfaces only what a repository should touch — read-only session for building statements, and flush() for pushing pending changes inside the current transaction. A repository inherits the DAL to get this plumbing; it never reaches into self._session directly, and flush() becomes a first-class operation instead of a sprinkled side-effect.

# src/infrastructure/database/tables.py

from sqlalchemy import Date, Integer, MetaData, String, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    metadata = MetaData()


class ArticlesTable(Base):
    __tablename__ = "articles"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    slug: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
    summary: Mapped[str] = mapped_column(String(500), nullable=False)
    body: Mapped[str] = mapped_column(Text, nullable=False)
    published_on: Mapped[Date] = mapped_column(Date, nullable=False)

ArticlesTable and the domain's Article describe the same idea, but they live in different layers on purpose.

The table is the persistence detail; the entity is the business shape so I don't keep Data Mapper and Domain in the same place.

# src/infrastructure/database/repositories/articles.py

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

from src.domain.articles import (
    Article,
    ArticleDraft,
    ArticleSummary,
    ArticleUpdate,
    BookshelfRepository,
)
from src.domain.errors import ArticleNotFound, ArticleSlugAlreadyExists
from src.infrastructure.database.dal import SqlAlchemyDAL
from src.infrastructure.database.tables import ArticlesTable


class SqlAlchemyArticlesRepository(SqlAlchemyDAL, BookshelfRepository):
    def __init__(self, session: AsyncSession | None = None) -> None:
        SqlAlchemyDAL.__init__(self, session)
        BookshelfRepository.__init__(self)

    async def load_articles(self) -> list[ArticleSummary]:
        stmt = select(
            ArticlesTable.id,
            ArticlesTable.title,
            ArticlesTable.slug,
            ArticlesTable.summary,
            ArticlesTable.published_on,
        )
        rows = (await self.session.execute(stmt)).all()
        self._articles = [
            ArticleSummary(
                id=row.id,
                title=row.title,
                slug=row.slug,
                summary=row.summary,
                published_on=row.published_on,
            )
            for row in rows
        ]
        return self._articles

    async def add_article(self, draft: ArticleDraft) -> Article:
        row = ArticlesTable(**draft.model_dump())
        self.session.add(row)
        try:
            await self.flush()
        except IntegrityError as exc:
            if _is_unique_slug_violation(exc):
                raise ArticleSlugAlreadyExists(draft.slug) from exc
            raise
        return self._to_entity(row)

    async def update_article(self, id: int, data: ArticleUpdate) -> Article:
        row = await self._fetch_row_by_id(id)
        for field, value in data.model_dump().items():
            setattr(row, field, value)
        await self.flush()
        return self._to_entity(row)

    async def _article_by_id(self, identifier: int) -> Article:
        return self._to_entity(await self._fetch_row_by_id(identifier))

    async def _article_by_slug(self, identifier: str) -> Article:
        return self._to_entity(await self._fetch_row_by_slug(identifier))

    async def _delete_by_id(self, identifier: int) -> None:
        row = await self._fetch_row_by_id(identifier)
        await self.session.delete(row)
        await self.flush()

    async def _delete_by_slug(self, identifier: str) -> None:
        row = await self._fetch_row_by_slug(identifier)
        await self.session.delete(row)
        await self.flush()

    async def _fetch_row_by_id(self, identifier: int) -> ArticlesTable:
        stmt = select(ArticlesTable).where(ArticlesTable.id == identifier)
        row = (await self.session.execute(stmt)).scalar_one_or_none()
        if row is None:
            raise ArticleNotFound(identifier)
        return row

    async def _fetch_row_by_slug(self, identifier: str) -> ArticlesTable:
        stmt = select(ArticlesTable).where(ArticlesTable.slug == identifier)
        row = (await self.session.execute(stmt)).scalar_one_or_none()
        if row is None:
            raise ArticleNotFound(identifier)
        return row

    @staticmethod
    def _to_entity(row: ArticlesTable) -> Article:
        return Article(
            id=row.id,
            title=row.title,
            slug=row.slug,
            summary=row.summary,
            body=row.body,
            published_on=row.published_on,
        )


def _is_unique_slug_violation(exc: IntegrityError) -> bool:
    message = str(getattr(exc, "orig", exc)).lower()
    return "unique" in message and "slug" in message

ArticleNotFound now carries a generic identifier rather than a hard-coded slug — both lookup paths use the same error.

The repository uses self.session and self.flush() — both inherited from SqlAlchemyDAL. It does not own the session, it does not call _session.flush() directly, and it does not need to know how the session arrived. Another aggregate's SQL repository would inherit the same DAL and look structurally identical.

The repository also observes storage-specific failure modes and converts them into domain-vocabulary errors: row is None becomes ArticleNotFound, and a unique-key violation on slug becomes ArticleSlugAlreadyExists. The HTTP layer never sees the database exceptions directly — the mapper from the previous section catches only the domain errors and emits the 404 or 409.

InMemoryArticlesRepository also lives in the same file and follows the same contract; tests pull it from src/tests/fakes/articles.py.

2.10 Transactional Database Queries

get_session() hands out a session but does not auto-commit. Atomicity is opt-in: a write block is bracketed explicitly. There are two equivalent ways to do that — pick whichever fits the call site.

# src/infrastructure/database/transaction.py

from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import asynccontextmanager
from functools import wraps
from typing import ParamSpec, TypeVar

from sqlalchemy.ext.asyncio import AsyncSession

from src.infrastructure.database.dal import current_session


P = ParamSpec("P")
R = TypeVar("R")


@asynccontextmanager
async def transaction() -> AsyncIterator[AsyncSession]:
    session = current_session()
    if session.in_transaction():
        async with session.begin_nested():
            yield session
    else:
        async with session.begin():
            yield session


def transactional(
    func: Callable[P, Awaitable[R]],
) -> Callable[P, Awaitable[R]]:
    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        async with transaction():
            return await func(*args, **kwargs)

    return wrapper

Both forms read the session out of the same ContextVar set by get_session(). The decorator form is what the HTTP layer above uses — @transactional sits between @router.<verb>(...) and the function so the entire handler runs as one transaction.

The context-manager form is for finer-grained control inside a use case. The interesting property is intra-transaction visibility: every repository write runs self.flush() from SqlAlchemyDAL before returning, so a later call in the same transaction() block sees the earlier work even though nothing has committed yet.

async with transaction():
    article = await articles.publish_article(draft, repository)
    # The DAL flushed during create(), so `article.id` is populated
    # and any follow-up call in this same transaction sees the row.

    await articles.update_article(article.slug, edits, repository)

If either call raises, the whole block rolls back. Nested calls open a SAVEPOINT instead of failing, so wrapping a use case that is already inside a transactional handler is safe:

async with transaction():             # outer: BEGIN
    await repo.create(...)
    async with transaction():         # inner: SAVEPOINT sp1
        await repo.update(...)
                                      # inner exits clean  → RELEASE sp1
    await repo.delete(...)
                                      # outer exits clean  → COMMIT

If the inner block raises, the outer transaction is still alive — only sp1 is rolled back. If the outer raises after the inner has released, everything rolls back. The reentrancy property means a use case never has to ask "am I already in a transaction?" before opening one.

Three properties fall out of this design:

• Reads stay cheap at the application boundary — no explicit transaction wrapper is opened until something asks for one.
• Writes are auditable — the boundary of "what commits together" is visible at the call site.
• Intra-transaction reads work — the DAL's flush() makes a write visible to the next query without committing.
• Composition is safe — transaction() is reentrant via SAVEPOINTs.

2.11 Entrypoint. Composition Root

The main.py file should tell about the overall application setup. Since I am developing the REST HTTP API, I compose available HTTP Routers with HTTP Resources into the Application Factory.

# src/main.py

from fastapi import FastAPI

from src.http.resources.articles import router as articles_router
from src.infrastructure.application import create_app


app: FastAPI = create_app(rest_routers=(articles_router,))

2.12 Testing

One of the main benefits of using the Repository pattern is: you can create a simple version of the Repository to exclude the complexity of the external system.

The code is represented according to the pytest testing framework with fixtures defined in the conftest.py.

It's also fine to not have the "InMemory" database representation if you rely on SQL more. This example is good to show how complex external system could be replaced with internal in-memory representations that hide irrelevant complexity during testing.

# src/tests/unit/conftest.py

from datetime import date

import pytest

from src.domain.articles import ArticleDraft
from src.tests.fakes.articles import InMemoryArticlesRepository


@pytest.fixture
def repository() -> InMemoryArticlesRepository:
    return InMemoryArticlesRepository(items=[])


@pytest.fixture
def sample_draft() -> ArticleDraft:
    return ArticleDraft(
        title="First post",
        slug="first-post",
        summary="A summary.",
        body="Hello.",
        published_on=date(2024, 1, 1),
    )

The creation flow is the most worthwhile place for parametrization — it is the only path where multiple input shapes round-trip through the whole stack.

# src/tests/unit/test_articles.py
from datetime import date

import pytest

from src.application import articles
from src.domain.articles import ArticleDraft


@pytest.mark.parametrize(
    "draft",
    [
        pytest.param(
            ArticleDraft(
                title="Short",
                slug="short",
                summary="Short summary",
                body="Body.",
                published_on=date(2024, 1, 1),
            ),
            id="short-title",
        ),
        pytest.param(
            ArticleDraft(
                title="Headline with multiple words",
                slug="headline-with-multiple-words",
                summary="A longer summary that spans several words.",
                body="Several sentences. With punctuation.",
                published_on=date(2024, 6, 15),
            ),
            id="multi-word-title",
        ),
        pytest.param(
            ArticleDraft(
                title="A" * 100,
                slug="very-long-title",
                summary="x",
                body="y",
                published_on=date(2023, 12, 31),
            ),
            id="max-length-title",
        ),
    ],
)
@pytest.mark.asyncio
async def test_publish_article_persists_various_drafts(
    repository, draft: ArticleDraft
) -> None:
    created = await articles.publish_article(repository, draft)
    fetched = await articles.get_article(repository, draft.slug)
    assert fetched == created

3. In Next Articles

For Parts 2 and 3 the diff lives almost entirely in infrastructure/ and at the edges of application/; the domain row gains one folder and one file across both parts.

Part 4 flips the picture — the domain row is where the work lands, with application following close behind. A seven-state article publication lifecycle with role-aware transitions, a split publication pipeline (mechanical gates submission, cognitive advises the supervisor), and persistence-shape vs HTTP-contract masking on a single reject_message field. The dependency direction never moves; what moves is which layer absorbs the change.

                   Part 1            Part 2                   Part 3                   Part 4
                   (foundation)      (external tool)          (external API)           (internal feature)
                   ──────────────    ─────────────────────    ──────────────────────   ──────────────────────
http/              full surface      +1 dispatcher route      +1 dispatcher route      +7 verb-named routes
application/       5 use cases       +3 editorial funcs       +1 import func           +7 lifecycle funcs
domain/            articles/         +cognitive_layer/        +articles/inbound.py     +users/ (one enum)
                   (1 aggregate)      (1 small folder)         (1 file)                 +articles/policies.py
                                                                                         (lifecycle merged in)
                                                                                         +articles/authors.py
                                                                                         +articles/publication/
infrastructure/    database/ +       +pydantic_bindings.py    +integrations/           (no new infra packages —
                   application/                                (medium.py, reddit.py)    Part 2's binding gains
                                                                                         2 prompts)

Domain holds the rule; infrastructure holds the material. Renovate the material without redrawing the blueprint.