Clean Architecture - Dependencies Point Inward

Before we begin, study these two codebases. Both implement user registration. One will become a maintenance nightmare within six months. Which one, and why?

# Codebase A
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from passlib.hash import bcrypt

router = APIRouter()

@router.post("/register")
def register(email: str, password: str, db: Session = Depends(get_db)):
    if db.query(User).filter(User.email == email).first():
        raise HTTPException(400, "Email taken")
    hashed = bcrypt.hash(password)
    user = User(email=email, password=hashed)
    db.add(user)
    db.commit()
    send_welcome_email(email)  # direct SMTP call
    return {"id": user.id}

# Codebase B
from fastapi import APIRouter, Depends
from app.domain.services import UserRegistrationService
from app.dependencies import get_registration_service

router = APIRouter()

@router.post("/register")
def register(
    email: str,
    password: str,
    service: UserRegistrationService = Depends(get_registration_service),
):
    result = service.register(email, password)
    return {"id": result.user_id}

Codebase A tangles HTTP handling, database access, password hashing, and email sending into a single function. Changing your database, email provider, or hashing algorithm requires modifying endpoint code. Codebase B delegates to a service that knows nothing about FastAPI, SQLAlchemy, or SMTP. That separation is the core idea behind Clean Architecture.

What You Will Learn

• The four layers of Clean Architecture and the dependency rule that governs them
• How to model domain entities that carry no framework dependencies
• Building a service layer (use cases) that orchestrates business logic
• Repository and gateway patterns as interface adapters
• Dependency inversion using Python's Protocol classes
• Testing each layer in isolation without spinning up databases or servers
• When Clean Architecture is worth the structural overhead and when it is not

Prerequisites

• Solid understanding of Python classes, dataclasses, and inheritance
• Familiarity with typing.Protocol and abstract base classes
• Experience with FastAPI and SQLAlchemy (from Intermediate course)
• Basic knowledge of SOLID principles, especially Dependency Inversion

Part 1 - The Dependency Rule

Uncle Bob's Clean Architecture organizes code into concentric layers. The single most important rule: source code dependencies must point inward. Inner layers know nothing about outer layers.

The Four Layers

Layer	Contains	Depends On	Example
Entities	Business objects, validation rules	Nothing	User, Order, Money
Use Cases	Application-specific business rules	Entities	RegisterUser, PlaceOrder
Interface Adapters	Converters between use cases and external formats	Use Cases, Entities	Repositories, Controllers, Presenters
Frameworks & Drivers	External tools and delivery mechanisms	Everything above	FastAPI, SQLAlchemy, Redis, SMTP

Why This Ordering Matters

Consider what happens when you upgrade from SQLAlchemy 1.4 to 2.0. In a tangled codebase, that upgrade touches every file that queries data. In Clean Architecture, the change is confined to the outermost layer - your repository implementations. Your domain logic, use cases, and even your controller logic remain untouched.

# The dependency rule in practice:
# domain/entities.py - depends on NOTHING external
from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID, uuid4

@dataclass
class User:
    email: str
    hashed_password: str
    id: UUID = field(default_factory=uuid4)
    created_at: datetime = field(default_factory=datetime.utcnow)
    is_active: bool = True

    def deactivate(self) -> None:
        self.is_active = False

    def validate_email(self) -> bool:
        return "@" in self.email and "." in self.email.split("@")[1]

:::danger The Cardinal Sin If your User entity imports from sqlalchemy import Column, the dependency rule is broken. Domain entities must never depend on frameworks. SQLAlchemy models belong in the adapters layer and map to domain entities. :::

Part 2 - Domain Entities: The Inner Core

Domain entities encode business rules that are true regardless of which framework delivers them. They are the most stable part of your system.

Designing Rich Entities

Entities are not just data holders. They encapsulate behavior and enforce invariants.

# domain/entities.py
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Optional
from uuid import UUID, uuid4


class Currency(Enum):
    USD = "USD"
    EUR = "EUR"
    GBP = "GBP"


