domain.events.registry

src.domain.events.registry

Domain Events Registry - Single Source of Truth.

This registry catalogs ALL domain events in the system with their metadata. Used for: - Container wiring (automated subscription) - Validation tests (verify no drift) - Documentation generation (always accurate) - Gap detection (missing handlers, audit actions, etc.)

Architecture: - Domain layer (no dependencies on infrastructure) - Imported by container for automated wiring - Verified by tests to catch drift

Adding new events: 1. Define event dataclass in appropriate *_events.py file 2. Add entry to EVENT_REGISTRY below 3. Run tests - they'll tell you what's missing: - Handler methods needed - AuditAction enums needed - Container subscriptions needed (auto-wired)

Reference

• docs/architecture/domain-events-architecture.md
• F7.7 Phase 1.5: Event Registry implementation

Classes

EventCategory

Bases: Enum

Event categories for organization and filtering.

Source code in src/domain/events/registry.py

class EventCategory(Enum):
    """Event categories for organization and filtering."""

    AUTHENTICATION = "authentication"
    AUTHORIZATION = "authorization"
    PROVIDER = "provider"
    DATA_SYNC = "data_sync"
    SESSION = "session"
    RATE_LIMIT = "rate_limit"
    ADMIN = "admin"

WorkflowPhase

Bases: Enum

3-state workflow phases for ATTEMPT → OUTCOME pattern.

Source code in src/domain/events/registry.py

class WorkflowPhase(Enum):
    """3-state workflow phases for ATTEMPT → OUTCOME pattern."""

    ATTEMPTED = "attempted"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    ALLOWED = "allowed"  # For rate limit events (special case)
    DENIED = "denied"  # For rate limit events (special case)
    OPERATIONAL = "operational"  # For single-state operational events

EventMetadata dataclass

Metadata for a domain event.

Attributes:

Name	Type	Description
event_class	Type[DomainEvent]	The event dataclass.
category	EventCategory	Event category.
workflow_name	str	Name of workflow (e.g., "user_registration").
phase	WorkflowPhase	Workflow phase (attempted/succeeded/failed/operational).
requires_logging	bool	LoggingEventHandler handles this event.
requires_audit	bool	AuditEventHandler handles this event.
requires_email	bool	EmailEventHandler handles this event.
requires_session	bool	SessionEventHandler handles this event.
audit_action_name	str	Expected AuditAction enum name (for validation).

Source code in src/domain/events/registry.py

@dataclass(frozen=True)
class EventMetadata:
    """Metadata for a domain event.

    Attributes:
        event_class: The event dataclass.
        category: Event category.
        workflow_name: Name of workflow (e.g., "user_registration").
        phase: Workflow phase (attempted/succeeded/failed/operational).
        requires_logging: LoggingEventHandler handles this event.
        requires_audit: AuditEventHandler handles this event.
        requires_email: EmailEventHandler handles this event.
        requires_session: SessionEventHandler handles this event.
        audit_action_name: Expected AuditAction enum name (for validation).
    """

    event_class: Type[DomainEvent]
    category: EventCategory
    workflow_name: str
    phase: WorkflowPhase
    requires_logging: bool = True  # Default: all events logged
    requires_audit: bool = True  # Default: all events audited
    requires_email: bool = False  # Default: no email
    requires_session: bool = False  # Default: no session handling
    audit_action_name: str = ""  # Auto-computed if empty

Functions

get_all_events

get_all_events() -> list[Type[DomainEvent]]

Get all registered event classes.

Returns:

Type	Description
list[Type[DomainEvent]]	List of event classes in registry.

Source code in src/domain/events/registry.py

def get_all_events() -> list[Type[DomainEvent]]:
    """Get all registered event classes.

    Returns:
        List of event classes in registry.
    """
    return [meta.event_class for meta in EVENT_REGISTRY]

get_events_requiring_handler

get_events_requiring_handler(
    handler_type: str,
) -> list[Type[DomainEvent]]

Get events requiring specific handler.

Parameters:

Name	Type	Description	Default
handler_type	str	"logging", "audit", "email", or "session"	required

Returns:

Type	Description
list[Type[DomainEvent]]	List of event classes requiring that handler.

Raises:

Type	Description
ValueError	If handler_type is invalid.

Source code in src/domain/events/registry.py

def get_events_requiring_handler(handler_type: str) -> list[Type[DomainEvent]]:
    """Get events requiring specific handler.

    Args:
        handler_type: "logging", "audit", "email", or "session"

    Returns:
        List of event classes requiring that handler.

    Raises:
        ValueError: If handler_type is invalid.
    """
    field_map = {
        "logging": "requires_logging",
        "audit": "requires_audit",
        "email": "requires_email",
        "session": "requires_session",
    }

    if handler_type not in field_map:
        raise ValueError(
            f"Invalid handler_type: {handler_type}. "
            f"Must be one of: {list(field_map.keys())}"
        )

    field = field_map[handler_type]
    return [meta.event_class for meta in EVENT_REGISTRY if getattr(meta, field)]

get_workflow_events

get_workflow_events(
    workflow_name: str,
) -> dict[WorkflowPhase, Type[DomainEvent]]

Get all events for a workflow.

Parameters:

Name	Type	Description	Default
workflow_name	str	Workflow name (e.g., "user_registration")	required

Returns:

Type	Description
dict[WorkflowPhase, Type[DomainEvent]]	Dict mapping phase to event class.

Source code in src/domain/events/registry.py

def get_workflow_events(workflow_name: str) -> dict[WorkflowPhase, Type[DomainEvent]]:
    """Get all events for a workflow.

    Args:
        workflow_name: Workflow name (e.g., "user_registration")

    Returns:
        Dict mapping phase to event class.
    """
    return {
        meta.phase: meta.event_class
        for meta in EVENT_REGISTRY
        if meta.workflow_name == workflow_name
    }

get_expected_audit_actions

get_expected_audit_actions() -> (
    dict[Type[DomainEvent], str]
)

Get mapping of event to expected AuditAction enum name.

Returns:

Type	Description
dict[Type[DomainEvent], str]	Dict mapping event class to audit action name.

Source code in src/domain/events/registry.py

def get_expected_audit_actions() -> dict[Type[DomainEvent], str]:
    """Get mapping of event to expected AuditAction enum name.

    Returns:
        Dict mapping event class to audit action name.
    """
    return {
        meta.event_class: meta.audit_action_name
        for meta in EVENT_REGISTRY
        if meta.requires_audit
    }

get_statistics

get_statistics() -> dict[str, int | dict[str, int]]

Get registry statistics.

Returns:

Type	Description
dict[str, int | dict[str, int]]	Dict with counts by category, handler requirements, etc.

Source code in src/domain/events/registry.py

def get_statistics() -> dict[str, int | dict[str, int]]:
    """Get registry statistics.

    Returns:
        Dict with counts by category, handler requirements, etc.
    """
    from collections import Counter

    return {
        "total_events": len(EVENT_REGISTRY),
        "by_category": dict(Counter(meta.category.value for meta in EVENT_REGISTRY)),
        "by_phase": dict(Counter(meta.phase.value for meta in EVENT_REGISTRY)),
        "requiring_logging": sum(1 for m in EVENT_REGISTRY if m.requires_logging),
        "requiring_audit": sum(1 for m in EVENT_REGISTRY if m.requires_audit),
        "requiring_email": sum(1 for m in EVENT_REGISTRY if m.requires_email),
        "requiring_session": sum(1 for m in EVENT_REGISTRY if m.requires_session),
        "total_workflows": len({m.workflow_name for m in EVENT_REGISTRY}),
    }