Python Clean Architecture Practice Problems & Exercises

Practice: Clean Architecture

11 problems3 Easy4 Medium4 Hard⏱ 60–90 min

← Back to lesson

---

Easy

#1Layer ClassifierEasy

layersclean-architecturedependency-rule

Write a classify_layer(component_name) function that returns which Clean Architecture layer a component belongs to based on naming conventions.

def classify_layer(name):
    # Return one of: "domain", "application", "interface", "infrastructure"
    pass

components = [
    "UserRepository", "CreateUserUseCase", "User",
    "FastAPIRouter", "EmailSender", "UserService",
]
for c in components:
    print(f"{c}: {classify_layer(c)}")

Solution

def classify_layer(name):
    infra_signals = ["Repository", "Sender", "Client", "DB", "Cache", "Queue", "Adapter"]
    app_signals   = ["UseCase", "Service", "Handler", "Command", "Query"]
    iface_signals = ["Router", "Controller", "Presenter", "View", "Serializer", "DTO"]

    for sig in infra_signals:
        if sig in name:
            return "infrastructure"
    for sig in app_signals:
        if sig in name:
            return "application"
    for sig in iface_signals:
        if sig in name:
            return "interface"
    return "domain"

components = [
    "UserRepository", "CreateUserUseCase", "User",
    "FastAPIRouter", "EmailSender", "UserService",
]
for c in components:
    print(f"{c}: {classify_layer(c)}")

The four layers:

• Domain: Pure business objects — User, Order, Product. No imports from outer layers.
• Application: Orchestrates domain objects to fulfil use cases — CreateUserUseCase, UserService.
• Interface/Adapters: Translates between the app and the outside world — FastAPIRouter, UserDTO.
• Infrastructure: Concrete implementations of ports — UserRepository (SQL), EmailSender (SMTP).

Expected Output

UserRepository: infrastructure\nCreateUserUseCase: application\nUser: domain\nFastAPIRouter: interface\nEmailSender: infrastructure\nUserService: application

Hints

Hint 1: Clean Architecture has four concentric layers: domain (entities), application (use cases), interface/adapters, and infrastructure (DB, HTTP, email, etc.).

Hint 2: The dependency rule says outer layers can depend on inner layers, never the reverse. Domain knows nothing about infrastructure.

#2Pure Domain EntityEasy

domainentityvalue-object

Implement a pure Money value object that belongs in the domain layer. It must have no external dependencies and enforce business rules.

Python

from dataclasses import dataclass

@dataclass(frozen=True)
class Money:
    amount: int   # in cents
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Amount cannot be negative")
        if len(self.currency) != 3:
            raise ValueError("Currency must be a 3-letter ISO code")

    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Cannot add different currencies")
        return Money(self.amount + other.amount, self.currency)

    def __str__(self):
        return f"Money({self.amount}, {self.currency})"

# Tests
m1 = Money(100, "USD")
m2 = Money(50, "USD")
m3 = Money(100, "EUR")

print(m1)
print(m1.add(m2))
print(m1 == Money(100, "USD"))
try:
    m1.add(m3)
except ValueError as e:
    print(e)

Solution

from dataclasses import dataclass

@dataclass(frozen=True)
class Money:
    amount: int
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Amount cannot be negative")
        if len(self.currency) != 3:
            raise ValueError("Currency must be a 3-letter ISO code")

    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Cannot add different currencies")
        return Money(self.amount + other.amount, self.currency)

    def __str__(self):
        return f"Money({self.amount}, {self.currency})"

Why frozen=True? Value objects must be immutable — equality is defined by value, not identity. frozen=True makes the dataclass hashable and prevents mutation, which is exactly the right default for domain value objects.

Domain layer rules:

• No import sqlalchemy, import fastapi, import requests — ever.
• Business invariants are enforced in __post_init__ or factory methods, not in a database constraint.
• Returning a new Money instance from add() is correct — the domain object is immutable.

Expected Output

Money(100, USD)\nMoney(150, USD)\nTrue\nCannot add different currencies

Hints

Hint 1: A domain entity lives in the innermost layer. It has NO imports from frameworks, databases, or HTTP libraries.

Hint 2: Value objects are immutable and compared by value rather than identity. Use __eq__ and __hash__.

#3Spot the Dependency Rule ViolationEasy

dependency-ruleviolationclean-architecture

Identify all violations of the Clean Architecture dependency rule in the code below. Explain each violation and how to fix it.

# domain/user.py
import sqlalchemy  # <-- ???

class User:
    def __init__(self, user_id: int, email: str):
        self.id = user_id
        self.email = email

    def save(self):
        # Saves to DB directly
        engine = sqlalchemy.create_engine("postgresql://...")
        with engine.connect() as conn:
            conn.execute("INSERT INTO users ...")


