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
Policy	Authorization service	Enforce access rules

Imports¶

from aod.application import (
    UseCase,
    AsyncUseCase,
    Port,
    Command,
    Query,
    CommandPort,
    QueryPort,
    Logger,
    EventBus,
)
from aod.application.policies import (
    PolicyContract,
    PolicyExpression,
    PolicyManager,
    PolicyPort,
)
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. The caller opens one transaction for each operation:

class CreateUser(UseCase):
    def run(self) -> None:
        pass

use_case = container.adapt(CreateUser)
with container.transaction(cache=container.cache_context()):
    use_case.run()

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 or container.cache_context(). The framework supports handler-level caching:

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.

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}"),
        ]

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: ...

# Query keys and command invalidations live in the same cache
cache = MyCache(keys=[UserById()])
container = AdapterContainer(caches=[cache])
use_case = container.adapt(MyUseCase)
with container.transaction(cache=container.cache_context()):
    use_case.run(...)

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.

Policy Authorization¶

Policies are independent application services for authorization. A PolicyContract carries the data needed to decide one authorization check; a PolicyHandler implements the check.

from aod.application.policies import PolicyContract, PolicyManager
from aod.application.policies import PolicyPort
from aod.infrastructure.policies import PolicyHandler

class OwnerContract(PolicyContract):
    user_id: str
    document_id: str

class OwnerPolicyHandler(PolicyHandler[OwnerContract]):
    documents: QueryPort[GetDocument]

    def handle(self, contract: OwnerContract) -> None:
        doc = self.documents.handle(GetDocument(document_id=contract.document_id))
        if doc.owner_id != contract.user_id:
            raise PermissionError("not the document owner")

# Enforcement is explicit — before or after a UseCase, inside the same Transaction
manager = PolicyManager(OwnerPolicyHandler(documents=query_handler))
with Transaction():
    manager.enforce(owner_contract)
    use_case.run()

See the Policy System spec for full details.

Event Collection¶

Events emitted during run() are collected while Transaction() is active:

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