API Reference¶

Comprehensive reference for all publicly importable classes, grouped by layer.

Domain Layer¶

from aod.domain import Entity, RootEntity, ValueObject, Service
from aod.domain import Field, PrivateField
from aod.events import Event, IntegrationEvent
from aod.domain.validation import AfterValidator, BeforeValidator, field_invariance, invariance, mutable, get_base_model

Entity¶

class Entity(ReconstructMixin, BaseGuarded)

Base class for mutable domain objects with identity.

Constructor¶

Entity(**fields)

Methods¶

Mutation Rules¶

• Public methods can mutate fields (auto-wrapped with mutation context).
• Direct field mutation from outside raises MutationForbiddenException.
• _event_emitter is available as a PrivateField.

RootEntity¶

class RootEntity(Entity)

Aggregate root marker class. Inherits all Entity behavior.

Constructor¶

Same as Entity.

Constraints¶

• Cannot be nested as a field type in other entities (enforced by BoundedContext type checking).
• Services and Commands/Queries can reference RootEntity but not non-root Entity.

ValueObject¶

class ValueObject(ReconstructMixin, BaseSealed)

Immutable, identity-less value object.

Constructor¶

ValueObject(**fields)

Methods¶

Constraints¶

• Fields cannot reference Entity or RootEntity (enforced by BoundedContext).
• Mutation is always blocked (BaseSealed).

Service¶

class Service(BaseBehaviour)

Stateless domain service with event emission capability.

Constructor¶

Service(**fields)

Methods¶

Public methods are auto-wrapped with mutation context (can mutate fields).

Constraints¶

• Method parameters and return types cannot reference non-root Entity (enforced by BoundedContext).
• _event_emitter is available as a PrivateField.

Event¶

class Event(BaseSealed)

Immutable domain event with auto-timestamp.

Constructor¶

Event(**fields)

Fields¶

IntegrationEvent¶

class IntegrationEvent(Event)

Marker subclass of Event for events that cross bounded context boundaries.

Constructor¶

Same as Event. Inherits emitted_at auto-timestamp.

Schema Layer¶

Schema classes for introspection and documentation generation.

from aod.schema import App, AutoDoc, BoundedContext, Infrastructure, Module

BoundedContext¶

class BoundedContext

Groups related domain types together.

Constructor¶

BoundedContext(
    aggregate_roots: Iterable[RootEntityType] | None = None,
    services: Iterable[ServiceType] | None = None,
    use_cases: Iterable[UseCaseType] | None = None,
    *,
    name: str | None = None,
)

Attributes¶

Type Checking¶

When constructed, runs validation: - All aggregate_roots must be RootEntity subclasses. - All services must be Service subclasses. - Recursively discovers nested entities and value objects from root entity fields. - Validates: no non-root entities nested in root entities, no entities in value objects, no non-root entities in service method signatures.

App¶

class App

Top-level application container.

Constructor¶

App(name: str, modules: Iterable[Module], description: str = "")

Validation¶

Raises DuplicateDomainTypeError if the same entity or service class appears in multiple contexts across modules.

Module¶

class Module

Groups a bounded context with its infrastructure layer.

Constructor¶

Module(name: str, context: BoundedContext, infrastructure: Infrastructure)

Validation¶

Raises MissingHandlerError if any contract has no matching handler. Raises MissingPortError if any port has no implementation.

Infrastructure¶

class Infrastructure

Configuration for infrastructure handlers, ports, and sessions.

Constructor¶

Infrastructure(handlers: Iterable[type] | None = None, ports: Iterable[type] | None = None, sessions: Iterable[type] | None = None)

AutoDoc¶

class AutoDoc

Generates a complete zensical documentation site from an App.

Constructor¶

AutoDoc(app: App, output_dir: str | Path, site_name: str | None = None, site_description: str | None = None, repo_url: str | None = None)

Domain Layer (continued)¶

Field¶

Field(
    default: Any = ...,
    *,
    default_factory: Callable[[], Any] | None = ...,
    gt: SupportsGt | None = ...,
    ge: SupportsGe | None = ...,
    lt: SupportsLt | None = ...,
    le: SupportsLe | None = ...,
    multiple_of: float | None = ...,
    strict: bool | None = ...,
    min_length: int | None = ...,
    max_length: int | None = ...,
    pattern: str | re.Pattern[str] | None = ...,
    allow_inf_nan: bool | None = ...,
    max_digits: int | None = ...,
    decimal_places: int | None = ...,
    init: bool = True,
) -> Any

