Port¶

Ports define interfaces between the application layer and the outside world. They abstract infrastructure details and enable easy testing through spy implementations.

Import¶

from aod.application import Port

Port Types¶

The framework provides two categories of ports:

Category	Purpose	Examples
Handler ports	Database operations via handlers	`CommandPort[T]`, `QueryPort[T]`
Service ports	Cross-cutting concerns	`Logger`, `EventBus`, `Cache`

Handler Ports: CommandPort / QueryPort¶

Handler ports are the primary way UseCases interact with the database. UseCases declare CommandPort[Command] and QueryPort[Query] as fields and call self.port.handle(...) to delegate operations:

from aod.application import UseCase, CommandPort, QueryPort, Command, Query

class PlaceOrder(Command[Order, None]):
    order_id: str
    total: float

class GetOrder(Query[Order, Order | None]):
    order_id: str

class OrderUseCase(UseCase):
    place_order: CommandPort[PlaceOrder]
    get_order: QueryPort[GetOrder]

    def run(self, order_id: str, total: float) -> None:
        self.place_order.handle(PlaceOrder(
            order_id=order_id, total=total,
        ))

Infrastructure provides concrete implementations via CommandHandler[C] and QueryHandler[Q].

Service Ports¶

Service ports abstract cross-cutting infrastructure concerns. Subclass Port and define abstract methods:

from aod.application import Port

class UserClient(Port):
    def save(self, user: User) -> None: ...
    def find(self, user_id: str) -> User | None: ...
    def delete(self, user_id: str) -> None: ...

Class Reference¶

`Port`¶

Base class for all ports.

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

Parameter	Type	Description
`*fields`	Any	All declared fields as keyword arguments

Key characteristics:

Mutable inside methods — Port methods can mutate fields because they execute inside a mutation context
Immutable from outside — Field mutation outside a method raises MutationForbiddenException
Abstract methods — Methods decorated with @abstractmethod are enforced (skipped by method wrapping)
Supports @abstractmethod — Use it for required interface methods

class UserClient(Port):
    calls: list[str] = []

    def save(self, user: User) -> None:
        self.calls.append(f"save:{user.id}")  # Allowed inside methods

    def get_calls(self) -> list[str]:
        return list(self.calls)

client = UserClient()
client.calls = []  # MutationForbiddenException — mutation outside method

Built-in Ports¶

`Logger`¶

Logging interface with four severity levels.

Import: from aod.application import Logger

Constructor: Takes no required parameters. Any declared Port fields can be passed as keyword arguments.

Abstract methods:

Method	Parameters	Returns	Description
`debug`	`msg: str`, `**context: object`	`None`	Log a debug-level message
`info`	`msg: str`, `**context: object`	`None`	Log an info-level message
`warning`	`msg: str`, `**context: object`	`None`	Log a warning-level message
`error`	`msg: str`, `**context: object`	`None`	Log an error-level message

Parameters for all methods:

Parameter	Type	Description
`msg`	`str`	The log message
`**context`	`object`	Additional structured context as keyword arguments

from aod.application import Logger

class MyLogger(Logger):
    def debug(self, msg: str, **context: object) -> None:
        print(f"DEBUG: {msg}", context)
    def info(self, msg: str, **context: object) -> None:
        print(f"INFO: {msg}", context)
    def warning(self, msg: str, **context: object) -> None:
        print(f"WARN: {msg}", context)
    def error(self, msg: str, **context: object) -> None:
        print(f"ERROR: {msg}", context)

`EventBus`¶

Interface for publishing domain events.

Import: from aod.application import EventBus

Constructor: Takes no required parameters. Any declared Port fields can be passed as keyword arguments.

Abstract methods:

Method	Parameters	Returns	Description
`publish`	`*events: Event`	`None`	Publish one or more domain events

Parameters:

Parameter	Type	Description
`*events`	`Event`	One or more domain event instances to publish

from aod.application import EventBus

class MyEventBus(EventBus):
    def publish(self, *events: Event) -> None:
        for event in events:
            print(f"Published: {event}")

`Cache`¶

Interface for caching query results with automatic invalidation.

Import: from aod.application.cache import Cache

Cache provides the backing store interface. Higher-level cache behavior (read-through, invalidation, deferred writes) is built on top and activated via CacheManager context — the container wraps operations automatically when caches are configured.

Constructor:

Parameter	Type	Description
`keys`	`list[CacheKey]`	Cache key definitions that map queries and invalidations

Abstract methods:

Method	Parameters	Returns	Description
`get`	`key: str`	`Any`	Retrieve a value by key. Returns `None` if not found
`set`	`key: str`, `value: Any`, `ttl: float \\| None = None`	`None`	Store a value with optional TTL in seconds
`delete`	`key: str`	`None`	Remove a value by key

set parameters:

Parameter	Type	Description
`key`	`str`	Cache key
`value`	`Any`	Value to store
`ttl`	`float \\| None`	Time-to-live in seconds. `None` means no expiration

ContractCacheKey¶

ContractCacheKey[TQuery] defines how a Query maps to a cache key and which Command types invalidate it.

Import: from aod.application.cache import ContractCacheKey

Abstract methods:

Method	Parameters	Returns	Description
`key`	`query: TQuery`	`str`	Map a query instance to a cache key string
`invalidate`	(none)	`list[CacheInvalidation]`	Return invalidations — which commands clear which keys

OperationCacheKey¶

OperationCacheKey[TOperation] defines cache keys for UseCase/Projection-based operations, mapping an operation's entry-point parameters to a cache key.

Import: from aod.application.cache import OperationCacheKey

CacheInvalidation¶

CacheInvalidation is the base invalidation descriptor. Use ContractCacheInvalidation (for Command-based invalidation) or OperationCacheInvalidation (for operation-based invalidation).

