### Build an aggregate root

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Example of building an immutable aggregate root for a bank account, including factory patterns and state-change methods.

```python
from dataclasses import dataclass
from typing import NewType, Self
from seedwork.domain import AggregateRoot


BankAccountId = NewType("BankAccountId", str)


@dataclass(frozen=True, eq=False, kw_only=True)
class BankAccount(AggregateRoot[BankAccountId]):
    balance: Money

    @classmethod
    def open(cls, id: BankAccountId, initial_balance: Money) -> Self:
        return cls(
            id=id,
            balance=initial_balance,
            domain_events=(
                AccountOpened(
                    payload=AccountOpenedPayload(
                        account_id=id,
                        initial_balance=initial_balance.amount,
                        currency=initial_balance.currency,
                    )
                ),
            ),
        )

    @classmethod
    def reconstitute(cls, id: BankAccountId, balance: Money) -> Self:
        return cls(id=id, balance=balance)  # no domain_events — already published

    def deposit(self, amount: Money) -> Self:
        return self._evolve(
            balance=Money(
                amount=self.balance.amount + amount.amount,
                currency=self.balance.currency,
            )
        )._record(
            MoneyDeposited(
                payload=MoneyDepositedPayload(
                    account_id=self.id,
                    amount=amount.amount,
                    currency=amount.currency,
                )
            )
        )

    def withdraw(self, amount: Money) -> Self:
        if amount.amount > self.balance.amount:
            raise InsufficientFundsError()
        return self._evolve(
            balance=Money(
                amount=self.balance.amount - amount.amount,
                currency=self.balance.currency,
            )
        )
```

--------------------------------

### Wire the Buses

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Assemble the bus stack at the composition root using builders. Wrap the repository with DomainEventPublishingRepository to ensure events are published automatically after each save. This example shows wiring for both write and read sides.

```python
from seedwork.infrastructure import (
    CommandBusBuilder,
    DomainEventPublishingRepository,
    InMemoryRepository,
    QueryBusBuilder,
)

# Write side — InMemoryRepository is useful for tests and prototyping
repo: InMemoryRepository[BankAccountId, BankAccount] = InMemoryRepository()
publishing_repo = DomainEventPublishingRepository(repo, my_event_publisher)

command_bus = (
    CommandBusBuilder()
    .register(DepositMoneyCommand, DepositMoneyHandler(publishing_repo))
    .with_transaction(uow)
    .build()
)

# Read side
query_bus = (
    QueryBusBuilder()
    .register(GetBalanceQuery, GetBalanceHandler(read_repo))
    .build()
)
```

--------------------------------

### Define domain events

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Example of defining domain events with typed payloads, extending DomainEventRecord.

```python
from dataclasses import dataclass
from seedwork.domain import DomainEventRecord


@dataclass(frozen=True)
class AccountOpenedPayload:
    account_id: str
    initial_balance: float
    currency: str


@dataclass(frozen=True)
class AccountOpened(DomainEventRecord[AccountOpenedPayload]):
    pass


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


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

--------------------------------

### Define domain errors

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Example of defining custom domain errors by subclassing DomainError.

```python
from seedwork.domain import DomainError


class InsufficientFundsError(DomainError):
    def __init__(self) -> None:
        super().__init__("Insufficient funds", "INSUFFICIENT_FUNDS")


class AccountNotFoundError(DomainError):
    def __init__(self, account_id: str) -> None:
        super().__init__(f"Account {account_id} not found", "ACCOUNT_NOT_FOUND")
```

--------------------------------

### Define value objects

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Example of defining value objects like Money, subclassing ValueObject and using dataclasses with invariants.

```python
from dataclasses import dataclass
from seedwork.domain import DomainError, ValueObject


class NegativeAmountError(DomainError):
    def __init__(self) -> None:
        super().__init__("Amount cannot be negative", "NEGATIVE_AMOUNT")


class EmptyCurrencyError(DomainError):
    def __init__(self) -> None:
        super().__init__("Currency cannot be empty", "EMPTY_CURRENCY")