# application/create_user.py
from domain.user import User
import smtplib  # <-- ???

class CreateUserUseCase:
    def execute(self, email: str):
        user = User(1, email)
        user.save()
        # Send welcome email directly
        smtp = smtplib.SMTP("localhost")
        smtp.sendmail("[email protected]", email, "Welcome!")

Solution

Violation 1: domain/user.py imports sqlalchemy

The domain layer is the innermost circle. It must know nothing about persistence. Having User.save() call a database means the domain is tightly coupled to infrastructure. If you switch from PostgreSQL to DynamoDB, the domain entity breaks.

Fix: Remove persistence from the entity entirely. Move it to an IUserRepository port (interface in the domain) implemented by a SqlUserRepository in the infrastructure layer.

Violation 2: application/create_user.py imports smtplib

The application layer should depend on an abstraction of email sending, not on SMTP directly. Sending email is an infrastructure concern — the concrete transport (SMTP, SendGrid, SES) is an implementation detail.

Fix: Define a IEmailNotifier protocol in the application or domain layer. Inject a concrete SmtpEmailNotifier at the composition root.

Corrected structure:

# domain/user.py  — no imports from outer layers
class User:
    def __init__(self, user_id: int, email: str):
        self.id = user_id
        self.email = email

# domain/ports.py
from typing import Protocol
class IUserRepository(Protocol):
    def save(self, user: "User") -> None: ...

class IEmailNotifier(Protocol):
    def send_welcome(self, email: str) -> None: ...

# application/create_user.py
from domain.user import User
from domain.ports import IUserRepository, IEmailNotifier

class CreateUserUseCase:
    def __init__(self, repo: IUserRepository, notifier: IEmailNotifier):
        self.repo = repo
        self.notifier = notifier

    def execute(self, email: str) -> User:
        user = User(1, email)
        self.repo.save(user)
        self.notifier.send_welcome(email)
        return user

Expected Output

See solution for analysis

Hints

Hint 1: The dependency rule: source code dependencies can only point INWARD. A domain entity importing from a database driver violates the rule.

Hint 2: Look for imports in inner-layer classes that reference outer-layer modules.

---

Medium

#4Repository Pattern with ProtocolMedium

repositoryprotocoldependency-inversion

Implement the Repository Pattern: a IUserRepository Protocol, an in-memory implementation, and a CreateUserUseCase that depends only on the Protocol.

Python

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

@dataclass
class User:
    id: int
    email: str

    def __str__(self):
        return f"User(id={self.id}, email={self.email})"


class IUserRepository(Protocol):
    def save(self, user: User) -> None: ...
    def find_by_id(self, user_id: int) -> Optional[User]: ...
    def find_all(self) -> list[User]: ...


@dataclass
class InMemoryUserRepository:
    _store: dict = field(default_factory=dict)

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

    def find_by_id(self, user_id: int) -> Optional[User]:
        return self._store.get(user_id)

    def find_all(self) -> list[User]:
        return list(self._store.values())


class CreateUserUseCase:
    def __init__(self, repo: IUserRepository):
        self._repo = repo
        self._next_id = 1

    def execute(self, email: str) -> User:
        user = User(id=self._next_id, email=email)
        self._next_id += 1
        self._repo.save(user)
        return user


# Composition root — wire it all together
repo = InMemoryUserRepository()
use_case = CreateUserUseCase(repo)

u1 = use_case.execute("[email protected]")
use_case.execute("[email protected]")

print(f"Created: {u1}")
print(f"Found: {repo.find_by_id(1)}")
print(f"All users: {len(repo.find_all())}")

Solution

The code above is already complete and runnable. The key architectural insights:

Why Protocol over ABC?

• Protocol enables structural subtyping — any class with the right methods satisfies the protocol without explicit inheritance.
• This makes it easier to swap implementations (in-memory, SQL, Redis) without changing any use case code.
• You can retrofit existing classes as repositories without modifying them (open/closed principle).

The composition root pattern: The final three lines (repo = ..., use_case = ...) are the "composition root" — the single place in your application where concrete implementations are wired to abstractions. In a FastAPI app, this would live in a dependencies.py or a DI container.

Testing benefit:

def test_create_user():
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(repo)
    user = use_case.execute("[email protected]")
    assert user.email == "[email protected]"
    assert repo.find_by_id(user.id) is not None

No database, no HTTP server, no mocking needed — the in-memory repository IS the test double.

Expected Output

Created: User(id=1, [email protected])\nFound: User(id=1, [email protected])\nAll users: 2

