component-reference.md

Component Reference

All components are exported from the package root (seedwork).

---

Domain layer

Entity[TId]

• Role: Base class for DDD entities. Identity over attributes — two entities are equal when they share the same id of the same concrete class, regardless of other fields.
• Usage: Subclass as @dataclass(frozen=True, eq=False, kw_only=True) and declare id via inheritance. Use __post_init__ for additional validation. Raises NullEntityIdError if id is None.
• Key methods: __eq__ compares by id when both objects are the same concrete class. __hash__ is based on id. _evolve(**changes) -> Self returns a new instance with the given fields replaced.

AggregateRoot[TId]

• Role: Root of an aggregate. Single entry point for state changes. Accumulates domain events without side effects — all behavior methods return new instances.
• Fields: domain_events: tuple[DomainEvent, ...] — immutable, keyword-only, excluded from repr, hash, and eq. Passed to the constructor when seeding events (e.g. in factory class methods).
• Key methods: _evolve(**changes) -> Self — inherited from Entity; produces a new instance with updated fields. _record(*events) -> Self — returns a new instance with the given events appended to domain_events.
• Usage pattern: Use two factory patterns: open/create for new aggregates — pass domain_events=(event,) in the constructor; reconstitute for loading from persistence — pass no domain_events (those have already been published). Behavior methods chain _evolve(state_change)._record(event) and return the new instance. DomainEventPublishingRepository reads domain_events and publishes after save.

ValueObject

• Role: Immutable domain concept defined entirely by its attributes. Subclass as @dataclass(frozen=True, kw_only=True). Equality and hashing are structural — delegated to the dataclass.
• Usage: Declare fields directly on the subclass. Use __post_init__ for validation; raise a DomainError subclass on invalid input (co-located in the same file). All fields are keyword-only.

DomainEvent / DomainEventRecord[TPayload]

• DomainEvent — Protocol defining the structural interface for domain events: id: str and occurred_at: datetime.
• DomainEventRecord[TPayload] — frozen dataclass; declares payload: TPayload first, then id: str (default UUID) and occurred_at: datetime (default UTC now).
• Pattern: define a frozen dataclass Payload, then a frozen dataclass event extending DomainEventRecord[Payload]. Name events in past tense. Keep payload fields primitive (serializable).

@dataclass(frozen=True)
class MoneyDepositedPayload:
    account_id: str
    amount: float
    currency: str

@dataclass(frozen=True)
class MoneyDeposited(DomainEventRecord[MoneyDepositedPayload]):
    pass

Repository[TId, TAggregate]

• Methods: find_by_id(entity_id: TId) -> TAggregate | None, save(aggregate: TAggregate) -> None, delete_by_id(entity_id: TId) -> None. All are async.
• Define a typed sub-interface in the domain layer; implement in infrastructure.

UnitOfWork

• Protocol (structural — no inheritance required). Implementations must provide __aenter__(self) -> Self and __aexit__(self, exc_type, exc_val, exc_tb) -> None. __aexit__ should commit when exc_type is None and roll back otherwise — TransactionalCommandBus relies on this contract.

DomainError

• Base Exception subclass. Constructor (message: str, code: str). Exposes self.code. Always subclass with a named class — DomainError itself is not meant to be raised directly.

---

Application layer

Result / ResultError

• Result.succeeded() / Result.failed(errors: list[ResultError]). Check with result.ok: bool. .errors: tuple[ResultError, ...] (immutable).
• Use for expected domain failures at the application boundary; let infrastructure exceptions propagate.

Command / CommandBus / CommandHandler[TCommand]

• Command — frozen dataclass base. Subclass as @dataclass(frozen=True, kw_only=True) and declare fields directly.
• CommandHandler[TCommand] — Protocol. execute(self, command: TCommand) -> None (async).
• CommandBus — Protocol. dispatch(self, command: Command) -> Result (async).

Query[TResult] / QueryBus / QueryHandler[TQuery, TResult]

• Query[TResult] — generic frozen dataclass base. Subclass as @dataclass(frozen=True, kw_only=True) and declare the result type as a type parameter: class MyQuery(Query[MyResponse]).
• QueryHandler[TQuery, TResult] — Protocol. execute(self, query: TQuery) -> TResult | None (async). Return None to signal absence.
• QueryBus — Protocol. ask(self, query: Query[TResult]) -> TResult | None (async). The return type is inferred from the query's type parameter — no Any, no cast at the call site.

