Clean Architecture & Dependency Injection in Large-Scale Python Systems

Building scalable backend systems requires decoupling domain business logic from database frameworks (ORMs), message queues, and Web APIs.

In this session, we implement Hexagonal / Clean Architecture in Python using Dataclasses, Protocols, and Dependency Injection.

1. Domain Entities & Value Objects

The core domain contains pure Python data classes without any dependency on SQLAlchemy, Pydantic, or web frameworks.

from dataclasses import dataclass, field
from datetime import datetime
from typing import Protocol, Optional

@dataclass
class OrderItem:
    product_id: str
    quantity: int
    unit_price: float

    @property
    def subtotal(self) -> float:
        return self.quantity * self.unit_price

@dataclass
class Order:
    id: str
    customer_id: str
    items: list[OrderItem] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.utcnow)

    def calculate_total(self) -> float:
        return sum(item.subtotal for item in self.items)

2. Port Definitions (Repository Protocols)

Ports are abstract interfaces defined using typing.Protocol.

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

    def get_by_id(self, order_id: str) -> Optional[Order]:
        ...

3. Application Service Layer

The service orchestrates domain operations and interacts strictly with repository protocols:

class CreateOrderUseCase:
    def __init__(self, repo: OrderRepository):
        self.repo = repo

    def execute(self, order_id: str, customer_id: str, items_data: list[dict]) -> Order:
        items = [
            OrderItem(
                product_id=item["product_id"],
                quantity=item["quantity"],
                unit_price=item["unit_price"]
            )
            for item in items_data
        ]
        order = Order(id=order_id, customer_id=customer_id, items=items)
        self.repo.save(order)
        return order

4. InMemory Adapter for Unit Testing

Because the service relies on Protocols, writing unit tests is lightning fast without needing a database:

class InMemoryOrderRepository:
    def __init__(self):
        self._db: dict[str, Order] = {}

    def save(self, order: Order) -> None:
        self._db[order.id] = order

    def get_by_id(self, order_id: str) -> Optional[Order]:
        return self._db.get(order_id)

# Test execution
def test_create_order():
    repo = InMemoryOrderRepository()
    use_case = CreateOrderUseCase(repo)
    
    order = use_case.execute("ORD-001", "CUST-99", [{"product_id": "P1", "quantity": 2, "unit_price": 50.0}])
    assert order.calculate_total() == 100.0
    print("Test passed: Total is", order.calculate_total())

if __name__ == "__main__":
    test_create_order()

Benefits of this Approach

1. Fast Unit Tests: Core business logic tests run in milliseconds without DB setups.
2. Framework Flexibility: Upgrade or replace ORMs and frameworks without touching business rules.
3. Clear Code Ownership: Domain rules stay explicit and concentrated.