Hints

Hint 1: Define an IUserRepository Protocol in the domain or application layer. Implement it with an in-memory version for testing.

Hint 2: The use case accepts the Protocol type in its constructor — it never imports the concrete implementation.

#5Use Case Result PatternMedium

use-caseresult-typeerror-handling

Implement a PlaceOrderUseCase that uses a Result type (Success/Failure) instead of exceptions for expected business rule violations.

from dataclasses import dataclass
from typing import Generic, TypeVar, Union

T = TypeVar("T")

@dataclass
class Success(Generic[T]):
    value: T

@dataclass
class Failure:
    error: str
    code: str

Result = Union[Success, Failure]

CATALOG = {42: "Widget", 7: "Gadget"}

class PlaceOrderUseCase:
    def execute(self, user_id: int, item_id: int, quantity: int) -> Result:
        if quantity <= 0:
            return Failure("Quantity must be positive", "INVALID_QUANTITY")
        if item_id not in CATALOG:
            return Failure(f"Item {item_id} not in catalog", "ITEM_NOT_FOUND")
        order = {"user_id": user_id, "item_id": item_id, "quantity": quantity}
        return Success(order)

use_case = PlaceOrderUseCase()

cases = [
    (1, 42, 3),
    (1, 42, -1),
    (1, 999, 2),
]

for user_id, item_id, qty in cases:
    result = use_case.execute(user_id, item_id, qty)
    if isinstance(result, Success):
        o = result.value
        print(f"Success: Order placed for user {o['user_id']}, item {o['item_id']}, qty {o['quantity']}")
    else:
        print(f"Failure: {result.error} [{result.code}]")

Solution

The solution is already shown above. The key design decisions:

Result type vs exceptions:

Aspect	Result type	Exception
Expected failures	Explicit in signature	Hidden from callers
Unexpected errors	Should still raise	Appropriate
IDE support	Full type checking	None
Caller must handle	Yes (type forces it)	No (easy to forget)

When to use each:

• Use Failure / Result for domain rule violations that are expected — invalid quantity, item not found, insufficient balance. These are part of the business logic.
• Raise exceptions for unexpected infrastructure errors — database connection failure, network timeout. These indicate something went wrong outside the domain.

Python 3.10+ alternative with match:

match use_case.execute(1, 42, 3):
    case Success(value=order):
        print(f"Order: {order}")
    case Failure(error=msg, code=code):
        print(f"Error {code}: {msg}")

Starter Code

from dataclasses import dataclass
from typing import Generic, TypeVar, Union

T = TypeVar("T")

@dataclass
class Success(Generic[T]):
    value: T

@dataclass
class Failure:
    error: str
    code: str

Result = Union[Success[T], Failure]

# Implement PlaceOrderUseCase that returns Result instead of raising exceptions
class PlaceOrderUseCase:
    def execute(self, user_id: int, item_id: int, quantity: int) -> Result:
        pass

Expected Output

Success: Order placed for user 1, item 42, qty 3\nFailure: Quantity must be positive [INVALID_QUANTITY]\nFailure: Item 999 not in catalog [ITEM_NOT_FOUND]

Hints

Hint 1: Return Success(value=...) for the happy path and Failure(error=..., code=...) for each business rule violation.

Hint 2: The caller checks isinstance(result, Success) or isinstance(result, Failure) — no try/except needed.

#6Layered Dependency Graph ValidatorMedium

dependency-rulegraphvalidation

Build a dependency rule validator that checks whether any module imports from a layer further out than itself.

from typing import Callable

LAYER_ORDER = ["domain", "application", "interface", "infrastructure"]

def get_layer(module_name: str) -> str:
    prefix = module_name.split(".")[0]
    return prefix if prefix in LAYER_ORDER else "unknown"

def validate_dependencies(modules: dict[str, list[str]]) -> list[str]:
    violations = []
    for module, imports in modules.items():
        src_layer = get_layer(module)
        if src_layer not in LAYER_ORDER:
            continue
        src_idx = LAYER_ORDER.index(src_layer)
        for imp in imports:
            dep_layer = get_layer(imp)
            if dep_layer not in LAYER_ORDER:
                continue
            dep_idx = LAYER_ORDER.index(dep_layer)
            if dep_idx > src_idx:
                violations.append(
                    f"VIOLATION: {module} -> {imp} "
                    f"({src_layer} cannot depend on {dep_layer})"
                )
    return violations

# Test 1: clean graph
clean = {
    "domain.user": [],
    "application.create_user": ["domain.user"],
    "interface.user_router": ["application.create_user"],
    "infrastructure.sql_repo": ["domain.user"],
}
result = validate_dependencies(clean)
print("No violations" if not result else "\n".join(result))

