Application Layer¶

The application layer orchestrates domain objects through use cases. It defines interfaces (ports) for infrastructure and provides contracts for communication.

Building Blocks¶

Block	Description	Purpose
UseCase	Application operation	Orchestrate domain logic
Port	Interface definition	Abstract infrastructure
Command	Write operation	Request to change state
Query	Read operation	Request to read state
Handler	Command/Query processor	Execute contracts

Imports¶

from aod.application import (
    UseCase,
    AsyncUseCase,
    Port,
    Command,
    Query,
    CommandPort,
    QueryPort,
    Logger,
    EventBus,
)
from aod.application.cache import Cache, AsyncCache

Quick Example — CQRS¶

from aod.application import UseCase, Command, CommandPort

# Define a command (immutable write request)
class CreateUser(Command[User, None]):
    user_id: str
    name: str
    email: str

# Define a use case with a CommandPort
class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

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

# The container injects the matching CommandHandler
uc = container.adapt(CreateUserUseCase)
uc.run(user_id="1", name="Alice", email="alice@example.com")

Key Concepts¶

Port Types¶

Use case fields must be Port subclasses. The framework provides two kinds:

Kind	Purpose	Examples
Handler ports	Database operations	`CommandPort[T]`, `QueryPort[T]`
Service ports	External concerns	Custom `Port` subclasses

CommandPort / QueryPort for Database Operations¶

# Correct: CommandPort as field, values in run()
class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

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

# Wrong: values as fields
class CreateUserUseCase(UseCase):
    user_id: int  # InvalidUseCasePortFieldError!
    name: str     # InvalidUseCasePortFieldError!

Custom Service Port for External Concerns¶

For non-database dependencies (API clients, notification services, etc.), create custom Port subclasses:

from aod.application import Port, UseCase


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


class NotifyUser(UseCase):
    notification: NotificationClient

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

Blocked Field Types¶

Session and AsyncSession are rejected on UseCases:

from aod.infrastructure import Session

class CreateUserUseCase(UseCase):
    session: Session  # InvalidUseCasePortFieldError!

    def run(self) -> None:
        pass

Instead, use CommandPort[Command] or QueryPort[Query]:

from aod.application import CommandPort, Command

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

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

    def run(self) -> None:
        pass

Auto-Wired Fields¶

Use cases do not expose a transaction field. A fresh Transaction is created internally for each run() call:

class CreateUser(UseCase):
    # A Transaction is created internally for each run() call,
    # managing begin/commit/rollback and cache flushing automatically.

    def run(self) -> None:
        pass

Logger and EventBus are not auto-wired. Declare them as normal ports when you need them:

class CreateUser(UseCase):
    logger: Logger
    event_bus: EventBus

    def run(self, user_id: str, name: str) -> None:
        ...

Cache¶

Cache is activated via CacheManager context and wired automatically by the container. The framework supports two caching strategies:

Handler-level (recommended): ContractCacheKey[TQuery] maps a Query to a cache key and lists which Commands invalidate it. Read-through on QueryHandlers, invalidation on CommandHandlers — no code changes needed in handlers.

Operation-level: OperationCacheKey[TOperation] maps a UseCase/Projection entry-point call to a cache key, for caching the entire operation result.

from aod.application.cache import (
    Cache, AsyncCache,
    ContractCacheKey, ContractCacheInvalidation,
    OperationCacheKey, OperationCacheInvalidation,
)

# Handler-level: cache query results, invalidate on commands
class UserById(ContractCacheKey[GetUser]):
    def key(self, query: GetUser) -> str:
        return f"user:{query.user_id}"
    def invalidate(self) -> list[CacheInvalidation]:
        return [
            ContractCacheInvalidation(target_type=CreateUser, key_fn=lambda c: f"user:{c.name}"),
        ]

# Operation-level: cache the entire use case result
class UserLookup(OperationCacheKey[GetUserUseCase]):
    def key(self, user_id: int) -> str:
        return f"lookup:{user_id}"
    def invalidate(self) -> list[CacheInvalidation]:
        return [
            OperationCacheInvalidation(
                target_type=UpdateUserUseCase,
                key_fn=lambda user_id, name: f"lookup:{user_id}",
            ),
        ]

class MyCache(Cache):
    def get(self, key: str) -> object | None: ...
    def set(self, key: str, value: object, ttl: int | None = None) -> None: ...
    def delete(self, key: str) -> None: ...

# Both key types live in the same cache
cache = MyCache(keys=[UserById(), UserLookup()])
container = AdapterContainer(caches=[cache])
use_case = container.adapt(MyUseCase)
use_case.run(...)  # cache context active automatically

Warning: AsyncCache solo funciona en contextos async (AsyncUseCase, AsyncReadProjection, AsyncWriteProjection). En use cases y projections sincrónos, las lecturas devuelven None y las escrituras se descartan silenciosamente.

Event Collection¶

Events emitted during run() are automatically collected:

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, email=""))
        self._event_emitter.emit(UserCreated(user_id=user_id))

uc = CreateUserUseCase(save_user=handler)
uc.run(user_id="1", name="Alice")
assert len(uc.events) == 1