UseCase¶

UseCases orchestrate domain objects through Ports. They execute application logic and emit events; transaction management is explicit and belongs to the caller.

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, Transaction
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)
with Transaction(cache=container.cache_context()):
    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()`

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 not wrapped:

Open Transaction() around the call when transactional behavior is needed
Invoke the wrapped run() body
The entrypoint captures events into self.events
On success, the transaction commits registered sessions
On failure, the transaction rolls back and re-raises

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. Surround the call with AsyncTransaction(cache=...); sync/async adapters are bridged 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 collected while Transaction() is active. 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)
with Transaction():
    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 transaction: 1. Opens an EventCollector context before run() executes 2. The wrapped entrypoint captures its emitted events into self.events 3. The operation notifies its logger and event bus after success 4. Replaces self.events on each entrypoint call

If run() raises an exception, the transaction rolls back and self.events still contains the events emitted before the failure:

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 reports the error and re-raises it

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

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

uc = CreateUserUseCase(save_user=handler)
try:
with Transaction():
    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`¶