print("---")

# Test 2: violations
dirty = {
    "domain.user": ["infrastructure.db_session"],
    "application.create_user": ["domain.user", "interface.user_schema"],
    "interface.user_router": ["application.create_user"],
}
result = validate_dependencies(dirty)
print("\n".join(result))

Solution

The solution is shown above. Key insight: the dependency rule becomes mechanical once you assign a numeric depth to each layer. Any arrow pointing outward (to a higher index) is a violation.

Why this validator is useful:

• In large codebases, violations creep in silently. A test like this can be run in CI:

def test_dependency_rule():
    # Auto-discover modules and their imports using importlib/ast
    violations = validate_dependencies(discover_imports())
    assert violations == [], "\n".join(violations)

Automating import discovery with ast:

import ast, glob

def discover_imports() -> dict[str, list[str]]:
    result = {}
    for path in glob.glob("src/**/*.py", recursive=True):
        module = path.replace("/", ".").removesuffix(".py")
        tree = ast.parse(open(path).read())
        imports = []
        for node in ast.walk(tree):
            if isinstance(node, ast.ImportFrom) and node.module:
                imports.append(node.module)
        result[module] = imports
    return result

Starter Code

from typing import Callable

LAYER_ORDER = ["domain", "application", "interface", "infrastructure"]

def validate_dependencies(modules: dict[str, list[str]], classify: Callable[[str], str]) -> list[str]:
    """Check that no module imports from an outer layer.
    Return a list of violation messages.
    """
    pass

Expected Output

No violations\n---\nVIOLATION: domain.user -> infrastructure.db_session (domain cannot depend on infrastructure)\nVIOLATION: application.create_user -> interface.user_schema (application cannot depend on interface)

Hints

Hint 1: Build a mapping of module name to layer index using LAYER_ORDER.index(). A violation occurs when a module at layer index N imports from a module at index M where M > N.

Hint 2: The classify function maps a module path like "domain.user" to its layer name. You can parse the first segment.

#7Interactor with Input/Output PortsMedium

interactorportsdtoclean-architecture

Implement a RegisterUserInteractor that uses Input/Output ports (DTOs) and depends only on Protocol interfaces.

from dataclasses import dataclass, field
from typing import Protocol

@dataclass
class RegisterUserRequest:
    email: str
    password_hash: str
    full_name: str

@dataclass
class RegisterUserResponse:
    user_id: int
    email: str
    success: bool

class IUserStore(Protocol):
    def create(self, email: str, password_hash: str, full_name: str) -> int: ...
    def email_exists(self, email: str) -> bool: ...

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

class RegisterUserInteractor:
    def __init__(self, store: IUserStore, bus: IEventBus):
        self._store = store
        self._bus = bus

    def execute(self, req: RegisterUserRequest) -> RegisterUserResponse:
        if self._store.email_exists(req.email):
            return RegisterUserResponse(user_id=0, email=req.email, success=False)
        user_id = self._store.create(req.email, req.password_hash, req.full_name)
        self._bus.publish("user.registered", {"user_id": user_id, "email": req.email})
        return RegisterUserResponse(user_id=user_id, email=req.email, success=True)

# --- Fakes for testing ---
@dataclass
class FakeUserStore:
    _users: dict = field(default_factory=dict)
    _counter: int = 0

    def email_exists(self, email: str) -> bool:
        return email in self._users

    def create(self, email: str, password_hash: str, full_name: str) -> int:
        self._counter += 1
        self._users[email] = {"id": self._counter, "name": full_name}
        return self._counter

@dataclass
class FakeEventBus:
    published: list = field(default_factory=list)

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

# Compose and run
store = FakeUserStore()
bus = FakeEventBus()
interactor = RegisterUserInteractor(store, bus)

r1 = interactor.execute(RegisterUserRequest("[email protected]", "hashed", "Alice"))
print(r1)

store._users["[email protected]"] = {"id": 99, "name": "Bob"}
r2 = interactor.execute(RegisterUserRequest("[email protected]", "hashed", "Carol"))
print("Failed: Email already registered" if not r2.success else r2)

Solution

The solution is above. Notice the pattern:

Interactor = orchestrator, not business logic owner

The interactor's job is to:

1. Validate the request (at the use case level, not domain level)
2. Call domain and infrastructure ports in the right order
3. Return a clean output DTO

Business invariants (like "email must be valid") live in the domain. Infrastructure concerns (like "send an HTTP request to SendGrid") live in concrete implementations of IEventBus.

