Handlers¶
Handlers process commands and queries. They bridge the gap between the application and infrastructure layers, implementing CommandPort[T] and QueryPort[T] from the application layer.
UseCases declare CommandPort[Command] / QueryPort[Query] as fields; infrastructure handlers provide the concrete implementation that is injected by the container.
Import¶
For async operations:
Basic Usage¶
Subclass CommandHandler or QueryHandler parameterized by your contract type, and implement the handle() method:
from aod.infrastructure import CommandHandler, QueryHandler, Session
class PostgresSession(Session):
def execute(self, operation: object) -> None: ...
def query(self, operation: object) -> object: ...
class CreateUserHandler(CommandHandler[CreateUser]):
session: PostgresSession
def handle(self, command: CreateUser) -> None:
user = User(id=command.user_id, name=command.name, email=command.email)
self.session.execute(user)
class GetUserHandler(QueryHandler[GetUser]):
session: PostgresSession
def handle(self, query: GetUser) -> User | None:
return self.session.query(f"SELECT * FROM users WHERE id = {query.user_id}")
Class Reference¶
BaseHandler¶
Base class for all handlers. Provides mutation-guarded behaviour. Has no intrinsic session field — subclasses declare their own session fields with concrete types.
Constructor parameters:
| Parameter | Type | Description |
|---|---|---|
**fields |
Any | Fields declared on the handler subclass (e.g. session=PostgresSession()) |
AsyncBaseHandler¶
Async variant of BaseHandler. Same as BaseHandler — no inherited session field. Subclasses declare their own.
CommandHandler[TCommand]¶
Sync command handler. Inherits from BaseHandler, CommandPort (which is HandlerProtocol(Port)), and Generic[TCommand].
Type Parameters:
| Parameter | Constraint | Description |
|---|---|---|
TCommand |
Must be a Command subclass |
The command type this handler processes |
handle(self, command: TCommand) -> object¶
Abstract method. Implement to process a command.
Parameters:
| Parameter | Type | Description |
|---|---|---|
command |
TCommand |
The command instance to process |
Returns: Any object.
QueryHandler[TQuery]¶
Sync query handler. Inherits from BaseHandler, QueryPort (which is HandlerProtocol(Port)), and Generic[TQuery].
Type Parameters:
| Parameter | Constraint | Description |
|---|---|---|
TQuery |
Must be a Query subclass |
The query type this handler processes |
handle(self, query: TQuery) -> object¶
Abstract method. Implement to process a query.
Parameters:
| Parameter | Type | Description |
|---|---|---|
query |
TQuery |
The query instance to process |
Returns: Any object.
AsyncCommandHandler[TCommand]¶
Async command handler. Inherits from AsyncBaseHandler, AsyncCommandPort (which is HandlerProtocol(Port)), and Generic[TCommand].
async handle(self, command: TCommand) -> object¶
Async abstract method. Implement to process a command asynchronously.
AsyncQueryHandler[TQuery]¶
Async query handler. Inherits from AsyncBaseHandler, AsyncQueryPort (which is HandlerProtocol(Port)), and Generic[TQuery].
async handle(self, query: TQuery) -> object¶
Async abstract method. Implement to process a query asynchronously.
PolicyHandler[TPolicyContract]¶
Sync policy handler. Inherits from BaseOperation, PolicyPort (which is HandlerProtocol(Port)), and Generic[TPolicyContract]. Unlike command/query handlers, PolicyHandler does not declare a session field — it accesses data via QueryPort/CommandPort fields.
Type Parameters:
| Parameter | Constraint | Description |
|---|---|---|
TPolicyContract |
Must be a PolicyContract subclass |
The policy contract type this handler processes |
handle(self, contract: TPolicyContract) -> None¶
Abstract method. Implement to enforce an authorization decision. Raise an exception to deny access.
Parameters:
| Parameter | Type | Description |
|---|---|---|
contract |
TPolicyContract |
The policy contract instance to evaluate |
Returns: None.
AsyncPolicyHandler[TPolicyContract]¶
Async policy handler. Same as PolicyHandler but with an async handle method.
async handle(self, contract: TPolicyContract) -> None¶
Async abstract method. Implement to enforce an authorization decision asynchronously.
Generic Type Binding¶
Handlers are generic over their contract type. The generic argument is used at runtime to match commands/queries:
class CreateUserHandler(CommandHandler[CreateUser]):
session: PostgresSession
def handle(self, command: CreateUser) -> None:
pass # This handler only accepts CreateUser commands
Session Field¶
Handlers include a session field for database access:
class CreateUserHandler(CommandHandler[CreateUser]):
session: PostgresSession # Injected by the container
def handle(self, command: CreateUser) -> None:
self.session.execute(...)
Async Handlers¶
from aod.infrastructure import AsyncCommandHandler, AsyncQueryHandler, AsyncSession
class AsyncPostgresSession(AsyncSession):
async def execute(self, operation: object) -> None: ...
async def query(self, operation: object) -> object: ...
class AsyncCreateUserHandler(AsyncCommandHandler[CreateUser]):
session: AsyncPostgresSession
async def handle(self, command: CreateUser) -> None:
user = User(id=command.user_id, name=command.name)
await self.session.execute(user)
class AsyncGetUserHandler(AsyncQueryHandler[GetUser]):
session: AsyncPostgresSession
async def handle(self, query: GetUser) -> User | None:
return await self.session.query(...)
Handler Registration¶
Handlers are registered in AdapterContainer:
from aod.infrastructure import AdapterContainer
container = AdapterContainer(sessions={PostgresSession}, handlers=[CreateUserHandler, GetUserHandler])
Handler Discovery¶
The container discovers and instantiates handlers by contract type:
container = AdapterContainer()
handler = container.get_handler(CreateUser) # Returns CreateUserHandler instance
result = handler.handle(CreateUser(...))
Testing¶
Use spy_session for handler testing:
from aod.testing.doubles import spy_session
session = spy_session(MySession)()
handler = CreateUserHandler(session=session)
handler.handle(CreateUser(user_id="1", name="Alice"))
assert handler.handle.called
Common Patterns¶
Command Handler¶
class PlaceOrderHandler(CommandHandler[PlaceOrder]):
session: PostgresSession
def handle(self, command: PlaceOrder) -> None:
order = Order(id=command.order_id, total=command.total)
self.session.execute(order)
Query Handler¶
class GetUserHandler(QueryHandler[GetUser]):
session: PostgresSession
def handle(self, query: GetUser) -> User | None:
return self.session.query(f"SELECT * FROM users WHERE id = {query.user_id}")
Handler with Validation¶
Validation belongs in the domain, not the handler. The handler delegates to domain objects that enforce their own constraints:
class CreateUserHandler(CommandHandler[CreateUser]):
session: PostgresSession
def handle(self, command: CreateUser) -> None:
user = User(id=command.user_id, name=command.name, email=command.email)
self.session.execute(user)
Handler with Multiple Dependencies¶
class CreateUserHandler(CommandHandler[CreateUser]):
session: PostgresSession
validator: UserValidator
def handle(self, command: CreateUser) -> None:
self.validator.validate(command)
user = User(id=command.user_id, name=command.name)
self.session.execute(user)