# Clean Architecture / DDD 2026 Best Practices

> Researched May 15, 2026 from analysis of top-starred FastAPI+DDD projects.
> Sources: ivan-borovets/fastapi-clean-example (⭐554), Enforcer/clean-architecture (⭐565),
> AdamHavlicek/fastapi-todo-ddd (⭐140), alefeans/python-clean-architecture

## 1. Port Interfaces: Protocol over ABC

**Consensus:** Use `typing.Protocol` (structural subtyping) + `@abstractmethod` instead of `abc.ABC`.

**Arguments for Protocol:**
- Adapters don't need to inherit from the port — anything with matching shape is valid
- Mocks/stubs are trivially conformant without `create_autospec`
- Refactoring the port interface breaks adapter code at mypy time, not runtime
- More natural fit for Python's duck-typing philosophy

**Example from fastapi-clean-example:**
```python
from abc import abstractmethod
from typing import Protocol

class UserTxStorage(Protocol):
    """Transactional: commit required."""
    @abstractmethod
    def add(self, user: User) -> None: ...
    @abstractmethod
    async def get_by_id(self, user_id: UserId, *, for_update: bool = False) -> User | None: ...
```

## 2. Read/Write Repository Separation

Don't have a single `Repository[T]` interface with 15 methods. Split into:

- **Storage ports** (`UserTxStorage`, `AuthSessionStorage`) — write operations, full entity loading, transactional methods. Used by commands/interactors.
- **Reader ports** (`UserReader`, `SessionReader`) — read operations, optimized column selection, query models (not entities). Used by query services.

**Why:** Writes need consistency (load full entity, apply domain rules, flush, commit). Reads need speed (SELECT specific columns, pagination, sorting). Combining them in one interface forces every read operation to load full entities.

**Adapter sharing:** A single implementation class (e.g., `SqlaUserTxStorage`) can implement both `UserTxStorage` and `AuthzUserFinder` (a query interface) — the separation is at the *interface boundary*, not necessarily at the class level. Register it with two `provide()` calls in Dishka, one `provides=` per interface.

## 3. Granular Ports: Flusher + TransactionManager

**Innovation from fastapi-clean-example:** Instead of a single `UnitOfWork.commit()`, split into two interfaces:

```python
class Flusher(Protocol):
    """Flush pending changes to validate constraints mid-transaction."""
    async def flush(self) -> None: ...

class TransactionManager(Protocol):
    """Commit the entire unit of work."""
    async def commit(self) -> None: ...
```

**Flow in an interactor:**
1. Validate permissions (current_user_service)
2. Create/mutate domain entities
3. `storage.add(entity)` — mark for insertion in session
4. `flusher.flush()` — triggers SQL flush, catches constraint violations
5. Map `IntegrityError` to domain exception (e.g., `UsernameAlreadyExistsError`)
6. `transaction_manager.commit()` — finalize the transaction

**Why separate:** You want to catch `UniqueViolation` / `IntegrityError` for business-rule violations (duplicate username) *before* committing. With a single `commit()`, you lose the opportunity to map DB errors to domain exceptions in a controlled way.

## 4. Domain Entities: Classical (Imperative) ORM Mapping

**Approach:** Domain entities are pure Python classes with no SQLAlchemy dependency. SQLAlchemy's `mapper_registry.map_imperatively()` maps them to tables.

**From fastapi-clean-example:**
```python
# domain/entities/user.py — pure Python
class User(Entity[UserId]):
    def __init__(self, *, id_: UserId, username: Username, ...):
        super().__init__(id_=id_)
        self.username = username
        ...

# infrastructure/mappings/user.py — SQLAlchemy knows about domain entities
mapper_registry.map_imperatively(
    User,
    users_table,
    properties={
        "id_": users_table.c.id,
        "username": composite(Username, users_table.c.username),
        "password_hash": users_table.c.password_hash,
        "_created_at": composite(UtcDatetime, users_table.c.created_at),
        "updated_at": composite(UtcDatetime, users_table.c.updated_at),
    },
)
```

**Value objects via `composite()`:**
```python
# The Username type is mapped as a composite of a string column
Column("username", String(Username.MAX_LEN), nullable=False, unique=True),
```
SQLAlchemy automatically constructs `Username(value)` on reads and extracts `.value` on writes.