The composition is entirely in the test / main code. The interactor never calls FakeUserStore() — it only knows about IUserStore. This is the power of Clean Architecture: you can run the entire application logic with zero I/O, completely in-process, as fast as unit tests can run.

Starter Code

from dataclasses import dataclass
from typing import Protocol

# Input/Output data transfer objects (live at the boundary)
@dataclass
class RegisterUserRequest:
    email: str
    password_hash: str
    full_name: str

@dataclass
class RegisterUserResponse:
    user_id: int
    email: str
    success: bool

# Port interfaces
class IUserStore(Protocol):
    def create(self, email: str, password_hash: str, full_name: str) -> int: ...
    def email_exists(self, email: str) -> bool: ...

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

# Implement the interactor
class RegisterUserInteractor:
    pass

Expected Output

RegisterUserResponse(user_id=1, [email protected], success=True)\nFailed: Email already registered

Hints

Hint 1: An interactor (use case) takes a request DTO, calls port methods, and returns a response DTO. It never touches HTTP, databases, or queues directly.

Hint 2: Inject IUserStore and IEventBus through the constructor. Implement simple in-memory fakes for the test.

---

Hard

#8Clean Architecture Module Layout GeneratorHard

project-structureclean-architecturescaffolding

Build a scaffolding function that generates a complete Clean Architecture project layout with stub files.

import os
from pathlib import Path

def scaffold_clean_architecture(root: str, domain_name: str) -> list[str]:
    root_path = Path(root)
    created = []

    structure = {
        "domain": [
            "__init__.py",
            f"{domain_name}.py",        # Entity
            "value_objects.py",
            "ports.py",                 # Abstract interfaces / Protocols
            "exceptions.py",
        ],
        "application": [
            "__init__.py",
            "dtos.py",                  # Input/Output DTOs
            f"create_{domain_name}.py", # Use case
            f"get_{domain_name}.py",    # Use case
        ],
        "interface": [
            "__init__.py",
            f"{domain_name}_router.py", # FastAPI / Flask router
            f"{domain_name}_schema.py", # Pydantic schemas
        ],
        "infrastructure": [
            "__init__.py",
            f"sql_{domain_name}_repo.py",  # Concrete repository
            "database.py",
            "settings.py",
        ],
    }

    for layer, files in structure.items():
        layer_path = root_path / layer
        layer_path.mkdir(parents=True, exist_ok=True)
        for fname in files:
            fpath = layer_path / fname
            if not fpath.exists():
                fpath.write_text(f"# {layer}/{fname}\n")
            created.append(str(fpath))

    return created

# Test
created = scaffold_clean_architecture("/tmp/myapp", "order")
for f in created:
    print(f)

Solution

import os
from pathlib import Path

def scaffold_clean_architecture(root: str, domain_name: str) -> list[str]:
    root_path = Path(root)
    created = []

    structure = {
        "domain": [
            "__init__.py",
            f"{domain_name}.py",
            "value_objects.py",
            "ports.py",
            "exceptions.py",
        ],
        "application": [
            "__init__.py",
            "dtos.py",
            f"create_{domain_name}.py",
            f"get_{domain_name}.py",
        ],
        "interface": [
            "__init__.py",
            f"{domain_name}_router.py",
            f"{domain_name}_schema.py",
        ],
        "infrastructure": [
            "__init__.py",
            f"sql_{domain_name}_repo.py",
            "database.py",
            "settings.py",
        ],
    }

    for layer, files in structure.items():
        layer_path = root_path / layer
        layer_path.mkdir(parents=True, exist_ok=True)
        for fname in files:
            fpath = layer_path / fname
            if not fpath.exists():
                fpath.write_text(f"# {layer}/{fname}\n")
            created.append(str(fpath))

    return created

Enforcing the layout with tests:

After scaffolding, add an architecture fitness function that runs in CI:

def test_no_infrastructure_imports_in_domain():
    domain_files = list(Path("domain").rglob("*.py"))
    bad_imports = ["sqlalchemy", "fastapi", "redis", "httpx"]
    for f in domain_files:
        src = f.read_text()
        for bad in bad_imports:
            assert bad not in src, f"{f} imports {bad} — domain layer violation"

This test runs in milliseconds and catches the most common mistake: reaching for a framework import from inside the domain.

Starter Code

import os
from pathlib import Path

def scaffold_clean_architecture(root: str, domain_name: str) -> list[str]:
    """Create the Clean Architecture directory structure and stub files.
    Return the list of created file paths.
    Layers: domain/, application/, interface/, infrastructure/
    """
    pass

Expected Output

See solution for file list

Hints

Hint 1: Create four top-level packages: domain, application, interface, infrastructure. Each needs an __init__.py.