@dataclass(frozen=True, kw_only=True)
class Money(ValueObject):
    amount: float
    currency: str

    def __post_init__(self) -> None:
        if self.amount < 0:
            raise NegativeAmountError()
        if not self.currency:
            raise EmptyCurrencyError()


Money(amount=10.0, currency="EUR") == Money(amount=10.0, currency="EUR")  # True
Money(amount=10.0, currency="EUR") == Money(amount=20.0, currency="EUR")  # False
```

--------------------------------

### Development Setup

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/README.md

Clone the repository and install development dependencies.

```bash
git clone https://github.com/aseguragonzalez/python-seedwork.git
cd python-seedwork
make install
```

--------------------------------

### QueryBusBuilder Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/component-reference.md

Demonstrates building a QueryBus with registered handlers.

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

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

--------------------------------

### Dispatch Commands and Queries

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Dispatch commands using the command bus and queries using the query bus. Commands return a Result object, allowing for error handling. Queries return the declared result type or None.

```python
# Commands return Result — DomainError is caught and converted to Result.failed
result = await command_bus.dispatch(
    DepositMoneyCommand(account_id="acc-1", amount=50.0, currency="EUR")
)
if not result.ok:
    for error in result.errors:
        print(error.code, error.description)

# Queries return the declared result type or None
balance = await query_bus.ask(GetBalanceQuery(account_id="acc-1"))
# balance: BalanceResponse | None  ← inferred, no cast needed
if balance is None:
    ...
```

--------------------------------

### Define Commands and Handlers

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Commands represent write intentions. The handler's responsibility is orchestration: loading the aggregate, invoking the domain method, and saving the returned instance. This example shows a DepositMoneyCommand and its handler.

```python
from dataclasses import dataclass
from seedwork.application import Command, CommandHandler


@dataclass(frozen=True, kw_only=True)
class DepositMoneyCommand(Command):
    account_id: str
    amount: float
    currency: str


class DepositMoneyHandler(CommandHandler[DepositMoneyCommand]):
    def __init__(self, repository: BankAccountRepository) -> None:
        self._repository = repository

    async def execute(self, command: DepositMoneyCommand) -> None:
        account_id = BankAccountId(command.account_id)
        account = await self._repository.find_by_id(account_id)
        if account is None:
            raise AccountNotFoundError(command.account_id)

        updated = account.deposit(Money(amount=command.amount, currency=command.currency))
        await self._repository.save(updated)
        # DomainEventPublishingRepository publishes events after save automatically
```

--------------------------------

### Full Wiring Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

A comprehensive example demonstrating the wiring of various components including event bus, repository, and command bus.

```python
from seedwork.infrastructure import (
    CommandBusBuilder,
    DeferredDomainEventBus,
    DomainEventPublishingRepository,
    RegistryCommandBus,
)
from seedwork.testing import InMemoryIntegrationEventPublisher, InMemoryTaskScheduler

event_bus = DeferredDomainEventBus()

publisher = InMemoryIntegrationEventPublisher()   # or OutboxIntegrationEventPublisher
scheduler = InMemoryTaskScheduler()               # or OutboxTaskScheduler
scheduler.register(SendWelcomeEmailTask.TYPE, SendWelcomeEmailTaskHandler())

event_bus.subscribe(AccountOpened, AccountOpenedDomainEventHandler(publisher, scheduler))

account_repo = DomainEventPublishingRepository(
    SqlAlchemyBankAccountRepository(session),
    event_bus,
)

registry = RegistryCommandBus()
command_bus = (
    CommandBusBuilder(registry)
    .register(OpenAccountCommand, OpenAccountCommandHandler(account_repo))
    .with_transaction(uow)
    .with_domain_event_coordination(event_bus)
    .build()
)
```

--------------------------------

### RegistryQueryBus Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/component-reference.md

Shows how to register a query handler and ask for query results.

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

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

--------------------------------

### RegistryCommandBus Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/component-reference.md

Demonstrates registering a command handler and dispatching commands, including handling DomainErrors.

```python
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"
```

--------------------------------

### InMemoryRepository Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/component-reference.md

Provides an example of using the generic in-memory repository for tests.

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

--------------------------------

### Define Repository Interface

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Repository interfaces are defined in the domain layer, while their implementations reside in the infrastructure layer.

```python
from seedwork.domain import Repository