Pydantic field wrapper with all standard constraints.

PrivateField¶

PrivateField(default: Any = Unset(), *, default_factory: Callable[[], Any] | None = None) -> Any

Pydantic PrivateAttr wrapper. Fields are not part of the public API surface — excluded from validation models, serialization, and __init__.

Validation Decorators¶

from aod.domain.validation import field_invariance, invariance, AfterValidator, BeforeValidator, mutable

Application Layer¶

from aod.application import UseCase, Port, Logger, EventBus, Cache, CacheManager, CacheKey, CacheInvalidation, ContractCacheKey, OperationCacheKey, ContractCacheInvalidation, OperationCacheInvalidation, Command, Query
from aod.application import CommandPort, QueryPort
from aod.application import ApplicationException
from aod.application.async_ import UseCase, Logger, EventBus, Cache
from aod.application.async_ import CommandPort, QueryPort

UseCase¶

class UseCase(BaseOperation)

Base class for synchronous application use cases.

Constructor¶

UseCase(**fields)

Fields¶

Methods¶

Auto-Wrapping Behavior¶

When run() is called:

1. uow.begin() starts a transaction.
2. Events are collected via EventCollector during execution.
3. On success: uow.commit(), events logged on each declared logger, events published on each declared event bus.
4. On failure: uow.rollback(), exception logged on each declared logger, exception re-raised.
5. If commit fails: uow.rollback(), commit failure logged on each declared logger, exception re-raised.

Field Validation¶

• Fields must be Port subclasses.
• Recommended field types: Commandport[TCommand] and QueryPort[TQuery].
• Session and AsyncSession are not allowed (raise InvalidUseCasePortFieldError).
• BaseHandler and AsyncBaseHandler are also rejected.

AsyncUseCase¶

class AsyncUseCase(BaseOperation)

Constructor¶

AsyncUseCase(**fields)

Methods¶

Port¶

class Port(BaseGuarded)

Abstract base class for defining dependency interfaces.

• No constructor parameters.
• Subclasses declare abstract methods and fields.
• Public methods are auto-wrapped with mutation context.

CommandPort¶

class CommandPort(HandlerProtocol, Generic[TCommand])

Application-layer handler port for write operations. UseCases declare CommandPort[Command] as fields.

Methods¶

Type Parameters¶

Infrastructure provides concrete implementations via CommandHandler[C].

QueryPort¶

class QueryPort(HandlerProtocol, Generic[TQuery])

Application-layer handler port for read operations. UseCases declare QueryPort[Query] as fields.

Methods¶

Type Parameters¶

Infrastructure provides concrete implementations via QueryHandler[Q].

Logger¶

class Logger(Port)

Synchronous logging port.

Methods¶

Parameters¶

AsyncLogger¶

from aod.application.async_ import Logger

Same interface as Logger but all methods are async.

EventBus¶

class EventBus(Port)

Synchronous event publishing port.

Methods¶

Parameters¶

AsyncEventBus¶

from aod.application.async_ import EventBus

Same interface but publish is async.

Cache¶

class Cache(Port)

Synchronous cache port.

Methods¶

Parameters¶

CacheManager¶

from aod.application import CacheManager
from aod.application.cache import CacheManager

CacheManager is a context manager that activates cache context for the duration of a block. Wraps a list of Cache/AsyncCache instances. Inside the context, get_cache_context() returns a context that routes cache operations to the registered caches.

with CacheManager(cache):
    result = use_case.run(...)  # handler reads through cache

The AdapterContainer.adapt() method wraps operations with CacheManager automatically when caches are configured — you rarely need it directly.

AsyncCache¶

from aod.application.async_ import Cache

Same interface but get, set, and delete are async.

Warning: AsyncCache only works in async operations (AsyncUseCase, AsyncReadProjection, AsyncWriteProjection). Sync UseCase/Projection silently skip async cache operations — reads return None and writes are discarded.

CacheKey¶

from aod.application import CacheKey

CacheKey(BaseGuarded) — Abstract base for cache key definitions. Subclass via ContractCacheKey (for Query/Command patterns) or OperationCacheKey (for operation-based patterns).

ContractCacheKey¶

from aod.application import ContractCacheKey

