Quick Reference

Invariants, patterns, and common recipes

Core Invariants

Invariant Enforced By
All entities have owner_id Domain model constructor
Events are immutable frozen=True on dataclass
No cross-component imports Boundary tests
Cross-component refs are UUIDs only Code review, type hints
Use Clock, not datetime.now() Linter rule, code review
Event subscribers are idempotent IdempotencyStore pattern

Dependency Rule

Layer dependencies 
Domain   ←   Application   ←   Adapters   ←   API
(inner)                                     (outer)

# Domain: NO external deps
# Application: Domain + Ports only
# Adapters: Implements Ports
# API: FastAPI routers (thin)

Inner layers don't know about outer layers. Dependencies point inward.

Interaction Pattern Quick Guide

Need Pattern
Block an operation Hook → return Veto
Modify input before processing Hook → return Patch
React after something happens Event subscription
Check permissions Policy.require()
Query across components QuerySpec by owner_id

Naming Conventions

Thing Convention Example
Component lowercase singular todos, budget
Module class PascalCase + Module TodosModule
Entity PascalCase singular Task, BudgetPlan
Event PascalCase past tense TaskCompleted, NoteCreated
Hook point component.action todos.before_complete_task
Facet name component.purpose todos.preferences
Use case VerbNoun CompleteTask, CreateNote
Repository EntityRepository TaskRepository
Error DescriptiveError TaskNotFoundError

Common Recipes

Publish an event after state change

use_case.py 
entity.do_something()
repo.save(entity)              # commit FIRST
event_bus.publish(SomethingDone(...))

Check authorization

use_case.py 
entity = repo.get(entity_id)
policy.require(actor_id, "action", entity)  # raises if denied
# proceed with action

Register a hook

module.py 
def register_hooks(self, dispatcher):
    dispatcher.register("component.hook_point", self.handler)

def handler(self, context: dict) -> HookResponse:
    if should_block:
        return Veto("reason")
    return Allow()

Subscribe to an event

module.py 
def register_subscribers(self, event_bus):
    event_bus.subscribe(SomeEvent, self.on_some_event)

def on_some_event(self, envelope: EventEnvelope):
    if idempotency.exists(f"key:{envelope.id}"):
        return
    # process event
    idempotency.mark(f"key:{envelope.id}")

Query by owner

use_case.py 
spec = QuerySpec(filters=[
    Filter("owner_id", FilterOp.EQ, owner_id)
])
entities = repo.query(spec)

Test Doubles

Production Test Double
SystemClock FixedClock
EventBus InMemoryEventBus
IdempotencyStore InMemoryIdempotencyStore
OwnerOnlyPolicy AllowAllPolicy
PostgresRepo InMemoryRepo