class BankAccountRepository(Repository[BankAccountId, BankAccount]):
    ...
```

--------------------------------

### Define Queries and Handlers

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/getting-started.md

Queries are read-only operations. Each query specifies its return type as a type parameter, enabling full type safety at the call site. A dedicated read repository, defined as an ad-hoc Protocol in the application layer, should be used instead of a domain Repository for query handlers.

```python
from dataclasses import dataclass
from typing import Protocol
from seedwork.application import Query, QueryHandler


@dataclass(frozen=True)
class BalanceResponse:
    account_id: str
    balance: float
    currency: str


@dataclass(frozen=True, kw_only=True)
class GetBalanceQuery(Query[BalanceResponse]):
    account_id: str


class BankAccountReadRepository(Protocol):
    async def find_balance(self, account_id: str) -> BalanceResponse | None: ...


class GetBalanceHandler(QueryHandler[GetBalanceQuery, BalanceResponse]):
    def __init__(self, repository: BankAccountReadRepository) -> None:
        self._repository = repository

    async def execute(self, query: GetBalanceQuery) -> BalanceResponse | None:
        return await self._repository.find_balance(query.account_id)
```

--------------------------------

### Domain Event Usage Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Example of how to use the create() classmethod for a domain event within an aggregate.

```python
AccountOpened.create(
    initial_balance=initial_balance.amount,
    currency=initial_balance.currency,
    aggregate_id=str(self.id),
)
```

--------------------------------

### CommandBusBuilder Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/component-reference.md

Shows how to build a CommandBus using the builder pattern, including registering handlers and adding transaction support.

```python
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"))
```

--------------------------------

### Installation

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/index.md

Install the python-seedwork package using pip.

```bash
pip install python-seedwork
```

--------------------------------

### TransactionalCommandBus Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/component-reference.md

Illustrates wrapping command dispatch within a Unit of Work context.

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

--------------------------------

### Command and Handler Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Defines an OpenAccountCommand with validation and its corresponding handler.

```python
from __future__ import annotations
from dataclasses import dataclass
from seedwork.application import Command, CommandHandler
from seedwork.domain import DomainError


class InvalidInitialBalanceError(DomainError):
    def __init__(self) -> None:
        super().__init__("initial_balance must be non-negative", "INVALID_INITIAL_BALANCE")


@dataclass(frozen=True, kw_only=True)
class OpenAccountCommand(Command):
    account_id: str
    owner_id: str
    initial_balance: float
    currency: str

    def __post_init__(self) -> None:
        if self.initial_balance < 0:
            raise InvalidInitialBalanceError()

class OpenAccountCommandHandler(CommandHandler[OpenAccountCommand]):
    def __init__(self, repository: BankAccountRepository) -> None:
        self._repository = repository

    async def handle(self, command: OpenAccountCommand) -> None:
        account = BankAccount.open(
            id=BankAccountId(command.account_id),
            owner_id=UserId(command.owner_id),
            initial_balance=Money(amount=command.initial_balance, currency=command.currency),
        )
        await self._repository.save(account)
```

--------------------------------

### Aggregate Root Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Example of a BankAccount Aggregate Root, demonstrating state changes and event recording.

```python
from __future__ import annotations
from dataclasses import dataclass
from typing import NewType
from uuid import UUID

from seedwork.domain import AggregateRoot

UserId = NewType("UserId", UUID)

@dataclass(frozen=True, eq=False, kw_only=True)
class BankAccount(AggregateRoot[BankAccountId]):
    owner_id: UserId
    balance: Money

    def validate(self) -> None:
        pass

    @classmethod
    def open(cls, id: BankAccountId, owner_id: UserId, initial_balance: Money) -> BankAccount:
        return cls(id=id, owner_id=owner_id, balance=initial_balance)._record(
            AccountOpened.create(
                initial_balance=initial_balance.amount,
                currency=initial_balance.currency,
                aggregate_id=str(id),
            )
        )

    def deposit(self, amount: Money) -> BankAccount:
        new_balance = Money(amount=self.balance.amount + amount.amount, currency=self.balance.currency)
        return self._evolve(balance=new_balance)._record(
            MoneyDeposited.create(
                amount=amount.amount,
                aggregate_id=str(self.id),
            )
        )
