UseCase¶

UseCases orchestrate domain objects through Ports. They are the central building block of the application layer, handling transaction management, logging, and event publishing automatically.

Commands and Queries are internal — created by the UseCase, not passed by the caller. Use Pydantic BaseModel subclasses for run() input.

Import¶

from aod.application import UseCase
from pydantic import BaseModel

For async operations:

from aod.application.async_ import UseCase

Basic Usage¶

Subclass UseCase, define CommandPort[Command] and QueryPort[Query] fields, and implement the run() method.

from aod.application import UseCase, CommandPort, Command
from pydantic import BaseModel

class CreateUserInput(BaseModel):
    user_id: str
    name: str
    email: str

class CreateUser(Command[User, None]):
    user_id: str
    name: str
    email: str

class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

    def run(self, dto: CreateUserInput) -> User:
        user = User(id=dto.user_id, name=dto.name, email=dto.email)
        self.save_user.handle(CreateUser(
            user_id=user.id, name=user.name, email=user.email,
        ))
        self._event_emitter.emit(UserCreated(user_id=dto.user_id))
        return user

uc = container.adapt(CreateUserUseCase)
user = uc.run(CreateUserInput(user_id="1", name="Alice", email="alice@example.com"))

Class Reference¶

`UseCase`¶

Base class for synchronous use cases. Inherits from BaseOperation.

Parameters (constructor — auto-generated by Pydantic based on declared fields):

Parameter	Type	Description
`*port_fields`	`Port` subclass	All declared Port fields as keyword arguments

Auto-wired fields:

Field	Type	Default	Description
`events`	`list[Event]`	`[]`	Collected events from last `run()` call. Read-only outside mutation context
`_event_emitter`	`EventEmitter`	`EventEmitter()`	Private event emitter for emitting events during `run()`
`_loggers`	`list[Logger \\| AsyncLogger]`	`[]`	Private list of declared logger ports
`_event_buses`	`list[EventBus \\| AsyncEventBus]`	`[]`	Private list of declared event bus ports

Optional ports (declare explicitly when needed):

class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]
    logger: Logger
    event_bus: EventBus

    def run(self, dto: CreateUserInput) -> User:
        ...

`run(self, *args, **kwargs) -> Any`¶

Abstract method. Subclasses define specific parameters. Values are passed here, not as class fields. The method is automatically wrapped to:

Begin an internal Transaction
Open an EventCollector context
Invoke the original run() body
Collect emitted events into self.events
On success: commit Transaction (flushing caches internally), log completion, publish events
On failure: rollback Transaction, log error, re-raise

Parameters: Defined by the subclass — any number of positional and keyword arguments representing input values.

Returns: Any return value defined by the subclass.

`AsyncUseCase`¶

Base class for asynchronous use cases. Inherits from BaseOperation.

Parameters (constructor):

Parameter	Type	Description
`*port_fields`	`Port` subclass	All declared Port fields as keyword arguments

Auto-wired fields: Same as UseCase.

`async run(self, *args, **kwargs) -> Any`¶

Async abstract method. Same wrapping behavior as sync run() but bridges sync/async calls via should_await internally.

Field Validation¶

All declared fields on a UseCase must be Port subclasses. Non-Port fields raise InvalidUseCasePortFieldError:

class CreateUser(UseCase):
    user_id: str  # InvalidUseCasePortFieldError — not a Port
    name: str     # InvalidUseCasePortFieldError — not a Port

Blocked field types (rejected even if they are Port-like):

Type	Reason
`Session`	UseCases should not depend on sessions directly
`AsyncSession`	UseCases should not depend on sessions directly
`BaseHandler`	Handlers belong in infrastructure
`AsyncBaseHandler`	Handlers belong in infrastructure

Use CommandPort[Command] or QueryPort[Query] instead:

from aod.application import CommandPort, Command

class CreateUser(Command[User, None]):
    user_id: str
    name: str

class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

    def run(self, user_id: str, name: str) -> None:
        user = User(id=user_id, name=name)
        self.save_user.handle(CreateUser(user_id=user_id, name=name))

Event Collection¶

Events emitted during run() are automatically collected. This includes events emitted directly by the UseCase via self._event_emitter.emit(...) and events emitted by any entity, value object, or service touched during execution.

class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

    def run(self, dto: CreateUserInput) -> None:
        user = User(id=dto.user_id)
        user.register()  # emits UserRegistered
        self.save_user.handle(CreateUser(user_id=user.id, name=dto.name, email=dto.email))
        self._event_emitter.emit(UserCreated(user_id=dto.user_id))

uc = CreateUserUseCase(save_user=handler)
uc.run(CreateUserInput(user_id="1", name="Alice", email="alice@example.com"))
assert len(uc.events) == 2  # UserRegistered + UserCreated
assert isinstance(uc.events[0], UserRegistered)
assert isinstance(uc.events[1], UserCreated)

