Clean Architecture in Python — Deep Dive

Implement Clean Architecture in a Python project with concrete layer boundaries, dependency injection, and testing strategies.

Building a Clean Architecture Python project

This guide walks through implementing Clean Architecture in a real Python application — an order management system. Every pattern shown has been used in production codebases ranging from 10k to 200k lines.

Layer 1: Domain entities

Entities are pure Python. No framework imports, no ORM base classes, no decorators from external libraries.

from dataclasses import dataclass, field
from decimal import Decimal
from datetime import datetime
from enum import Enum

class OrderStatus(Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    CANCELLED = "cancelled"

@dataclass
class OrderLine:
    product_id: str
    quantity: int
    unit_price: Decimal

    @property
    def total(self) -> Decimal:
        return self.unit_price * self.quantity

@dataclass
class Order:
    id: str
    customer_id: str
    lines: list[OrderLine] = field(default_factory=list)
    status: OrderStatus = OrderStatus.PENDING
    created_at: datetime = field(default_factory=datetime.utcnow)

    @property
    def total(self) -> Decimal:
        return sum(line.total for line in self.lines)

    def confirm(self) -> None:
        if self.status != OrderStatus.PENDING:
            raise ValueError(f"Cannot confirm order in {self.status} state")
        self.status = OrderStatus.CONFIRMED

    def cancel(self) -> None:
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("Cannot cancel a shipped order")
        self.status = OrderStatus.CANCELLED

Notice: no SQLAlchemy Base, no Pydantic BaseModel, no framework coupling. The Order class expresses business rules — what state transitions are valid — and nothing else.

Layer 2: Use case ports (interfaces)

Use cases define what they need from the outside world through protocols.

from typing import Protocol

class OrderRepository(Protocol):
    def get(self, order_id: str) -> Order | None: ...
    def save(self, order: Order) -> None: ...

class PaymentGateway(Protocol):
    def charge(self, customer_id: str, amount: Decimal) -> str: ...

class EventPublisher(Protocol):
    def publish(self, event_name: str, payload: dict) -> None: ...

These protocols form the boundary between the use case layer and the adapters. The use case imports and depends on these protocols, never on concrete implementations.

Layer 3: Use cases (application logic)

@dataclass
class PlaceOrderResult:
    order_id: str
    payment_ref: str

class PlaceOrderUseCase:
    def __init__(
        self,
        repo: OrderRepository,
        payment: PaymentGateway,
        events: EventPublisher,
    ) -> None:
        self._repo = repo
        self._payment = payment
        self._events = events

    def execute(self, order: Order) -> PlaceOrderResult:
        # Business rule: minimum order value
        if order.total < Decimal("10.00"):
            raise ValueError("Order total must be at least $10")

        payment_ref = self._payment.charge(
            order.customer_id, order.total
        )
        order.confirm()
        self._repo.save(order)

        self._events.publish("order.confirmed", {
            "order_id": order.id,
            "total": str(order.total),
        })

        return PlaceOrderResult(
            order_id=order.id,
            payment_ref=payment_ref,
        )

The use case orchestrates domain entities and infrastructure ports. It contains application-specific rules (minimum order value) but delegates domain rules to the entity (order.confirm()).

Layer 4: Adapters

Repository adapter (SQLAlchemy)

from sqlalchemy.orm import Session

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

    def get(self, order_id: str) -> Order | None:
        row = self._session.query(OrderModel).get(order_id)
        return self._to_domain(row) if row else None

    def save(self, order: Order) -> None:
        model = self._to_model(order)
        self._session.merge(model)
        self._session.commit()

    def _to_domain(self, model: OrderModel) -> Order:
        return Order(
            id=model.id,
            customer_id=model.customer_id,
            lines=[...],
            status=OrderStatus(model.status),
        )

    def _to_model(self, order: Order) -> OrderModel:
        return OrderModel(
            id=order.id,
            customer_id=order.customer_id,
            status=order.status.value,
        )

The mapper methods (_to_domain, _to_model) are the translation layer between ORM models and domain entities. This prevents SQLAlchemy details from leaking into domain code.

API adapter (FastAPI)

from fastapi import APIRouter, Depends

router = APIRouter()

@router.post("/orders")
def create_order(
    request: CreateOrderRequest,
    use_case: PlaceOrderUseCase = Depends(get_place_order_use_case),
):
    order = request.to_domain()
    result = use_case.execute(order)
    return {"order_id": result.order_id, "payment_ref": result.payment_ref}

FastAPI is confined to the outermost layer. If you switch to gRPC, you replace this adapter without touching use cases or entities.

Wiring it all together

The composition root — where all dependencies are assembled — lives at the application boundary.

def create_place_order_use_case(
    db: Session = Depends(get_db),
) -> PlaceOrderUseCase:
    return PlaceOrderUseCase(
        repo=SqlAlchemyOrderRepo(db),
        payment=StripePaymentGateway(settings.stripe_key),
        events=RabbitMQPublisher(settings.rabbitmq_url),
    )

This is the only place where concrete classes appear together. Changing a database or payment provider means editing this one function.

Testing strategy

Unit tests (domain + use cases)

No database, no HTTP, no external services.

class FakeOrderRepo:
    def __init__(self):
        self.saved: list[Order] = []

    def get(self, order_id: str) -> Order | None:
        return next((o for o in self.saved if o.id == order_id), None)

    def save(self, order: Order) -> None:
        self.saved.append(order)

class FakePayment:
    def charge(self, customer_id: str, amount: Decimal) -> str:
        return "fake-ref-123"

class FakeEvents:
    def __init__(self):
        self.published: list[tuple] = []

    def publish(self, event_name: str, payload: dict) -> None:
        self.published.append((event_name, payload))

def test_place_order_confirms_and_saves():
    repo = FakeOrderRepo()
    use_case = PlaceOrderUseCase(repo, FakePayment(), FakeEvents())
    order = Order(id="1", customer_id="c1", lines=[
        OrderLine("prod1", 2, Decimal("15.00"))
    ])

    result = use_case.execute(order)

    assert result.order_id == "1"
    assert repo.saved[0].status == OrderStatus.CONFIRMED

Integration tests (adapters)

Test that the SQLAlchemy adapter correctly maps to and from domain objects using a real (test) database. Use case logic is not retested here.

End-to-end tests

A thin layer that verifies the full HTTP path works. These are few and focused on wiring correctness, not business logic.

Enforcing layer boundaries

Use import linting to prevent inner layers from importing outer layers:

# pyproject.toml with import-linter
[tool.importlinter]
root_package = "src"

[[tool.importlinter.contracts]]
name = "Domain does not import adapters"
type = "forbidden"
source_modules = ["src.domain"]
forbidden_modules = ["src.adapters", "src.infrastructure"]

CI fails if anyone accidentally imports SQLAlchemy inside a domain entity.

Tradeoffs

  • Indirection cost: More files and more layers than a flat Django app. Worth it above ~5k lines; overkill for a weekend script.
  • Mapping overhead: Converting between ORM models and domain entities takes effort. Libraries like attrs mappers or manual converters help, but it is still code you maintain.
  • Team alignment: Everyone must agree on the layer rules. A single “shortcut” import from domain to infrastructure undermines the entire architecture.

When Clean Architecture is not the right choice

  • Small CRUD apps where the business logic is trivial — the ceremony outweighs the benefit.
  • Data science scripts that are exploratory and disposable.
  • Prototypes where speed to first demo matters more than long-term maintainability.

Start with Clean Architecture when you know the project will grow, when the domain is complex, or when you expect to swap infrastructure components.

The one thing to remember: Clean Architecture in Python is about enforcing the dependency rule — inner layers define contracts, outer layers fulfill them — so your business logic remains testable, portable, and framework-independent.

pythonarchitectureclean-code

See Also

  • Python Aggregate Pattern Why grouping related objects under a single gatekeeper prevents data chaos in your Python application.
  • Python Bounded Contexts Why the same word means different things in different parts of your code — and why that is perfectly fine.
  • Python Bulkhead Pattern Why smart Python apps put walls between their parts — like a ship that stays afloat even with a hole in the hull.
  • Python Circuit Breaker Pattern How a circuit breaker saves your app from crashing — explained with a home electrical fuse analogy.
  • Python Connection Draining How to shut down a Python server without hanging up on people mid-conversation — like a store that locks the entrance but lets shoppers finish.

Related Topics

ab-testing How tech companies run thousands of experiments at once to improve their products — the scientific method applied to everything from button colors to recommendation algorithms. 3 levels → activation-functions Why neural networks need these tiny mathematical functions — and how ReLU's simplicity accidentally made deep learning possible. 3 levels → ai-agents-architecture How AI systems go from answering questions to actually doing things — the design patterns that turn language models into autonomous agents that browse, code, and plan. 3 levels →