```

--------------------------------

### DomainEventPublishingRepository Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/component-reference.md

Demonstrates using the decorator to publish domain events after saving an aggregate.

```python
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)
```

--------------------------------

### Domain Event Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Example of defining a Domain Event (AccountOpened) with a payload.

```python
from __future__ import annotations
from dataclasses import dataclass

from seedwork.domain.domain_event import BaseDomainEvent


@dataclass(frozen=True)
class AccountOpenedPayload:
    initial_balance: float
    currency: str


@dataclass(frozen=True)
class AccountOpened(BaseDomainEvent[AccountOpenedPayload]):
    @classmethod
    def create(cls, initial_balance: float, currency: str, aggregate_id: str) -> AccountOpened:
        return cls(
            payload=AccountOpenedPayload(initial_balance=initial_balance, currency=currency),
            aggregate_id=aggregate_id,
        )
```

--------------------------------

### Query and Handler Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Demonstrates the implementation of a Query and its corresponding QueryHandler, adhering to read-only principles and returning a view model or None.

```python
from __future__ import annotations
from dataclasses import dataclass
from seedwork.application import Query, QueryHandler

@dataclass(frozen=True, kw_only=True)
class GetAccountQuery(Query[AccountView]):
    account_id: str

@dataclass(frozen=True, kw_only=True)
class AccountView:
    id: str
    balance: float

class GetAccountQueryHandler(QueryHandler[GetAccountQuery, AccountView]):
    def __init__(self, repository: AccountReadRepository) -> None:
        self._repository = repository

    async def handle(self, query: GetAccountQuery) -> AccountView | None:
        return await self._repository.find_view(query.account_id)
```

--------------------------------

### Entity Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Example of an Entity implementation in the domain layer, demonstrating inheritance from `Entity`, use of `NewType` for identifiers, and the `validate` method.

```python
from __future__ import annotations
from dataclasses import dataclass
from typing import NewType
from uuid import UUID

from seedwork.domain import Entity

AccountId = NewType("AccountId", UUID)
UserId = NewType("UserId", UUID)

@dataclass(frozen=True, eq=False, kw_only=True)
class Account(Entity[AccountId]):
    owner_id: UserId
    balance: int

    def validate(self) -> None:
        pass  # add domain invariants here if needed
```

--------------------------------

### CommandBus Wiring Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Shows how to configure and build a CommandBus using CommandBusBuilder, including optional decorators for transactions and event coordination.

```python
from seedwork.infrastructure import CommandBusBuilder, RegistryCommandBus

registry = RegistryCommandBus()
command_bus = (
    CommandBusBuilder(registry)
    .register(OpenAccountCommand, OpenAccountCommandHandler(repository))
    .with_transaction(uow)                      # optional — TransactionalCommandBus
    .with_domain_event_coordination(event_bus)  # DomainEventCoordinatorCommandBus
    .build()
)
```

--------------------------------

### Integration Event Consumer Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Shows the structure of an IntegrationEventHandler for consuming integration events asynchronously.

```python
from seedwork.application import IntegrationEventHandler

class AccountOpenedIntegrationEventHandler(IntegrationEventHandler[AccountOpenedIntegrationEvent]):
    async def handle(self, event: AccountOpenedIntegrationEvent) -> None:
        # React to an integration event from another bounded context
        ...
```

--------------------------------

### Value Object Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Example of a Value Object implementation, showcasing immutability with `frozen=True`, keyword-only arguments with `kw_only=True`, and validation via the `validate` method.

```python
from __future__ import annotations
from dataclasses import dataclass

from seedwork.domain import DomainError, ValueObject


class NegativeAmountError(DomainError):
    def __init__(self) -> None:
        super().__init__("Amount cannot be negative", "NEGATIVE_AMOUNT")


@dataclass(frozen=True, kw_only=True)
class Money(ValueObject):
    amount: float
    currency: str

    def validate(self) -> None:
        if self.amount < 0:
            raise NegativeAmountError()