The wrapper: 1. Opens an EventCollector context before run() executes 2. Collects all emitted events into self.events 3. Publishes events on the event bus after a successful commit 4. Replaces self.events on each new call to run()

If run() raises an exception, collected events are discarded and self.events is cleared:

Private Methods¶

UseCases can have private helper methods:

class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

    def _validate(self, name: str) -> bool:
        return len(name) > 0

    def run(self, dto: CreateUserInput) -> None:
        if not self._validate(dto.name):
            raise ValueError("Invalid name")
        user = User(id=dto.user_id, name=dto.name)
        self.save_user.handle(CreateUser(user_id=user.id, name=dto.name, email=dto.email))

Error Handling¶

The auto-wrapper handles errors:

If run() raises: Transaction is rolled back, error is logged, exception is re-raised
If commit() fails: Transaction is rolled back, error is logged, exception is re-raised

class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

    def run(self) -> None:
        raise ValueError("Something went wrong")

uc = CreateUserUseCase(save_user=handler)
try:
    uc.run()
except ValueError:
    pass

assert uc.events == []  # Events cleared on failure

Testing¶

Use spy_adapter_container for testing use cases instead of mocking ports manually:

from aod.testing.doubles import spy_adapter_container

container = spy_adapter_container(AdapterContainer(sessions={MySession}, handlers=[CreateUserHandler]))
use_case = container.adapt(CreateUserUseCase)

use_case.run(CreateUserInput(user_id="1", name="Alice", email="alice@example.com"))

assert use_case.events
assert container.get_handler(CreateUser).handle.called

Common Patterns¶

Command Use Case¶

class PlaceOrderInput(BaseModel):
    order_id: str
    total: float

class PlaceOrderUseCase(UseCase):
    place_order: CommandPort[PlaceOrder]

    def run(self, dto: PlaceOrderInput) -> None:
        order = Order(id=dto.order_id, total=dto.total)
        self.place_order.handle(PlaceOrder(order_id=dto.order_id, total=dto.total))
        self._event_emitter.emit(OrderPlaced(order_id=dto.order_id))

Query Use Case¶

class GetUserInput(BaseModel):
    user_id: str

class GetUserUseCase(UseCase):
    get_user: QueryPort[GetUser]

    def run(self, dto: GetUserInput) -> User | None:
        return self.get_user.handle(GetUser(user_id=dto.user_id))

Use Case with Custom Service Port¶

For non-database dependencies (API clients, notifications), use a custom Port subclass:

from aod.application import Port


class NotificationClient(Port):
    def send_email(self, to: str, subject: str, body: str) -> None: ...


class NotifyUserInput(BaseModel):
    user_id: str
    message: str

class NotifyUser(UseCase):
    notification: NotificationClient

    def run(self, dto: NotifyUserInput) -> None:
        user = User(id=dto.user_id)
        self.notification.send_email(to=user.email, subject="Alert", body=dto.message)

Use Case with Validation¶

Validation belongs in the domain, not the use case. Define constraints on the entity and let Pydantic enforce them:

from aod.domain.validation import field_invariance


class User(RootEntity):
    id: str
    name: str
    email: str

    @field_invariance("name")
    def name_required(cls, v: str) -> str:
        if not v:
            raise ValueError("Name is required")
        return v

    @field_invariance("email")
    def email_valid(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("Invalid email")
        return v


class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

    def run(self, dto: CreateUserInput) -> None:
        user = User(id=dto.user_id, name=dto.name, email=dto.email)
        self.save_user.handle(CreateUser(user_id=user.id, name=user.name, email=user.email))

UseCase¶

Import¶

Basic Usage¶

Class Reference¶

`UseCase`¶

`run(self, *args, **kwargs) -> Any`¶

`AsyncUseCase`¶

`async run(self, *args, **kwargs) -> Any`¶

Field Validation¶

Event Collection¶

Private Methods¶

Error Handling¶

Testing¶

Common Patterns¶

Command Use Case¶

Query Use Case¶

Use Case with Custom Service Port¶

Use Case with Validation¶

Next Steps¶

Port

Contracts

Handlers

Container

UseCase¶

Import¶

Basic Usage¶

Class Reference¶

UseCase¶

run(self, *args, **kwargs) -> Any¶

AsyncUseCase¶

async run(self, *args, **kwargs) -> Any¶

Field Validation¶

Event Collection¶

Private Methods¶

Error Handling¶

Testing¶

Common Patterns¶

Command Use Case¶

Query Use Case¶

Use Case with Custom Service Port¶

Use Case with Validation¶

Next Steps¶

Port

Contracts

Handlers

Container

`UseCase`¶

`run(self, *args, **kwargs) -> Any`¶

`AsyncUseCase`¶

`async run(self, *args, **kwargs) -> Any`¶