Add a feature
Two ways: scaffold it, or walk the rings by hand. Both end with the same eight files.
Scaffold
uv run python scripts/new_feature.py orders
creates:
src/cleanarch/orders/- anOrderentity (id, name, created_at),OrderCreatedevent,OrderRepositoryport,CreateOrder/GetOrderuse cases, an in-memory adapter and a router withPOST /ordersandGET /orders/{id};src/cleanarch/bootstrap/features/orders.py- the wiring, already listed inFEATURES;tests/test_orders.py- a use-case test with fakes and an HTTP round trip.
make check is green and make run-memory serves POST /api/v1/orders before you change a
line. Now replace the placeholder fields with your model, inside-out.
By hand, inside-out
1. Domain - what is true regardless of I/O
@dataclass(frozen=True, slots=True)
class Order:
id: OrderId
customer: str
lines: tuple[OrderLine, ...]
status: OrderStatus = OrderStatus.DRAFT
def __post_init__(self) -> None:
if not self.lines:
raise EmptyOrder
def submit(self) -> DomainResult["Order"]:
if self.status is not OrderStatus.DRAFT:
raise OrderAlreadySubmitted
return DomainResult.of(replace(self, status=OrderStatus.SUBMITTED), OrderSubmitted(self.id))
Rules in __post_init__ and methods. Methods return DomainResult (new state + events) and
never mutate. Errors are specific DomainError subclasses. Write the domain tests now -
they need nothing (see tests/tournaments/test_tournament.py for the shape).
2. Application - one class per operation, ports for the outside
class OrderRepository(Protocol):
async def add(self, order: Order) -> None: ...
async def get(self, order_id: OrderId) -> Order | None: ...
async def save(self, order: Order) -> None: ...
class SubmitOrder:
def __init__(self, repository: OrderRepository, events: EventPublisher) -> None: ...
async def execute(self, order_id: OrderId) -> Order:
order = await self._repository.get(order_id)
if order is None:
raise OrderNotFound(order_id)
result = order.submit()
await self._repository.save(result.aggregate)
await self._events.publish(result.events)
return result.aggregate
Need a command object? Only when there are several input fields (CreateOrderCommand).
Need the current time? Take the shared Clock port in the constructor (as CreateTournament
does) and pass clock.now() into the domain; never call datetime.now() in a use case or
an entity.
Test use cases with InMemoryOrderRepository and RecordingEventPublisher (from
tests/conftest.py); the scaffolded tests/test_orders.py already does.
3. Infrastructure - implement the ports
Start with in_memory.py (a dict). Add sqlalchemy/ when you need persistence:
models.py- the row (OrderModel(Base)), not the entity;mapping.py-to_model/to_domain/update_model;repository.py- the port implementation on anAsyncSession,flush()notcommit().
Then alembic/env.py: import the model module, and
make migration m="create orders" # autogenerate
make migrate
Copy tests/tournaments/test_repository_contract.py: the same tests must pass on
your in-memory and SQL adapters, including the stale-write conflict (ADR-010).
4. HTTP - parse, execute, present
schemas.py:CreateOrderRequest.to_command(),OrderResponse.from_domain().dependencies.py:get_order_repository()placeholder + one factory per use case.router.py: every handler is three lines.
Add an HTTP test per error mapping (client fixture, see tests/tournaments/test_api.py).
5. Bootstrap - choose adapters
bootstrap/features/orders.py is where the feature meets the outside world. Its optional
hooks: wire_http(app, settings) (pick the repository from settings.use_in_memory, include
the router), subscribe(bus) (event handlers; they run after the request committed),
register_cli(subparsers) / run_cli(args, session, events). The example's module shows all
four. The module must be listed in bootstrap/features/__init__.py.
6. Check
make check
scripts/archcheck.py check (part of it) tells you if anything points the wrong way.