DomainEventPublisher

• Protocol. publish(self, events: Sequence[DomainEvent]) -> None (async). Accepts any sequence — tuples from aggregate.domain_events are passed directly.
• Do not inject into command handlers — use DomainEventPublishingRepository instead.

DomainEventHandler[TEvent]

• Protocol. handle(self, event: TEvent) -> None (async).

---

Infrastructure layer

RegistryCommandBus

• Routes commands to handlers via in-process registry keyed by command class.
• register(command_type, handler), dispatch(command) -> Result.
• Catches DomainError and converts to Result.failed. All other exceptions propagate.

bus = RegistryCommandBus()
bus.register(OpenAccountCommand, OpenAccountHandler(repo))

result = await bus.dispatch(OpenAccountCommand(account_id="acc-1", initial_balance=100.0))
result.ok  # True

# DomainError → Result.failed
result = await bus.dispatch(...)  # handler raises InsufficientFundsError
result.ok              # False
result.errors[0].code  # "INSUFFICIENT_FUNDS"

RegistryQueryBus

• Same registry pattern for queries. register(query_type, handler), ask(query) -> TResult | None. The bus is generically typed, so the return type matches the registered query handler result type.
• Raises KeyError when no handler is registered for the query type.

bus = RegistryQueryBus()
bus.register(GetBalanceQuery, GetBalanceHandler(read_repo))

balance = await bus.ask(GetBalanceQuery(account_id="acc-1"))
# balance: BalanceResponse | None

TransactionalCommandBus

• Decorator. Wraps dispatch in the UnitOfWork context manager (async with unit_of_work). Commit and rollback are the context manager's responsibility.

bus = TransactionalCommandBus(inner=registry_bus, unit_of_work=uow)
# Every dispatch runs inside: async with uow: inner.dispatch(command)

DomainEventPublishingRepository[TId, TAggregate]

• Decorator. Reads aggregate.domain_events and calls publisher.publish(aggregate.domain_events) after every save. delete_by_id and find_by_id delegate without side effects.

repo = DomainEventPublishingRepository(inner=BankAccountRepositoryImpl(), publisher=my_publisher)

account = BankAccount.open(BankAccountId("acc-1"), Money(amount=100.0, currency="EUR"))
await repo.save(account)
# inner_repo.save is called first, then publisher.publish(account.domain_events)

CommandBusBuilder

• .register(command_type, handler) — wire handler (last registration wins).
• .with_transaction(unit_of_work) — add TransactionalCommandBus.
• .use(middleware: Callable[[CommandBus], CommandBus]) — add custom middleware.
• .build() -> CommandBus — return assembled bus.
• Declaration order = stack order; first declared = outermost.

bus = (
    CommandBusBuilder()
    .register(OpenAccountCommand, OpenAccountHandler(repo))
    .register(DepositMoneyCommand, DepositMoneyHandler(repo))
    .with_transaction(uow)
    .build()
)

result = await bus.dispatch(DepositMoneyCommand(account_id="acc-1", amount=50.0, currency="EUR"))

QueryBusBuilder

• .register(query_type, handler) — wire handler.
• .use(middleware: Callable[[QueryBus], QueryBus]) — add custom middleware.
• .build() -> QueryBus — return assembled bus.

bus = (
    QueryBusBuilder()
    .register(GetBalanceQuery, GetBalanceHandler(read_repo))
    .build()
)

balance = await bus.ask(GetBalanceQuery(account_id="acc-1"))

InMemoryRepository[TId, TAggregate]

• Generic in-memory Repository implementation backed by a plain dict. Intended for use in tests and as a starting point for proof-of-concept implementations.
• Satisfies the Repository[TId, TAggregate] Protocol structurally — no inheritance declaration needed.
• All three methods (find_by_id, save, delete_by_id) are async and match the Repository contract exactly.

repo: InMemoryRepository[BankAccountId, BankAccount] = InMemoryRepository()
await repo.save(account)
found = await repo.find_by_id(BankAccountId("acc-1"))