**When to use:**
- ✅ 5+ entities with 2+ value objects each — the mapping boilerplate pays off in duplication eliminated
- ✅ Value objects with validation — composite() keeps type safety through the DB roundtrip
- ❌ Simple CRUD module with 1-2 tables — separate ORM models with a thin adapter layer is simpler

**When NOT to use:** Very early stage projects where the domain model is still in flux — classical mapping adds ceremony that hurts during rapid prototyping. Start with separate models, then consolidate.

## 5. Rich Value Objects

**Every best-practice repo emphasizes:** Value objects should be self-validating, immutable, and introduced from day one — even when entities are kept anemic.

**Base class pattern:**
```python
@dataclass(frozen=True, slots=True, repr=False)
class ValueObject:
    """Immutable value object: defined by attributes only, no identity."""

    def __new__(cls, *_args, **_kwargs):
        if cls is ValueObject: raise TypeError("Cannot instantiate base")
        if not fields(cls): raise TypeError(f"{cls.__name__} must have at least one field")
        return object.__new__(cls)

    def __post_init__(self): ...
    def __repr__(self) -> str:
        # If 1 field: shows value; if 2+: shows name=value
        # Fields with repr=False are hidden (for secrets)
```

**Concrete examples:**
- `Username(value: str)` — validates length 3-150, ASCII only, immutable
- `RawPassword(value: str)` — repr hides value (all fields `repr=False`)
- `UtcDatetime(value: datetime)` — enforces UTC awareness

**When entities stay anemic:** Rich VOs carry the behavioral weight — they're easier to design and refactor than rich entities. Prefer rich value objects over rich entities in the first 6 months of a project.

## 6. Dependency Injection: Dishka

**Problem with FastAPI Depends:**
1. Leaks FastAPI into every layer — `from fastapi import Depends` in application code
2. No APP-scoped singletons — every request re-creates stateless services
3. No lifecycle hooks — async cleanup of connections, executors
4. Hard to override for testing — nested `Depends(lambda x=Depends(y): ...)` gets ugly
5. No provider composition — flat namespace of callables

**Dishka solution:**
```python
from dishka import Provider, Scope, provide
from dishka import make_async_container
from dishka.integrations.fastapi import setup_dishka

class CoreProvider(Provider):
    scope = Scope.REQUEST

    # APP-scoped singleton
    user_service = provide(UserService, scope=Scope.APP)

    # REQUEST-scoped, aliased via provides=
    user_tx_storage = provide(SqlaUserTxStorage, provides=UserTxStorage)
    flusher = provide(SqlaFlusher, provides=Flusher)
    tx_manager = provide(SqlaTransactionManager, provides=TransactionManager)

    # Factory method for dependencies needing constructor args
    @provide(scope=Scope.APP)
    def provide_password_hasher(self, settings: PasswordHasherSettings) -> PasswordHasher:
        return BcryptPasswordHasher(
            pepper=settings.PEPPER.encode(),
            work_factor=settings.WORK_FACTOR,
        )

# In app factory
container = make_async_container(CoreProvider(), ...)
setup_dishka(container, app)
```

**Testing overrides:**
```python
class TestProvider(Provider):
    user_tx_storage = provide(InMemoryUserStorage, provides=UserTxStorage)
    utc_timer = provide(StubUtcTimer, provides=UtcTimer)

# Override just what you need
container = make_async_container(CoreProvider(), TestProvider())
```

**Scopes guide:**
- `Scope.APP`: database engine, thread pool executor, hasher, JWT processor, settings
- `Scope.REQUEST`: session, current user, interactors, unit of work
- `Scope.SESSION` (if available): HTTP client connection pooling

## 7. CQRS at Application Level

**Decision guide:** Split `use_cases.py` into `commands/` and `queries/` when the module has 3+ use cases.

**Commands** (mutations):
- Load full entities via storage ports
- Apply business rules (domain services)
- Use `Flusher` + `TransactionManager`
- Return `TypedDict` with minimal response data (just IDs and timestamps)

**Queries** (reads):
- Use reader ports that SELECT specific columns, not full entities
- Return query models (`UserQm` = query model, not entity) — flat, serializable dataclasses
- No side effects, no transactions
- Can paginate and sort at DB level