@dataclass
class Money:
    amount: Decimal
    currency: Currency

    def __post_init__(self) -> None:
        if self.amount < 0:
            raise ValueError("Money amount cannot be negative")
        # Enforce 2 decimal places
        self.amount = self.amount.quantize(Decimal("0.01"))

    def __add__(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError(
                f"Cannot add {self.currency.value} and {other.currency.value}"
            )
        return Money(self.amount + other.amount, self.currency)

    def __sub__(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError(
                f"Cannot subtract {other.currency.value} from {self.currency.value}"
            )
        result = self.amount - other.amount
        if result < 0:
            raise ValueError("Subtraction would result in negative money")
        return Money(result, self.currency)


@dataclass
class Account:
    owner_id: UUID
    balance: Money
    id: UUID = field(default_factory=uuid4)
    created_at: datetime = field(default_factory=datetime.utcnow)
    is_frozen: bool = False

    def deposit(self, amount: Money) -> None:
        if self.is_frozen:
            raise PermissionError("Cannot deposit to a frozen account")
        self.balance = self.balance + amount

    def withdraw(self, amount: Money) -> None:
        if self.is_frozen:
            raise PermissionError("Cannot withdraw from a frozen account")
        self.balance = self.balance - amount  # raises if insufficient

    def freeze(self) -> None:
        self.is_frozen = True

    def unfreeze(self) -> None:
        self.is_frozen = False


@dataclass
class Transfer:
    from_account_id: UUID
    to_account_id: UUID
    amount: Money
    id: UUID = field(default_factory=uuid4)
    executed_at: Optional[datetime] = None
    status: str = "pending"

    def mark_completed(self) -> None:
        self.status = "completed"
        self.executed_at = datetime.utcnow()

    def mark_failed(self, reason: str) -> None:
        self.status = f"failed: {reason}"

Notice: no imports from SQLAlchemy, FastAPI, Pydantic, or any external library. These entities work in a plain Python script, a CLI tool, a web app, or a batch job.

Value Objects vs Entities

Money is a value object - defined by its attributes, not by identity. Two Money(Decimal("10.00"), Currency.USD) instances are interchangeable. Account is an entity - defined by its id. Two accounts with the same balance are distinct objects.

# Value objects: equality by value
m1 = Money(Decimal("10.00"), Currency.USD)
m2 = Money(Decimal("10.00"), Currency.USD)
# m1 == m2  ->  True (with @dataclass eq=True)

# Entities: equality by identity
a1 = Account(owner_id=uuid4(), balance=m1)
a2 = Account(owner_id=uuid4(), balance=m1)
# a1 == a2  ->  False (different ids)

:::tip When to Use Dataclasses vs Pydantic Use plain dataclass for domain entities - they carry no framework dependency. Use Pydantic BaseModel in the interface adapters layer for request/response validation. Map between them at the boundary. :::

Part 3 - Use Cases: Application Business Rules

Use cases orchestrate domain entities to fulfill a specific application need. They express what the system does without knowing how external systems work.

Defining Ports with Protocol

Use cases depend on abstractions (ports), not concrete implementations. In Python, Protocol classes define these ports.

# domain/ports.py
from typing import Protocol, Optional
from uuid import UUID
from domain.entities import Account, Transfer, User


class UserRepository(Protocol):
    def get_by_email(self, email: str) -> Optional[User]: ...
    def get_by_id(self, user_id: UUID) -> Optional[User]: ...
    def save(self, user: User) -> None: ...


class AccountRepository(Protocol):
    def get_by_id(self, account_id: UUID) -> Optional[Account]: ...
    def get_by_owner(self, owner_id: UUID) -> list[Account]: ...
    def save(self, account: Account) -> None: ...


class TransferRepository(Protocol):
    def save(self, transfer: Transfer) -> None: ...


class PasswordHasher(Protocol):
    def hash(self, password: str) -> str: ...
    def verify(self, password: str, hashed: str) -> bool: ...


class EmailSender(Protocol):
    def send_welcome(self, to: str, username: str) -> None: ...
    def send_transfer_notification(
        self, to: str, amount: str, direction: str
    ) -> None: ...


class UnitOfWork(Protocol):
    """Ensures atomicity across multiple repository operations."""
    def __enter__(self) -> "UnitOfWork": ...
    def __exit__(self, *args) -> None: ...
    def commit(self) -> None: ...
    def rollback(self) -> None: ...

Implementing a Use Case

# application/use_cases.py
from dataclasses import dataclass
from decimal import Decimal
from uuid import UUID

from domain.entities import Account, Money, Transfer, User, Currency
from domain.ports import (
    AccountRepository,
    EmailSender,
    PasswordHasher,
    TransferRepository,
    UnitOfWork,
    UserRepository,
)


@dataclass
class RegisterUserResult:
    user_id: UUID
    email: str


class RegisterUser:
    """Use case: register a new user and create their default account."""

    def __init__(
        self,
        user_repo: UserRepository,
        account_repo: AccountRepository,
        hasher: PasswordHasher,
        email_sender: EmailSender,
        uow: UnitOfWork,
    ) -> None:
        self._user_repo = user_repo
        self._account_repo = account_repo
        self._hasher = hasher
        self._email_sender = email_sender
        self._uow = uow

    def execute(self, email: str, password: str) -> RegisterUserResult:
        # Business rule: no duplicate emails
        existing = self._user_repo.get_by_email(email)
        if existing is not None:
            raise ValueError(f"User with email {email} already exists")

        # Business rule: password minimum length
        if len(password) < 8:
            raise ValueError("Password must be at least 8 characters")

        hashed = self._hasher.hash(password)
        user = User(email=email, hashed_password=hashed)

        default_account = Account(
            owner_id=user.id,
            balance=Money(Decimal("0.00"), Currency.USD),
        )

        with self._uow:
            self._user_repo.save(user)
            self._account_repo.save(default_account)
            self._uow.commit()

        # Side effect: send email (outside transaction - it's okay if this fails)
        self._email_sender.send_welcome(email, email.split("@")[0])

        return RegisterUserResult(user_id=user.id, email=user.email)


class TransferFunds:
    """Use case: transfer money between two accounts."""

    def __init__(
        self,
        account_repo: AccountRepository,
        transfer_repo: TransferRepository,
        email_sender: EmailSender,
        uow: UnitOfWork,
    ) -> None:
        self._account_repo = account_repo
        self._transfer_repo = transfer_repo
        self._email_sender = email_sender
        self._uow = uow

    def execute(
        self, from_id: UUID, to_id: UUID, amount: Decimal, currency: Currency
    ) -> Transfer:
        from_account = self._account_repo.get_by_id(from_id)
        to_account = self._account_repo.get_by_id(to_id)

        if from_account is None or to_account is None:
            raise ValueError("One or both accounts not found")

        money = Money(amount, currency)
        transfer = Transfer(
            from_account_id=from_id,
            to_account_id=to_id,
            amount=money,
        )

        try:
            with self._uow:
                from_account.withdraw(money)
                to_account.deposit(money)
                transfer.mark_completed()

                self._account_repo.save(from_account)
                self._account_repo.save(to_account)
                self._transfer_repo.save(transfer)
                self._uow.commit()
        except (ValueError, PermissionError) as e:
            transfer.mark_failed(str(e))
            raise

        return transfer

:::note Use Cases Are Single-Purpose Each use case class handles exactly one operation. RegisterUser does not also handle login. TransferFunds does not also create accounts. This keeps them testable and composable. :::

Part 4 - Interface Adapters: Bridging the Gap

Interface adapters translate between the format used by use cases/entities and the format expected by external systems (databases, HTTP, CLI).

Repository Implementations

# adapters/repositories.py
from typing import Optional
from uuid import UUID

from sqlalchemy.orm import Session

from domain.entities import Account, Money, Currency, User
from domain.ports import AccountRepository, UserRepository


# SQLAlchemy ORM model (lives in adapters, NOT in domain)
from adapters.orm_models import UserModel, AccountModel


class SqlAlchemyUserRepository:
    """Adapter: translates between domain User and SQLAlchemy UserModel."""

    def __init__(self, session: Session) -> None:
        self._session = session

    def get_by_email(self, email: str) -> Optional[User]:
        row = (
            self._session.query(UserModel)
            .filter(UserModel.email == email)
            .first()
        )
        if row is None:
            return None
        return self._to_entity(row)

    def get_by_id(self, user_id: UUID) -> Optional[User]:
        row = self._session.get(UserModel, str(user_id))
        if row is None:
            return None
        return self._to_entity(row)

    def save(self, user: User) -> None:
        model = UserModel(
            id=str(user.id),
            email=user.email,
            hashed_password=user.hashed_password,
            is_active=user.is_active,
            created_at=user.created_at,
        )
        self._session.merge(model)

    @staticmethod
    def _to_entity(row: UserModel) -> User:
        return User(
            email=row.email,
            hashed_password=row.hashed_password,
            id=UUID(row.id),
            created_at=row.created_at,
            is_active=row.is_active,
        )


class SqlAlchemyAccountRepository:
    def __init__(self, session: Session) -> None:
        self._session = session

    def get_by_id(self, account_id: UUID) -> Optional[Account]:
        row = self._session.get(AccountModel, str(account_id))
        if row is None:
            return None
        return Account(
            owner_id=UUID(row.owner_id),
            balance=Money(row.balance_amount, Currency(row.balance_currency)),
            id=UUID(row.id),
            created_at=row.created_at,
            is_frozen=row.is_frozen,
        )

    def get_by_owner(self, owner_id: UUID) -> list[Account]:
        rows = (
            self._session.query(AccountModel)
            .filter(AccountModel.owner_id == str(owner_id))
            .all()
        )
        return [
            Account(
                owner_id=UUID(r.owner_id),
                balance=Money(r.balance_amount, Currency(r.balance_currency)),
                id=UUID(r.id),
                created_at=r.created_at,
                is_frozen=r.is_frozen,
            )
            for r in rows
        ]

    def save(self, account: Account) -> None:
        model = AccountModel(
            id=str(account.id),
            owner_id=str(account.owner_id),
            balance_amount=account.balance.amount,
            balance_currency=account.balance.currency.value,
            is_frozen=account.is_frozen,
            created_at=account.created_at,
        )
        self._session.merge(model)

ORM Models (Framework Layer)

# adapters/orm_models.py
from sqlalchemy import Column, String, Boolean, DateTime, Numeric
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class UserModel(Base):
    __tablename__ = "users"

    id = Column(String, primary_key=True)
    email = Column(String, unique=True, nullable=False, index=True)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, nullable=False)


class AccountModel(Base):
    __tablename__ = "accounts"

    id = Column(String, primary_key=True)
    owner_id = Column(String, nullable=False, index=True)
    balance_amount = Column(Numeric(12, 2), nullable=False)
    balance_currency = Column(String(3), nullable=False)
    is_frozen = Column(Boolean, default=False)
    created_at = Column(DateTime, nullable=False)

Infrastructure Adapters

# adapters/security.py
from passlib.hash import bcrypt


class BcryptPasswordHasher:
    def hash(self, password: str) -> str:
        return bcrypt.hash(password)

    def verify(self, password: str, hashed: str) -> bool:
        return bcrypt.verify(password, hashed)


# adapters/notifications.py
import smtplib
from email.mime.text import MIMEText


class SmtpEmailSender:
    def __init__(self, host: str, port: int, username: str, password: str) -> None:
        self._host = host
        self._port = port
        self._username = username
        self._password = password

    def send_welcome(self, to: str, username: str) -> None:
        msg = MIMEText(f"Welcome, {username}!")
        msg["Subject"] = "Welcome to our platform"
        msg["From"] = self._username
        msg["To"] = to
        self._send(msg)

    def send_transfer_notification(
        self, to: str, amount: str, direction: str
    ) -> None:
        msg = MIMEText(f"Transfer {direction}: {amount}")
        msg["Subject"] = f"Transfer {direction}"
        msg["From"] = self._username
        msg["To"] = to
        self._send(msg)

    def _send(self, msg: MIMEText) -> None:
        with smtplib.SMTP(self._host, self._port) as server:
            server.starttls()
            server.login(self._username, self._password)
            server.send_message(msg)

Part 5 - The Full Project Structure

The FastAPI Layer (Outermost)

# frameworks/fastapi_app.py
from decimal import Decimal
from uuid import UUID

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, EmailStr

from application.use_cases import RegisterUser, TransferFunds
from domain.entities import Currency
from frameworks.dependencies import get_register_user, get_transfer_funds

router = APIRouter()


class RegisterRequest(BaseModel):
    email: EmailStr
    password: str


class RegisterResponse(BaseModel):
    user_id: UUID
    email: str


class TransferRequest(BaseModel):
    from_account_id: UUID
    to_account_id: UUID
    amount: Decimal
    currency: str


@router.post("/register", response_model=RegisterResponse)
def register(
    body: RegisterRequest,
    use_case: RegisterUser = Depends(get_register_user),
):
    try:
        result = use_case.execute(body.email, body.password)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    return RegisterResponse(user_id=result.user_id, email=result.email)


@router.post("/transfer")
def transfer(
    body: TransferRequest,
    use_case: TransferFunds = Depends(get_transfer_funds),
):
    try:
        result = use_case.execute(
            body.from_account_id,
            body.to_account_id,
            body.amount,
            Currency(body.currency),
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    return {"transfer_id": str(result.id), "status": result.status}

# frameworks/dependencies.py
from sqlalchemy.orm import Session

from adapters.notifications import SmtpEmailSender
from adapters.repositories import SqlAlchemyAccountRepository, SqlAlchemyUserRepository
from adapters.security import BcryptPasswordHasher
from adapters.unit_of_work import SqlAlchemyUnitOfWork
from application.use_cases import RegisterUser, TransferFunds


def get_db_session() -> Session:
    # Session factory - configured at app startup
    ...


def get_register_user(session: Session = Depends(get_db_session)) -> RegisterUser:
    return RegisterUser(
        user_repo=SqlAlchemyUserRepository(session),
        account_repo=SqlAlchemyAccountRepository(session),
        hasher=BcryptPasswordHasher(),
        email_sender=SmtpEmailSender("smtp.example.com", 587, "noreply", "secret"),
        uow=SqlAlchemyUnitOfWork(session),
    )

:::tip Pydantic Models Are NOT Domain Entities RegisterRequest and RegisterResponse are Pydantic models in the frameworks layer. They handle validation and serialization for the HTTP boundary. Domain entities are plain dataclasses that never see HTTP. :::

Part 6 - Testing Each Layer Independently

The power of Clean Architecture is that every layer can be tested without its outer layers.

Testing Domain Entities (No Mocks Needed)

# tests/test_entities.py
from decimal import Decimal
import pytest
from domain.entities import Account, Money, Currency

def test_deposit_increases_balance():
    account = Account(
        owner_id=uuid4(),
        balance=Money(Decimal("100.00"), Currency.USD),
    )
    account.deposit(Money(Decimal("50.00"), Currency.USD))
    assert account.balance.amount == Decimal("150.00")

def test_withdraw_insufficient_funds_raises():
    account = Account(
        owner_id=uuid4(),
        balance=Money(Decimal("10.00"), Currency.USD),
    )
    with pytest.raises(ValueError, match="negative"):
        account.withdraw(Money(Decimal("50.00"), Currency.USD))

def test_frozen_account_cannot_deposit():
    account = Account(
        owner_id=uuid4(),
        balance=Money(Decimal("100.00"), Currency.USD),
    )
    account.freeze()
    with pytest.raises(PermissionError):
        account.deposit(Money(Decimal("50.00"), Currency.USD))

def test_cannot_add_different_currencies():
    usd = Money(Decimal("10.00"), Currency.USD)
    eur = Money(Decimal("10.00"), Currency.EUR)
    with pytest.raises(ValueError, match="Cannot add"):
        usd + eur

Testing Use Cases (With Fakes)

# tests/fakes.py
from typing import Optional
from uuid import UUID
from domain.entities import User
from domain.ports import UserRepository


class FakeUserRepository:
    """In-memory implementation - satisfies the Protocol without a database."""

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

    def get_by_email(self, email: str) -> Optional[User]:
        return self._users.get(email)

    def get_by_id(self, user_id: UUID) -> Optional[User]:
        for user in self._users.values():
            if user.id == user_id:
                return user
        return None

    def save(self, user: User) -> None:
        self._users[user.email] = user


class FakePasswordHasher:
    def hash(self, password: str) -> str:
        return f"hashed_{password}"

    def verify(self, password: str, hashed: str) -> bool:
        return hashed == f"hashed_{password}"


class FakeEmailSender:
    def __init__(self) -> None:
        self.sent: list[tuple[str, str]] = []

    def send_welcome(self, to: str, username: str) -> None:
        self.sent.append((to, username))

    def send_transfer_notification(self, to: str, amount: str, direction: str) -> None:
        self.sent.append((to, f"{direction}:{amount}"))


class FakeUnitOfWork:
    def __init__(self) -> None:
        self.committed = False
        self.rolled_back = False

    def __enter__(self):
        return self

    def __exit__(self, *args):
        pass

    def commit(self):
        self.committed = True

    def rollback(self):
        self.rolled_back = True

# tests/test_use_cases.py
import pytest
from application.use_cases import RegisterUser
from tests.fakes import (
    FakeAccountRepository,
    FakeEmailSender,
    FakePasswordHasher,
    FakeUnitOfWork,
    FakeUserRepository,
)


def make_register_use_case():
    return RegisterUser(
        user_repo=FakeUserRepository(),
        account_repo=FakeAccountRepository(),
        hasher=FakePasswordHasher(),
        email_sender=FakeEmailSender(),
        uow=FakeUnitOfWork(),
    )


def test_register_creates_user():
    uc = make_register_use_case()
    result = uc.execute("[email protected]", "securepass")
    assert result.email == "[email protected]"


def test_register_rejects_duplicate_email():
    uc = make_register_use_case()
    uc.execute("[email protected]", "securepass")
    with pytest.raises(ValueError, match="already exists"):
        uc.execute("[email protected]", "otherpass")


def test_register_rejects_short_password():
    uc = make_register_use_case()
    with pytest.raises(ValueError, match="at least 8"):
        uc.execute("[email protected]", "short")


def test_register_sends_welcome_email():
    email_sender = FakeEmailSender()
    uc = RegisterUser(
        user_repo=FakeUserRepository(),
        account_repo=FakeAccountRepository(),
        hasher=FakePasswordHasher(),
        email_sender=email_sender,
        uow=FakeUnitOfWork(),
    )
    uc.execute("[email protected]", "securepass")
    assert len(email_sender.sent) == 1
    assert email_sender.sent[0][0] == "[email protected]"

No database. No HTTP server. No SMTP server. Tests run in milliseconds.

Part 7 - When Clean Architecture Is Worth It

Clean Architecture adds structural complexity. Not every project needs it.

When It Helps

Scenario	Why Clean Architecture Pays Off
Long-lived systems (years)	Frameworks change; domain logic stays
Multiple delivery mechanisms (API + CLI + worker)	Use cases are reusable across all
Complex business rules	Domain layer keeps them testable
Teams > 3 engineers	Clear boundaries reduce merge conflicts
Regulatory requirements	Domain logic is auditable and framework-independent

When It Is Overkill

Scenario	Better Alternative
CRUD-heavy app with little logic	Simple layered architecture (router → service → ORM)
Prototype or MVP	Direct framework usage; refactor later
Script or CLI tool	Single module is fine
Solo developer, small scope	Pragmatic layers without full ceremony

:::danger The Abstraction Trap Do not create ports and adapters for things that will never change. If you will always use PostgreSQL, a repository interface over SQLAlchemy still has value for testing - but a full adapter over your logging library is probably wasted effort. :::

The Pragmatic Middle Ground

You do not need to implement all four layers from day one. Start with two separations:

1. Domain logic has no framework imports - this is non-negotiable.
2. Use cases accept dependencies via constructor - this enables testing.

Add the full adapter layer when you actually need to swap implementations or support multiple delivery mechanisms.

Key Takeaways

• The dependency rule is the entire architecture: source code dependencies always point inward, from frameworks toward domain entities.
• Domain entities are plain Python: no ORM columns, no Pydantic validators, no framework decorators. They encode business rules that are true everywhere.
• Use cases are single-purpose orchestrators: they coordinate domain entities through ports (Protocol interfaces), never through concrete implementations.
• Interface adapters translate between worlds: repositories convert ORM rows to domain entities; controllers convert HTTP requests to use case inputs.
• Testing improves dramatically: domain entities need no mocks, use cases need only in-memory fakes, and adapter tests are focused on translation logic.
• Start simple, grow into it: extract domain logic first, add the full adapter layer when you genuinely need it. Premature architecture is as harmful as no architecture.

Graded Practice Challenges

Level 1 - Identify the Violation

Question 1: What dependency rule violation exists in this entity?

from sqlalchemy import Column, String
from sqlalchemy.orm import declarative_base
from pydantic import validator

Base = declarative_base()

class Product(Base):
    __tablename__ = "products"
    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)

    @validator("name")
    def name_must_not_be_empty(cls, v):
        if not v:
            raise ValueError("Name required")
        return v

Answer

This class depends on both SQLAlchemy (Base, Column) and Pydantic (validator). A domain entity should depend on neither. The SQLAlchemy model belongs in the adapters layer. The validation logic belongs in a plain dataclass entity. Additionally, @validator is a Pydantic v1 construct that does not belong in a SQLAlchemy model at all - these are two separate concerns mixed into one class.

Question 2: This use case directly instantiates a concrete repository. Why is this a problem?

class CreateOrder:
    def execute(self, items: list[dict]) -> Order:
        repo = PostgresOrderRepository("postgresql://localhost/db")
        order = Order(items=items)
        repo.save(order)
        return order

Answer

The use case creates PostgresOrderRepository directly, violating the Dependency Inversion Principle. The use case now depends on PostgreSQL - it cannot be tested without a running database, and switching to a different storage backend requires modifying use case code. The repository should be injected via the constructor and typed as a Protocol.

Question 3: Which layer should contain Pydantic request/response models for a REST API?

Answer

Pydantic request/response models belong in the Interface Adapters or Frameworks & Drivers layer (outermost layers). They handle serialization and validation for the HTTP boundary. Domain entities in the innermost layer should be plain dataclasses with no Pydantic dependency.

Level 2 - Refactoring Challenge

Take this tangled endpoint and refactor it into Clean Architecture layers:

@router.post("/orders")
def create_order(items: list[dict], db: Session = Depends(get_db)):
    total = sum(item["price"] * item["qty"] for item in items)
    if total > 10000:
        raise HTTPException(400, "Order exceeds limit")
    if any(item["qty"] < 1 for item in items):
        raise HTTPException(400, "Invalid quantity")
    order = OrderModel(items=json.dumps(items), total=total, status="pending")
    db.add(order)
    db.commit()
    requests.post("https://slack.com/webhook", json={"text": f"New order: ${total}"})
    return {"id": order.id}

Produce: (a) a domain Order entity with validation, (b) a CreateOrder use case with injected ports, (c) an adapter for the Slack notification, (d) a thin FastAPI endpoint.

Level 3 - Design Challenge

Design the Clean Architecture layer structure for an e-learning platform (like EngineersOfAI) that has:

• Courses with lessons and quizzes
• User enrollment and progress tracking
• Certificate generation upon completion
• Payment processing (Stripe)
• Email notifications

Produce: (a) the domain entities with their business rules, (b) the port interfaces, (c) the use case list with their dependencies, (d) a directory structure diagram. Explain which adapters you would build first and which you would defer.

What's Next

In the next lesson, Hexagonal Architecture (Ports and Adapters), we will explore a closely related pattern that focuses specifically on how your application communicates with the outside world - and how to make those communication channels swappable, testable, and explicit.