ContractCacheKey(CacheKey, Generic[TQuery]) — Maps a Query to a cache key and lists which Command types invalidate it.

Abstract Methods¶

OperationCacheKey¶

from aod.application import OperationCacheKey

OperationCacheKey(CacheKey, Generic[TOperation]) — Maps a UseCase/Projection operation to a cache key and lists which operations invalidate it.

CacheInvalidation¶

from aod.application import CacheInvalidation

CacheInvalidation(BaseSealed) — Base invalidation descriptor. Subclass via ContractCacheInvalidation or OperationCacheInvalidation.

ContractCacheInvalidation¶

from aod.application import ContractCacheInvalidation

Frozen dataclass pairing a Command type with a key extraction function. Used inside ContractCacheKey.invalidate().

Fields¶

OperationCacheInvalidation¶

from aod.application import OperationCacheInvalidation

Frozen dataclass pairing an operation type with a key extraction function. Used inside OperationCacheKey.invalidate().

Command¶

class Command(BaseSealed, Generic[TEntity, TResult])

Immutable command contract for write operations.

Constructor¶

Command(**fields)

Type Parameters¶

Constraints¶

• Fields cannot reference non-root Entity types (even inside list[Entity], Optional[Entity], etc.).
• The generic TEntity argument is validated to be a RootEntity subclass at class creation time.

Query¶

class Query(BaseSealed, Generic[TEntity, TResult])

Immutable query contract for read operations.

Constructor¶

Query(**fields)

Type Parameters¶

Constraints¶

• Same field restrictions as Command.
• TResult is validated to contain at least one RootEntity type.

Infrastructure Layer¶

from aod.infrastructure import Session, AsyncSession
from aod.infrastructure import ReadProjection, WriteProjection, Projection
from aod.infrastructure import AsyncReadProjection, AsyncWriteProjection, AsyncProjection
from aod.infrastructure import CommandHandler, QueryHandler
from aod.infrastructure import AsyncCommandHandler, AsyncQueryHandler
from aod.infrastructure import AdapterContainer
from aod.infrastructure import InfrastructureException

Session¶

class Session(Port)

Synchronous database session abstraction.

Methods¶

AsyncSession¶

class AsyncSession(Port)

Same interface as Session but execute, query, begin, commit, rollback, close are async. is_dirty() is sync.

ReadProjection¶

class ReadProjection(ReadProjectionBase)

Synchronous read projection.

Constructor¶

ReadProjection(**fields)

Methods¶

WriteProjection¶

class WriteProjection(WriteProjectionBase)

Synchronous write projection.

Constructor¶

WriteProjection(**fields)

Methods¶

Projection¶

class Projection(ReadProjection, WriteProjection)

Combined read and write projection.

Constructor¶

Same as ReadProjection and WriteProjection.

Methods¶

Both read(self, model: Any) -> Any and write(self, model: Any) -> Any.

AsyncReadProjection¶

class AsyncReadProjection(AsyncReadProjectionBase)

Constructor¶

AsyncReadProjection(**fields)

Methods¶

AsyncWriteProjection¶

class AsyncWriteProjection(AsyncWriteProjectionBase)

Constructor¶

AsyncWriteProjection(**fields) — Accepts fields with concrete session types (e.g., session: AsyncPostgresSession) and port dependencies.

Methods¶

AsyncProjection¶

class AsyncProjection(AsyncReadProjection, AsyncWriteProjection)

Combined async read and write projection. Both read and write are async.

CommandHandler¶

class CommandHandler(BaseHandler, AppCommandHandler, Generic[TCommand])

Synchronous command handler for a specific command type.

Constructor¶

CommandHandler(**fields)

Methods¶

QueryHandler¶

class QueryHandler(BaseHandler, AppQueryHandler, Generic[TQuery])

Synchronous query handler for a specific query type.

Constructor¶

QueryHandler(**fields)

Methods¶

AsyncCommandHandler¶

class AsyncCommandHandler(AsyncBaseHandler, AppAsyncCommandHandler, Generic[TCommand])

Constructor¶

AsyncCommandHandler(**fields)

Methods¶

AsyncQueryHandler¶

class AsyncQueryHandler(AsyncBaseHandler, AppAsyncQueryHandler, Generic[TQuery])

Constructor¶

AsyncQueryHandler(**fields)

Methods¶

AdapterContainer¶

class AdapterContainer(BaseBehaviour)

Dependency injection container. Can be used directly or subclassed.