Hint 2: In domain: entities.py, value_objects.py, ports.py. In application: use_cases.py, dtos.py. In interface: routers.py. In infrastructure: repositories.py.

#9Event-Driven Clean ArchitectureHard

domain-eventsevent-busclean-architecture

Implement an event-driven use case where PlaceOrderUseCase publishes a OrderPlaced domain event that two handlers react to independently.

from dataclasses import dataclass, field
from typing import Callable
from datetime import datetime

@dataclass
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.utcnow)

@dataclass
class OrderPlaced(DomainEvent):
    order_id: int = 0
    customer_email: str = ""
    item_id: int = 0
    quantity: int = 0
    total_cents: int = 0

class InMemoryEventBus:
    def __init__(self):
        self._handlers: dict[type, list[Callable]] = {}

    def subscribe(self, event_type: type, handler: Callable) -> None:
        self._handlers.setdefault(event_type, []).append(handler)

    def publish(self, event: DomainEvent) -> None:
        for handler in self._handlers.get(type(event), []):
            handler(event)

class PlaceOrderUseCase:
    def __init__(self, bus: InMemoryEventBus):
        self._bus = bus
        self._next_order_id = 1

    def execute(self, customer_email: str, item_id: int, quantity: int, unit_price_cents: int) -> int:
        order_id = self._next_order_id
        self._next_order_id += 1
        total = quantity * unit_price_cents
        event = OrderPlaced(
            order_id=order_id,
            customer_email=customer_email,
            item_id=item_id,
            quantity=quantity,
            total_cents=total,
        )
        self._bus.publish(event)
        return order_id

# Handlers (live in infrastructure / application layer, not domain)
def email_handler(event: OrderPlaced) -> None:
    print(f"Email sent to {event.customer_email}: Your order {event.order_id} is confirmed")

def inventory_handler(event: OrderPlaced) -> None:
    print(f"Inventory updated: item {event.item_id} qty reduced by {event.quantity}")

# Wire up
bus = InMemoryEventBus()
bus.subscribe(OrderPlaced, lambda e: print(f"Handling OrderPlaced: order_id={e.order_id}, total={e.total_cents}"))
bus.subscribe(OrderPlaced, email_handler)
bus.subscribe(OrderPlaced, inventory_handler)

use_case = PlaceOrderUseCase(bus)
use_case.execute("[email protected]", 42, 3, 9967)

Solution

The solution is above. The architectural benefits:

Decoupling via events:

• PlaceOrderUseCase does NOT import email_handler or inventory_handler. It only knows about InMemoryEventBus.
• Adding a new reaction (e.g., loyalty_points_handler) requires zero changes to the use case — just subscribe a new handler.
• Each handler can be tested independently with a fake OrderPlaced event.

Synchronous vs asynchronous:

• InMemoryEventBus dispatches synchronously (good for tests, simple apps).
• In production, replace with a Kafka/RabbitMQ bus that serializes events. The use case code does not change — only the bus implementation changes.

Domain events belong in the domain layer. The event classes (OrderPlaced) carry domain concepts, not HTTP requests. Handlers live in the application or infrastructure layer.

Starter Code

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

# Domain event base
@dataclass
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.utcnow)

# Implement:
# 1. OrderPlaced domain event
# 2. IEventBus protocol
# 3. InMemoryEventBus implementation
# 4. PlaceOrderUseCase that raises OrderPlaced
# 5. Two handlers: one sends email, one updates inventory

Expected Output

Handling OrderPlaced: order_id=1, total=29900\nEmail sent to [email protected]: Your order 1 is confirmed\nInventory updated: item 42 qty reduced by 3

Hints

Hint 1: Domain events are raised inside the use case after the state change. They are published to the event bus, not dispatched synchronously in the domain entity.

Hint 2: The IEventBus protocol has subscribe(event_type, handler) and publish(event). InMemoryEventBus stores handlers in a dict keyed by event class.

#10Adapter Pattern: Multiple Database BackendsHard

adapterrepositorymultiple-backendsclean-architecture

Implement three adapters for IProductRepository: in-memory, JSON file, and a validation function that confirms they behave identically.

from dataclasses import dataclass, field
from typing import Protocol, Optional
import json, os

@dataclass
class Product:
    id: int
    name: str
    price_cents: int

class IProductRepository(Protocol):
    def save(self, product: Product) -> None: ...
    def find(self, product_id: int) -> Optional[Product]: ...
    def delete(self, product_id: int) -> bool: ...
    def find_all(self) -> list[Product]: ...