```

--------------------------------

### Commands

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/CLAUDE.md

Common make commands for managing the project, including installation, validation, linting, formatting, type checking, testing, and cleaning.

```bash
make install        # uv sync + install pre-commit hooks (run once after clone)
make all            # full validation: lint + typecheck + tests + pre-commit hooks
make check          # lint + typecheck + tests (what CI runs)
make pre-commit     # run all pre-commit hooks against all files (includes markdownlint)
make lint           # ruff check src tests docs/examples
make format         # ruff format + ruff check --fix
make typecheck      # pyright
make test           # pytest with coverage (fails below 90%)
make test-no-cov    # pytest without coverage gate
make clean          # remove dist, .coverage, caches, __pycache__
```

--------------------------------

### DomainEventPublishingRepository Usage

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Example of how to use the DomainEventPublishingRepository to decorate a repository and publish domain events.

```python
from seedwork.infrastructure import DomainEventPublishingRepository

publishing_repo = DomainEventPublishingRepository(
    SqlAlchemyBankAccountRepository(session),
    event_bus,  # satisfies DomainEventBusPublisher
)
```

--------------------------------

### Repository Interface Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Defines a repository interface for a bank account, adhering to the seedwork pattern.

```python
from __future__ import annotations
from typing import Protocol
from .account import BankAccount

class BankAccountRepository(Repository[BankAccountId, BankAccount], Protocol):
    pass
```

--------------------------------

### Propagation Rules Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/architecture.md

Illustrates how correlation_id and causation_id propagate through a chain of operations.

```text
HTTP Request (correlation_id = X, causation_id = X)
  └── Command dispatched
        └── DomainEvent emitted        (correlation_id = X, causation_id = command.id)
              └── IntegrationEvent published  (correlation_id = X, causation_id = domainEvent.id)
                    └── BackgroundTask scheduled  (correlation_id = X, causation_id = domainEvent.id)
```

--------------------------------

### Domain Event Pattern

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/component-reference.md

Example of defining a domain event using DomainEventRecord and a payload dataclass.

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

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

--------------------------------

### DomainEventHandler Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Illustrates a DomainEventHandler that reacts to a domain event (AccountOpened). It shows how to publish integration events and schedule background tasks within the same transaction.

```python
from seedwork.application import DomainEventHandler
from .events import AccountOpened
from .request_context import correlation_id as _correlation_id

class AccountOpenedDomainEventHandler(DomainEventHandler[AccountOpened]):
    def __init__(
        self,
        publisher: IntegrationEventPublisher,
        task_scheduler: TaskScheduler,
    ) -> None:
        self._publisher = publisher
        self._task_scheduler = task_scheduler

    async def handle(self, event: AccountOpened) -> None:
        ie = AccountOpenedIntegrationEvent.from_domain_event(event)
        await self._publisher.publish([ie])
        task = SendWelcomeEmailTask.from_domain_event(event)
        await self._task_scheduler.schedule(task)
```

--------------------------------

### SendWelcomeEmailTask and Handler

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Example of a background task for sending a welcome email, including its handler. It demonstrates how to create a task from a domain event and the structure of a task handler.

```python
from __future__ import annotations
from typing import TYPE_CHECKING, ClassVar
from uuid import uuid4

from seedwork.application.background_tasks import BaseBackgroundTask
from .request_context import correlation_id as _correlation_id

if TYPE_CHECKING:
    from .events import AccountOpened


class SendWelcomeEmailTask(BaseBackgroundTask):
    TYPE: ClassVar[str] = "send_welcome_email"

    @classmethod
    def from_domain_event(cls, event: AccountOpened) -> SendWelcomeEmailTask:
        return cls(
            type=cls.TYPE,
            payload={"account_id": event.aggregate_id},
            correlation_id=_correlation_id.get(str(uuid4())),
            causation_id=event.id,
        )


class SendWelcomeEmailTaskHandler(TaskHandler[SendWelcomeEmailTask]):
    async def handle(self, task: SendWelcomeEmailTask) -> None:
        # idempotent — may be retried
        ...