Constructor¶

AdapterContainer(**fields)

Methods¶

Testing Layer¶

from aod.testing import build, events_of, assert_event_emitted, assert_no_events, check_invariant
from aod.testing import FakeDomain
from aod.testing.doubles import (
    SpyLogger, SpyEventBus, SpyCache,
    port_stub, spy_session, spy_adapter_container,
    spy_command_handler, spy_query_handler,
    spy_async_command_handler, spy_async_query_handler,
)
from aod.testing.doubles.application.async_ import SpyLogger, SpyEventBus, SpyCache

build¶

build(cls: type[T], **kwargs: Any) -> T

Create domain object skipping validation.

events_of¶

events_of(obj: BaseGuarded) -> list[Event]

Extract events from a domain object.

assert_event_emitted¶

assert_event_emitted(events: Sequence[Event], event_type: type[Event], **attrs: Any) -> Event

Assert a specific event was emitted.

assert_no_events¶

assert_no_events(events: Sequence[Event]) -> None

Assert no events were emitted.

check_invariant¶

check_invariant(cls: type, invariant_name: str, data: dict[str, Any] | None = None, **kwargs: Any) -> None

Run a single invariant validator.

FakeDomain¶

FakeDomain(model_cls: type[T], **defaults: Any)

Factory for domain objects with auto-generated data.

Methods¶

SpyLogger¶

Methods: debug(msg, **context), info(msg, **context), warning(msg, **context), error(msg, **context).

SpyEventBus¶

Method: publish(*events).

SpyCache¶

Methods: get(key), set(key, value, ttl=None), delete(key).

spy_session¶

spy_session(session_cls: type[Session | AsyncSession]) -> type

Create a stub class from any Session or AsyncSession subclass. Every required and custom method becomes a MagicMock (or AsyncMock for async sessions), and is_dirty() returns False by default.

Async Spy Classes¶

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

Same names as sync variants. All methods that perform I/O are async.

spy_adapter_container¶

spy_adapter_container(container: AdapterContainer) -> AdapterContainer

Create a version of a container where sessions and ports are replaced with stubs.

spy_command_handler / spy_query_handler¶

spy_command_handler(*, returns: Any = None, raises: Exception | None = None) -> CommandPort
spy_query_handler(*, returns: Any = None, raises: Exception | None = None) -> QueryPort
spy_async_command_handler(*, returns: Any = None, raises: Exception | None = None) -> AsyncCommandPort
spy_async_query_handler(*, returns: Any = None, raises: Exception | None = None) -> AsyncQueryPort

Factory functions that create handler port spies. Each call returns a fresh instance with its own handle() mock — no shared state.

These implement CommandPort/QueryPort directly so they pass the UseCase port check without __skip_port_check__. Use in manual-DI tests:

spy = spy_command_handler(returns=user)
uc = CreateUserUseCase(save_user=spy)
uc.run(name="Alice")
assert spy.handle.called

session_stub¶

session_stub(session_cls: type[Session | AsyncSession]) -> type

Create a stub class from any Session or AsyncSession subclass. Every required method becomes a MagicMock (or AsyncMock for async sessions), and is_dirty() returns False by default.

port_stub¶

port_stub(port_cls: type[Port]) -> type

Create a stub class from any Port subclass. Every public method records calls and lets you configure return values.

Stub Control¶

Every stub method is a unittest.mock.MagicMock (or AsyncMock for async methods):

Exceptions¶

from aod.exceptions import (
    DomainException, MutationForbiddenException, InvarianceException,
    InvalidEntityTypeError, InvalidRootEntityTypeError, InvalidNestedTypeError,
    InvalidServiceParameterError, DuplicateDomainTypeError, ModelValidationError,
    ClassExpectedError, InvalidCommandFieldTypeError, InvalidQueryResultTypeError,
    InvalidGenericTypeArgError, InvalidServiceTypeError,
    ApplicationException, UnresolvableEntityError, CommitOutsideUnitOfWorkError,
    InvalidUseCasePortFieldError, InvalidHandlerPortFieldError,
    InfrastructureException, HandlerResultTypeError, HandlerModelError,
    PortNotFoundError, SessionNotFoundError,
    DuplicateHandlerError, HandlerNotFoundError,
)

DomainException Hierarchy¶

ApplicationException Hierarchy¶

InfrastructureException Hierarchy¶

Next Steps¶