@dataclass
class InMemoryProductRepository:
    _store: dict = field(default_factory=dict)

    def save(self, product: Product) -> None:
        self._store[product.id] = product

    def find(self, product_id: int) -> Optional[Product]:
        return self._store.get(product_id)

    def delete(self, product_id: int) -> bool:
        return self._store.pop(product_id, None) is not None

    def find_all(self) -> list[Product]:
        return list(self._store.values())

class JsonFileProductRepository:
    def __init__(self, path: str):
        self._path = path
        if not os.path.exists(path):
            with open(path, "w") as f:
                json.dump({}, f)

    def _load(self) -> dict:
        with open(self._path) as f:
            raw = json.load(f)
        return {int(k): Product(**v) for k, v in raw.items()}

    def _dump(self, store: dict) -> None:
        with open(self._path, "w") as f:
            json.dump({str(k): vars(v) for k, v in store.items()}, f, indent=2)

    def save(self, product: Product) -> None:
        store = self._load()
        store[product.id] = product
        self._dump(store)

    def find(self, product_id: int) -> Optional[Product]:
        return self._load().get(product_id)

    def delete(self, product_id: int) -> bool:
        store = self._load()
        existed = product_id in store
        store.pop(product_id, None)
        self._dump(store)
        return existed

    def find_all(self) -> list[Product]:
        return list(self._load().values())

def run_product_workflow(repo: IProductRepository, label: str):
    repo.save(Product(1, "Widget", 999))
    repo.save(Product(2, "Gadget", 1999))
    found = repo.find(1)
    all_items = repo.find_all()
    deleted = repo.delete(2)
    remaining = repo.find_all()
    print(f"[{label}] Found: {found.name}, All: {len(all_items)}, Deleted: {deleted}, Remaining: {len(remaining)}")

# Run against both adapters
run_product_workflow(InMemoryProductRepository(), "InMemory")
json_repo = JsonFileProductRepository("/tmp/products.json")
run_product_workflow(json_repo, "JsonFile")
os.remove("/tmp/products.json")

Solution

The solution is above. The critical point: run_product_workflow is a use case that does not care which repository it receives. Swapping InMemoryProductRepository for JsonFileProductRepository requires only changing one line at the composition root.

Verifying adapter contract compliance:

import pytest

@pytest.fixture(params=["memory", "json"])
def repo(request, tmp_path):
    if request.param == "memory":
        return InMemoryProductRepository()
    return JsonFileProductRepository(str(tmp_path / "test.json"))

def test_save_and_find(repo):
    p = Product(1, "Widget", 999)
    repo.save(p)
    found = repo.find(1)
    assert found is not None
    assert found.name == "Widget"

def test_delete(repo):
    repo.save(Product(1, "X", 100))
    assert repo.delete(1) is True
    assert repo.find(1) is None
    assert repo.delete(1) is False

The same test suite runs against all adapters. Any adapter that passes the suite is a valid implementation of IProductRepository.

Starter Code

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

@dataclass
class Product:
    id: int
    name: str
    price_cents: int

class IProductRepository(Protocol):
    def save(self, product: Product) -> None: ...
    def find(self, product_id: int) -> Optional[Product]: ...
    def delete(self, product_id: int) -> bool: ...
    def find_all(self) -> list[Product]: ...

# Implement three adapters:
# 1. InMemoryProductRepository
# 2. JsonFileProductRepository (reads/writes a JSON file)
# 3. A function that runs the same use case against both

Expected Output

See solution for expected output

Hints

Hint 1: The JSON adapter reads the entire file on each operation (acceptable for this exercise). Use json.loads/json.dumps and store products as dicts.

Hint 2: The use case function accepts IProductRepository — it works identically whether you pass InMemory or JsonFile.

#11Full-Stack Clean Architecture: Order ServiceHard

clean-architecturefull-stackcomposition-rootintegration

Build a complete working order service spanning all four Clean Architecture layers. Start with the domain and work outward.

from dataclasses import dataclass, field
from typing import Protocol, Optional
from enum import Enum

# ── DOMAIN LAYER ──────────────────────────────────────────────────────────────

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

@dataclass
class Order:
    id: int
    customer_email: str
    item_id: int
    quantity: int
    total_cents: int
    status: OrderStatus = OrderStatus.PENDING

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

    def cancel(self) -> None:
        if self.status == OrderStatus.CANCELLED:
            raise ValueError("Order already cancelled")
        self.status = OrderStatus.CANCELLED

# ── DOMAIN PORTS ──────────────────────────────────────────────────────────────

class IOrderRepository(Protocol):
    def save(self, order: Order) -> None: ...
    def find(self, order_id: int) -> Optional[Order]: ...

class INotifier(Protocol):
    def notify(self, email: str, message: str) -> None: ...