```

--------------------------------

### Result Type Usage Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Illustrates how to check the outcome of a command dispatch using the Result type, distinguishing between success and failure.

```python
# Command bus returns Result — inspect at the entry point
result = await command_bus.dispatch(command)

if result.is_failed:      # @property bool
    return error_response(result.errors)

# also: result.is_ok
```

--------------------------------

### Domain Error Examples

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Custom domain error classes for business rule violations like insufficient funds and account not found.

```python
from seedwork.domain import DomainError

class InsufficientFundsError(DomainError):
    def __init__(self) -> None:
        super().__init__("Insufficient funds", "INSUFFICIENT_FUNDS")

class AccountNotFoundError(DomainError):
    def __init__(self, account_id: str) -> None:
        super().__init__(f"Account {account_id} not found", "ACCOUNT_NOT_FOUND")
```

--------------------------------

### Integration Event Concrete Event Example

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Defines a concrete integration event, inheriting from BaseIntegrationEvent and providing a factory method to convert from a domain event.

```python
from __future__ import annotations
from typing import TYPE_CHECKING, ClassVar
from uuid import uuid4

from seedwork.application.integration_events import BaseIntegrationEvent
from .request_context import correlation_id as _correlation_id

if TYPE_CHECKING:
    from .events import AccountOpened


class AccountOpenedIntegrationEvent(BaseIntegrationEvent):
    TYPE: ClassVar[str] = "bank_account.account_opened"
    VERSION: ClassVar[str] = "1.0"

    @classmethod
    def from_domain_event(cls, event: AccountOpened) -> AccountOpenedIntegrationEvent:
        return cls(
            type=cls.TYPE,
            version=cls.VERSION,
            aggregate_id=event.aggregate_id,
            payload={
                "account_id": event.aggregate_id,
                "initial_balance": event.payload.initial_balance,
                "currency": event.payload.currency,
            },
            correlation_id=_correlation_id.get(str(uuid4())),  # from execution context
            causation_id=event.id,                              # domain event that caused this
        )
```

--------------------------------

### Spy Usage in Tests

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Example of using spy objects for testing command dispatching and task scheduling.

```python
# spy usage in tests
publisher = InMemoryIntegrationEventPublisher()
scheduler = InMemoryTaskScheduler()
scheduler.register(SendWelcomeEmailTask.TYPE, SendWelcomeEmailTaskHandler())

await command_bus.dispatch(OpenAccountCommand(...))

assert len(publisher.published) == 1
assert publisher.published[0].type == "bank_account.account_opened"

await scheduler.execute_scheduled()   # executes all scheduled tasks in-process
assert len(scheduler.scheduled) == 0  # tasks were consumed
```

--------------------------------

### Outbox Pattern Sequence Diagram

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/architecture.md

Sequence diagram illustrating the Outbox Pattern, showing the interaction between CommandHandler, Database, Publisher/Worker, and Broker/TaskHandler.

```mermaid
sequenceDiagram
    participant Handler as CommandHandler
    participant DB as Database
    participant Publisher as Publisher / Worker
    participant Dest as Broker / TaskHandler

    Note over Handler,DB: Inside transaction
    Handler->>DB: save aggregate
    Handler->>DB: insert outbox record (status = pending)
    Note over Handler,DB: Commit

    Publisher->>DB: SELECT FOR UPDATE SKIP LOCKED (status = pending)
    DB-->>Publisher: outbox record
    Publisher->>Dest: deliver
    Dest-->>Publisher: ack
    Publisher->>DB: UPDATE status = published / delivered
```

--------------------------------

### Protocol vs. ABC for Ports

Source: https://github.com/aseguragonzalez/python-seedwork/blob/main/docs/coding-standards.md

Illustrates the use of Protocols for defining repository ports to achieve structural subtyping and zero coupling.

```python
# Port — Protocol, zero coupling to domain
# 'pass' body: Repository[BankAccountId, BankAccount] already carries the contract
class BankAccountRepository(Repository[BankAccountId, BankAccount], Protocol):
    pass

# Adapter — no import of the Protocol required
class SqlAlchemyBankAccountRepository:
    async def find_by_id(self, id: BankAccountId) -> BankAccount | None:
        ...
```