Import: from aod.application.cache import ContractCacheInvalidation

Parameter	Type	Description
`target_type`	`type[Command]`	The Command type that triggers invalidation
`key_fn`	`Callable[[Any], str]`	Function taking the command and returning the cache key to delete

Applying Cache¶

Cache is activated via CacheManager context — the container wraps operations automatically when caches are configured in AdapterContainer:

from aod.application.cache import Cache, ContractCacheKey, ContractCacheInvalidation, OperationCacheKey, OperationCacheInvalidation
from aod.infrastructure import QueryHandler, AdapterContainer

# ── Handler-level: cache query results, invalidate on commands ──

class GetUser(Query[User, User | None]):
    user_id: int

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

class DeleteUser(Command[User, None]):
    user_id: int

class UserCacheKey(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}"),
            ContractCacheInvalidation(target_type=DeleteUser, key_fn=lambda c: f"user:{c.user_id}"),
        ]

class RedisCache(Cache):
    def get(self, key: str) -> Any:
        return self.redis.get(key)
    def set(self, key: str, value: Any, ttl: float | None = None) -> None:
        self.redis.set(key, value, ex=ttl)
    def delete(self, key: str) -> None:
        self.redis.delete(key)

class GetUserHandler(QueryHandler[GetUser]):
    session: PsqlSession

    def handle(self, query: GetUser) -> User | None:
        return self.session.query(query)

# ── Operation-level: cache the entire use case result ──

class LookupUser(OperationCacheKey[LookupUserUseCase]):
    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}",
            ),
        ]

# Both key types in the same cache instance
cache = RedisCache(keys=[UserCacheKey(), LookupUser()])

container = AdapterContainer(
    caches=[cache],
    handlers=[GetUserHandler],
)
use_case = container.adapt(MyUseCase)
use_case.run(...)  # cache context active automatically

When container.adapt() is called, the container wraps run()/read()/write() with a CacheManager context. Inside that context, handler calls check the cache before executing — read-through for queries, invalidation for commands.

Warning: AsyncCache instances (with async get, set, delete) are only compatible with async operations (AsyncUseCase, AsyncReadProjection, AsyncWriteProjection). If you pass an AsyncCache to a sync UseCase or sync Projection, cache reads silently return None and cache writes are skipped — the framework cannot await in a sync context.

Async Ports¶

For async operations, import from aod.application.async_:

from aod.application.async_ import Logger, EventBus, Cache

Each async variant has the same methods but declared as async:

Port	Import Path	Difference
`AsyncLogger`	`aod.application.async_`	All log methods are `async`
`AsyncEventBus`	`aod.application.async_`	`publish()` is `async`
`AsyncCache`	`aod.application.async_`	`get`, `set`, `delete` are `async`. Only works in async operations.

Warning: AsyncCache only functions inside AsyncUseCase, AsyncReadProjection, or AsyncWriteProjection. Sync UseCase/Projection silently skip async cache operations.

CacheKey and CacheInvalidation are plain data structures — import them from aod.application.cache regardless of sync or async use.

from aod.application.async_ import Logger

class AsyncMyLogger(Logger):
    async def debug(self, msg: str, **context: object) -> None:
        await log_system.debug(msg, **context)
    async def info(self, msg: str, **context: object) -> None:
        await log_system.info(msg, **context)
    async def warning(self, msg: str, **context: object) -> None:
        await log_system.warning(msg, **context)
    async def error(self, msg: str, **context: object) -> None:
        await log_system.error(msg, **context)

Testing Spy Classes¶

Ready-made spy implementations for the built-in ports:

from aod.testing.doubles import SpyLogger, SpyEventBus, SpyCache

logger = SpyLogger()
event_bus = SpyEventBus()
cache = SpyCache()

Async spy classes (same names, import from aod.testing.doubles.application.async_):

from aod.testing.doubles.application.async_ import SpyLogger, SpyEventBus, SpyCache

CQRS Pattern¶

The recommended pattern uses CommandPort[T] and QueryPort[T] instead of custom repository ports. This keeps the application layer decoupled from data access details:

from aod.application import CommandPort, QueryPort

class OrderUseCase(UseCase):
    place_order: CommandPort[PlaceOrder]
    get_order: QueryPort[GetOrder]

Custom service ports are suitable for non-database concerns like API clients and notifications:

API Client Port¶

class WeatherClient(Port):
    def get_temperature(self, city: str) -> float: ...
    def get_forecast(self, city: str, days: int) -> list[Forecast]: ...

Notification Port¶

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

Port¶

Import¶

Port Types¶

Handler Ports: CommandPort / QueryPort¶

Service Ports¶

Class Reference¶

`Port`¶

Built-in Ports¶

`Logger`¶

`EventBus`¶

`Cache`¶

ContractCacheKey¶

OperationCacheKey¶

CacheInvalidation¶

Applying Cache¶

Async Ports¶

Testing Spy Classes¶

CQRS Pattern¶

API Client Port¶

Notification Port¶

Next Steps¶

UseCase

Contracts

Handlers

Container

Port¶

Import¶

Port Types¶

Handler Ports: CommandPort / QueryPort¶

Service Ports¶

Class Reference¶

Port¶

Built-in Ports¶

Logger¶

EventBus¶

Cache¶

ContractCacheKey¶

OperationCacheKey¶

CacheInvalidation¶

Applying Cache¶

Async Ports¶

Testing Spy Classes¶

CQRS Pattern¶

API Client Port¶

Notification Port¶

Next Steps¶

UseCase

Contracts

Handlers

Container

`Port`¶

`Logger`¶

`EventBus`¶

`Cache`¶