# ── APPLICATION LAYER ─────────────────────────────────────────────────────────

@dataclass
class PlaceOrderRequest:
    customer_email: str
    item_id: int
    quantity: int
    unit_price_cents: int

@dataclass
class OrderResponse:
    order_id: int
    status: str
    total_cents: int

class PlaceOrderUseCase:
    def __init__(self, repo: IOrderRepository, notifier: INotifier):
        self._repo = repo
        self._notifier = notifier
        self._next_id = 1

    def execute(self, req: PlaceOrderRequest) -> OrderResponse:
        order = Order(
            id=self._next_id,
            customer_email=req.customer_email,
            item_id=req.item_id,
            quantity=req.quantity,
            total_cents=req.quantity * req.unit_price_cents,
        )
        self._next_id += 1
        order.confirm()
        self._repo.save(order)
        self._notifier.notify(order.customer_email, f"Order {order.id} confirmed!")
        return OrderResponse(order.id, order.status.value, order.total_cents)

class GetOrderUseCase:
    def __init__(self, repo: IOrderRepository):
        self._repo = repo

    def execute(self, order_id: int) -> Optional[OrderResponse]:
        order = self._repo.find(order_id)
        if order is None:
            return None
        return OrderResponse(order.id, order.status.value, order.total_cents)

# ── INTERFACE LAYER ───────────────────────────────────────────────────────────

class OrderController:
    def __init__(self, place: PlaceOrderUseCase, get: GetOrderUseCase):
        self._place = place
        self._get = get

    def handle_place(self, body: dict) -> dict:
        req = PlaceOrderRequest(
            customer_email=body["email"],
            item_id=body["item_id"],
            quantity=body["quantity"],
            unit_price_cents=body["unit_price_cents"],
        )
        resp = self._place.execute(req)
        return {"order_id": resp.order_id, "status": resp.status, "total": resp.total_cents}

    def handle_get(self, order_id: int) -> dict:
        resp = self._get.execute(order_id)
        if resp is None:
            return {"error": "not found"}
        return {"order_id": resp.order_id, "status": resp.status, "total": resp.total_cents}

# ── INFRASTRUCTURE LAYER ──────────────────────────────────────────────────────

@dataclass
class InMemoryOrderRepository:
    _store: dict = field(default_factory=dict)

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

    def find(self, order_id: int) -> Optional[Order]:
        return self._store.get(order_id)

class ConsoleNotifier:
    def notify(self, email: str, message: str) -> None:
        print(f"[NOTIFY] {email}: {message}")

# ── COMPOSITION ROOT ──────────────────────────────────────────────────────────

repo = InMemoryOrderRepository()
notifier = ConsoleNotifier()
place_uc = PlaceOrderUseCase(repo, notifier)
get_uc = GetOrderUseCase(repo)
controller = OrderController(place_uc, get_uc)

# Simulate HTTP requests
r1 = controller.handle_place({"email": "[email protected]", "item_id": 7, "quantity": 2, "unit_price_cents": 4999})
print(f"Placed: {r1}")

r2 = controller.handle_get(1)
print(f"Retrieved: {r2}")

r3 = controller.handle_get(999)
print(f"Not found: {r3}")

Solution

The solution is above — this is a fully working order service with all four layers properly separated.

Architecture summary:

domain/       Order, OrderStatus, IOrderRepository, INotifier
application/  PlaceOrderUseCase, GetOrderUseCase, PlaceOrderRequest, OrderResponse
interface/    OrderController (translates dicts to/from DTOs)
infrastructure/ InMemoryOrderRepository, ConsoleNotifier

What makes this Clean Architecture (not just layered):

• PlaceOrderUseCase never imports InMemoryOrderRepository — it depends on the IOrderRepository Protocol.
• Order.confirm() enforces the domain invariant, not a database constraint or HTTP validator.
• The composition root is the only place that knows about all concrete implementations.
• Swapping InMemoryOrderRepository for SqlAlchemyOrderRepository requires changing one line in the composition root and zero lines anywhere else.

Starter Code

# Build a complete mini order service with all four layers:
# domain: Order entity, OrderStatus enum, Money value object
# application: PlaceOrderUseCase, GetOrderUseCase (with DTOs)
# interface: a process_request() function that simulates an HTTP handler
# infrastructure: InMemoryOrderRepository, ConsoleNotifier
# Compose everything at the bottom

Expected Output

See solution for expected output

Hints

Hint 1: Start from the inside out: write the domain first, then ports, then use cases, then interface, then infrastructure, then compose.

Hint 2: The interface layer translates raw dicts (simulating HTTP JSON bodies) into request DTOs and response DTOs back into dicts.