**Example from fastapi-clean-example:**
```
application/
├── commands/
│   ├── create_user.py        # Interactor: CreateUser.execute(request) → CreateUserResponse
│   ├── activate_user.py
│   ├── deactivate_user.py
│   ├── grant_admin.py
│   └── ports/
│       ├── user_tx_storage.py
│       ├── flusher.py
│       ├── transaction_manager.py
│       └── utc_timer.py
├── queries/
│   ├── list_users.py         # Query service: ListUsers.execute(params) → ListUsersQm
│   └── ports/
│       └── user_reader.py
```

**DTOs per use case pattern:**
```python
# Request — frozen dataclass with input validation
@dataclass(frozen=True, slots=True, kw_only=True)
class CreateUserRequest:
    username: str
    password: str
    role: UserRoleRequestEnum

# Response — TypedDict (faster creation than dataclass for wire format)
class CreateUserResponse(TypedDict):
    id: UUID
    created_at: datetime
```

## 8. Testing: Stubs over Mocks

**Why stubs:** MagicMock/AsyncMock silently drift from the real interface. A stub is a concrete class implementing the Protocol — mypy catches signature mismatches.

**Stub example:**
```python
class StubPasswordHasher(PasswordHasher):
    async def hash(self, raw_password: RawPassword) -> UserPasswordHash:
        return UserPasswordHash(hashlib.sha256(raw_password.value).digest())

    async def verify(self, raw_password, hashed_password) -> bool:
        return await self.hash(raw_password) == hashed_password
```

**When to use mocks instead:** Only for testing retry logic, timeout behavior, or external API failure modes where a stub can't simulate the condition.

## 9. Architecture Quirks: The "Adapter Tax" Tradeoff

**From fastapi-clean-example (README):** The original Clean Architecture diagram implies that *all* external dependencies (Web, DB, UI) are behind interface boundaries. The project argues this is overengineering for most real-world Python projects and proposes a pragmatic revision:

> "Dependencies must never point outwards **within the core**."

This means:
- The Presentation and Infrastructure layers CAN depend directly on external frameworks (FastAPI, SQLAlchemy)
- But Domain and Application layers MUST NOT depend on them
- If you need to switch frameworks, you replace the adapter layer, not rewrite the core

**Trade-off accepted:** Adapters on the outermost layer are coupled to their framework. This is acceptable because:
1. Framework-switching is rare in practice
2. The cost of abstracting *everything* is higher than the cost of rewriting adapters
3. The core (domain + application) remains perfectly reusable and testable

## 10. Key Technology Choices

| Concern | 2026 Recommended | Why |
|---------|-----------------|-----|
| DI | Dishka | Framework-agnostic, scoped providers, async lifecycle |
| ORM mapping | Classical imperative (map_imperatively) | No duplication between domain entities and DB models |
| Validation | Value objects in domain, Pydantic in presentation | VOs: self-validating types. Pydantic: HTTP schema validation |
| Password hashing | pwdlib (Argon2 + Bcrypt fallback) | Auto-algorithm migration |
| Logging | structlog | Structured, JSON-formatted, searchable |
| Linting | ruff | 100x faster than flake8, replaces isort/autoflake as well |
| Type checking | mypy (strict=true) | Catches Protocol conformance, Optional violations |
| Package mgr | uv | 10-100x faster than pip/poetry, lockfile support |
| Testing | pytest + stubs + httpx.AsyncClient | Stubs over mocks, ASGI-transport HTTP testing |
| Config | pydantic-settings | Type-safe env vars, computed fields, cross-field validation |

## Key Repositories

- **ivan-borovets/fastapi-clean-example** (⭐554) — Most comprehensive: Dishka DI, CQRS, classical mapping, Flusher+TransactionManager, stub testing, rich VOs. Actively updated May 2026. Structure: `src/app/core/{commands,queries,common}`, `src/app/{inbound,outbound}`, `src/app/setup/ioc/`.
- **Enforcer/clean-architecture** (⭐565) — Book companion repo. DDD-heavy auctions example with layered packages. Older but well-structured.
- **AdamHavlicek/fastapi-todo-ddd** (⭐140) — Simpler: per-feature domains with separate ORM models. Good for understanding module decomposition.
- **alefeans/python-clean-architecture** — Production template using SQLModel. Simpler approach with FastAPI Depends. Good for smaller projects.