spakky-event

spakky-event는 이벤트 발행, 소비, dispatch, handler stereotype을 제공하는 core 패키지입니다.

인프로세스 이벤트 시스템 — EventBus, EventTransport, EventMediator

플러그인 진입점

initialize(app)

Initialize the spakky-event plugin.

Registers event mediators, publishers, buses, and post-processors.

Source code in core/spakky-event/src/spakky/event/main.py

def initialize(app: SpakkyApplication) -> None:
    """Initialize the spakky-event plugin.

    Registers event mediators, publishers, buses, and post-processors.
    """
    app.add(AsyncTransactionalEventPublishingAspect)
    app.add(TransactionalEventPublishingAspect)

    app.add(EventMediator)
    app.add(AsyncEventMediator)
    app.add(EventPublisher)
    app.add(AsyncEventPublisher)
    app.add(AuthContextSnapshotHeaderInjector)
    app.add(DirectEventBus)
    app.add(AsyncDirectEventBus)

    app.add(EventHandlerRegistrationPostProcessor)

Publisher 인터페이스

Event publishing and transport interfaces.

Provides publisher, bus, and transport abstractions for event routing: - IEventPublisher: Routes events by type (domain vs integration). - IEventBus: Serializes and sends integration events via transport. - IEventTransport: Low-level transport for serialized event payloads.

IEventPublisher

Bases: ABC

Publishes events by routing to dispatcher or bus based on event type.

publish(event) abstractmethod

Publish an event (domain → dispatcher, integration → bus).

Source code in core/spakky-event/src/spakky/event/event_publisher.py

@abstractmethod
def publish(self, event: AbstractEvent) -> None:
    """Publish an event (domain → dispatcher, integration → bus)."""
    ...

IAsyncEventPublisher

Bases: ABC

Async counterpart of IEventPublisher.

publish(event) abstractmethod async

Publish an event asynchronously.

Source code in core/spakky-event/src/spakky/event/event_publisher.py

@abstractmethod
async def publish(self, event: AbstractEvent) -> None:
    """Publish an event asynchronously."""
    ...

IEventBus

Bases: ABC

Synchronous event bus for sending integration events.

send(event) abstractmethod

Serialize and send an integration event via transport.

Source code in core/spakky-event/src/spakky/event/event_publisher.py

@abstractmethod
def send(self, event: AbstractIntegrationEvent) -> None:
    """Serialize and send an integration event via transport."""
    ...

IAsyncEventBus

Bases: ABC

Asynchronous event bus for sending integration events.

send(event) abstractmethod async

Serialize and send an integration event via transport.

Source code in core/spakky-event/src/spakky/event/event_publisher.py

@abstractmethod
async def send(self, event: AbstractIntegrationEvent) -> None:
    """Serialize and send an integration event via transport."""
    ...

IEventTransport

Bases: ABC

Low-level synchronous transport for pre-serialized event payloads.

The caller owns the publish batch boundary: hand one or more payloads to send() and call flush() once at the end of the batch. Transports may buffer payloads until flush() so that broker round trips are batched.

send(event_name, payload, headers, partition_key=None) abstractmethod

Hand a serialized event payload to the broker client for delivery.

A permanent rejection attributable to this record must be reported as EventDeliveryRejectedError. A connection, timeout, queue, or other transport-wide failure keeps the client exception's original type so a caller does not spend this record's retry budget.

Parameters:

Name	Type	Description	Default
event_name	str	Destination topic / routing key.	required
payload	bytes	Pre-serialized event bytes.	required
headers	dict[str, str]	Metadata headers for trace and auth propagation.	required
partition_key	str | None	Key pinning the payload to one broker partition. None spreads payloads round-robin.	None

Raises:

Type	Description
EventDeliveryRejectedError	When the transport permanently rejects this specific record.
EventTransportNotRunningError	When the transport's broker client is not open, which happens outside the application lifecycle.

Source code in core/spakky-event/src/spakky/event/event_publisher.py

@abstractmethod
def send(
    self,
    event_name: str,
    payload: bytes,
    headers: dict[str, str],
    partition_key: str | None = None,
) -> None:
    """Hand a serialized event payload to the broker client for delivery.

    A permanent rejection attributable to this record must be reported as
    `EventDeliveryRejectedError`. A connection, timeout, queue, or other
    transport-wide failure keeps the client exception's original type so a
    caller does not spend this record's retry budget.

    Args:
        event_name: Destination topic / routing key.
        payload: Pre-serialized event bytes.
        headers: Metadata headers for trace and auth propagation.
        partition_key: Key pinning the payload to one broker partition.
            None spreads payloads round-robin.

    Raises:
        EventDeliveryRejectedError: When the transport permanently rejects
            this specific record.
        EventTransportNotRunningError: When the transport's broker client is
            not open, which happens outside the application lifecycle.
    """
    ...

flush() abstractmethod

Block until the broker client has finished sending what send() handed over.

A successful return means every record this caller handed over reached the broker. An implementation that learns of a rejection only through a delivery callback collects it and raises here — a caller that batches records has no other moment to learn that the batch did not arrive, and would otherwise record undelivered events as published.

Implementations must preserve the same failure taxonomy as send(): only a permanent record-specific rejection becomes EventDeliveryRejectedError; transport-wide failures retain the broker client's original exception type.

Raises:

Type	Description
EventDeliveryRejectedError	When one or more records are permanently rejected for a record-specific cause.

Source code in core/spakky-event/src/spakky/event/event_publisher.py

@abstractmethod
def flush(self) -> None:
    """Block until the broker client has finished sending what send() handed over.

    A successful return means every record this caller handed over reached
    the broker. An implementation that learns of a rejection only through a
    delivery callback collects it and raises here — a caller that batches
    records has no other moment to learn that the batch did not arrive, and
    would otherwise record undelivered events as published.

    Implementations must preserve the same failure taxonomy as send(): only
    a permanent record-specific rejection becomes
    `EventDeliveryRejectedError`; transport-wide failures retain the broker
    client's original exception type.

    Raises:
        EventDeliveryRejectedError: When one or more records are permanently
            rejected for a record-specific cause.
    """
    ...

IAsyncEventTransport

Bases: ABC

Low-level asynchronous transport for pre-serialized event payloads.

The caller owns the publish batch boundary: hand one or more payloads to send() and call flush() once at the end of the batch. Transports may buffer payloads until flush() so that broker round trips are batched. Publishers share one transport, so a batch's send() and flush() belong to the same execution context and flush() reports only that publisher's payloads.

send(event_name, payload, headers, partition_key=None) abstractmethod async

Hand a serialized event payload to the broker client for delivery.

A permanent rejection attributable to this record must be reported as EventDeliveryRejectedError. A connection, timeout, queue, or other transport-wide failure keeps the client exception's original type so a caller does not spend this record's retry budget.

Parameters:

Name	Type	Description	Default
event_name	str	Destination topic / routing key.	required
payload	bytes	Pre-serialized event bytes.	required
headers	dict[str, str]	Metadata headers for trace and auth propagation.	required
partition_key	str | None	Key pinning the payload to one broker partition. None spreads payloads round-robin.	None

Raises:

Type	Description
EventDeliveryRejectedError	When the transport permanently rejects this specific record.
EventTransportNotRunningError	When the transport's broker client is not open, which happens outside the application lifecycle.

Source code in core/spakky-event/src/spakky/event/event_publisher.py

@abstractmethod
async def send(
    self,
    event_name: str,
    payload: bytes,
    headers: dict[str, str],
    partition_key: str | None = None,
) -> None:
    """Hand a serialized event payload to the broker client for delivery.

    A permanent rejection attributable to this record must be reported as
    `EventDeliveryRejectedError`. A connection, timeout, queue, or other
    transport-wide failure keeps the client exception's original type so a
    caller does not spend this record's retry budget.

    Args:
        event_name: Destination topic / routing key.
        payload: Pre-serialized event bytes.
        headers: Metadata headers for trace and auth propagation.
        partition_key: Key pinning the payload to one broker partition.
            None spreads payloads round-robin.

    Raises:
        EventDeliveryRejectedError: When the transport permanently rejects
            this specific record.
        EventTransportNotRunningError: When the transport's broker client is
            not open, which happens outside the application lifecycle.
    """
    ...

flush() abstractmethod async

Block until the broker client has finished sending what send() handed over.

A successful return means every record this caller handed over reached the broker. An implementation that learns of a rejection only through a delivery callback collects it and raises here — a caller that batches records has no other moment to learn that the batch did not arrive, and would otherwise record undelivered events as published.

Implementations must preserve the same failure taxonomy as send(): only a permanent record-specific rejection becomes EventDeliveryRejectedError; transport-wide failures retain the broker client's original exception type.

Raises:

Type	Description
EventDeliveryRejectedError	When one or more records are permanently rejected for a record-specific cause.

Source code in core/spakky-event/src/spakky/event/event_publisher.py

@abstractmethod
async def flush(self) -> None:
    """Block until the broker client has finished sending what send() handed over.

    A successful return means every record this caller handed over reached
    the broker. An implementation that learns of a rejection only through a
    delivery callback collects it and raises here — a caller that batches
    records has no other moment to learn that the batch did not arrive, and
    would otherwise record undelivered events as published.

    Implementations must preserve the same failure taxonomy as send(): only
    a permanent record-specific rejection becomes
    `EventDeliveryRejectedError`; transport-wide failures retain the broker
    client's original exception type.

    Raises:
        EventDeliveryRejectedError: When one or more records are permanently
            rejected for a record-specific cause.
    """
    ...

Consumer 인터페이스

Event consumer interfaces for registering event handlers.

EventHandlerCallback = Callable[[EventT_contra], None]

Synchronous event handler callback type.

AsyncEventHandlerCallback = Callable[[EventT_contra], Awaitable[None]]

Asynchronous event handler callback type.

IEventConsumer

Bases: ABC

Synchronous event consumer interface for registering event handlers.

register(event, handler) abstractmethod

Register a handler callback for the given event type.

Source code in core/spakky-event/src/spakky/event/event_consumer.py

@abstractmethod
def register[EventT_contra: AbstractEvent](
    self,
    event: type[EventT_contra],
    handler: EventHandlerCallback[EventT_contra],
) -> None:
    """Register a handler callback for the given event type."""
    ...

IAsyncEventConsumer

Bases: ABC

Asynchronous event consumer interface for registering event handlers.

register(event, handler) abstractmethod

Register an async handler callback for the given event type.

Source code in core/spakky-event/src/spakky/event/event_consumer.py

@abstractmethod
def register[EventT_contra: AbstractEvent](
    self,
    event: type[EventT_contra],
    handler: AsyncEventHandlerCallback[EventT_contra],
) -> None:
    """Register an async handler callback for the given event type."""
    ...

Dispatcher 인터페이스

Event dispatcher interfaces for dispatching events to registered handlers.

This module provides unified dispatcher interfaces that handle all event types. Dispatchers are responsible for delivering events to registered handlers, while Consumers are responsible for handler registration. These interfaces are combined in Mediator implementations.

IEventDispatcher

Bases: ABC

Synchronous event dispatcher interface.

dispatch(event) abstractmethod

Dispatch an event to all registered handlers.

Source code in core/spakky-event/src/spakky/event/event_dispatcher.py

@abstractmethod
def dispatch(self, event: AbstractEvent) -> None:
    """Dispatch an event to all registered handlers."""
    ...

IAsyncEventDispatcher

Bases: ABC

Asynchronous event dispatcher interface.

dispatch(event) abstractmethod async

Dispatch an event to all registered async handlers.

Source code in core/spakky-event/src/spakky/event/event_dispatcher.py

@abstractmethod
async def dispatch(self, event: AbstractEvent) -> None:
    """Dispatch an event to all registered async handlers."""
    ...

EventBus

Default EventBus implementations that delegate to EventTransport.

DirectEventBus(transport, propagator, auth_snapshot_headers=None)

Bases: IEventBus

Synchronous event bus that serializes and delegates to IEventTransport.

Initialize with the given transport and trace propagator.

Source code in core/spakky-event/src/spakky/event/bus/transport_event_bus.py

def __init__(
    self,
    transport: IEventTransport,
    propagator: ITracePropagator,
    auth_snapshot_headers: AuthContextSnapshotHeaderInjector | None = None,
) -> None:
    """Initialize with the given transport and trace propagator."""
    self._transport = transport
    self._propagator = propagator
    self._auth_snapshot_headers = (
        auth_snapshot_headers or AuthContextSnapshotHeaderInjector()
    )
    self._adapters = {}

send(event)

Serialize and send an integration event via transport.

A direct publish is a batch of one, so the transport is flushed before returning and the caller gets the same delivery guarantee as before.

Source code in core/spakky-event/src/spakky/event/bus/transport_event_bus.py

@override
def send(self, event: AbstractIntegrationEvent) -> None:
    """Serialize and send an integration event via transport.

    A direct publish is a batch of one, so the transport is flushed before
    returning and the caller gets the same delivery guarantee as before.
    """
    event_type = type(event)
    if event_type not in self._adapters:
        self._adapters[event_type] = TypeAdapter(event_type)
    adapter = self._adapters[event_type]
    headers: dict[str, str] = {}
    self._propagator.inject(headers)
    self._auth_snapshot_headers.inject(headers)
    self._transport.send(
        event.event_name,
        adapter.dump_json(event),
        headers,
        event.partition_key,
    )
    self._transport.flush()

AsyncDirectEventBus(transport, propagator, auth_snapshot_headers=None)

Bases: IAsyncEventBus

Asynchronous event bus that serializes and delegates to IAsyncEventTransport.

Initialize with the given async transport and trace propagator.

Source code in core/spakky-event/src/spakky/event/bus/transport_event_bus.py

def __init__(
    self,
    transport: IAsyncEventTransport,
    propagator: ITracePropagator,
    auth_snapshot_headers: AuthContextSnapshotHeaderInjector | None = None,
) -> None:
    """Initialize with the given async transport and trace propagator."""
    self._transport = transport
    self._propagator = propagator
    self._auth_snapshot_headers = (
        auth_snapshot_headers or AuthContextSnapshotHeaderInjector()
    )
    self._adapters = {}

send(event) async

Serialize and send an integration event via async transport.

A direct publish is a batch of one, so the transport is flushed before returning and the caller gets the same delivery guarantee as before.

Source code in core/spakky-event/src/spakky/event/bus/transport_event_bus.py

@override
async def send(self, event: AbstractIntegrationEvent) -> None:
    """Serialize and send an integration event via async transport.

    A direct publish is a batch of one, so the transport is flushed before
    returning and the caller gets the same delivery guarantee as before.
    """
    event_type = type(event)
    if event_type not in self._adapters:
        self._adapters[event_type] = TypeAdapter(event_type)
    adapter = self._adapters[event_type]
    headers: dict[str, str] = {}
    self._propagator.inject(headers)
    self._auth_snapshot_headers.inject(headers)
    await self._transport.send(
        event.event_name,
        adapter.dump_json(event),
        headers,
        event.partition_key,
    )
    await self._transport.flush()

Auth Propagation

Auth snapshot metadata propagation for outbound integration events.

AuthContextSnapshotHeaderInjector(auth_snapshot_signer=None, auth_snapshot_propagation_configs=())

Bases: IApplicationContextAware

Inject signed AuthContextSnapshot metadata into outbound event headers.

Source code in core/spakky-event/src/spakky/event/auth_propagation.py

def __init__(
    self,
    auth_snapshot_signer: IAuthContextSnapshotSigner | None = None,
    auth_snapshot_propagation_configs: tuple[
        AuthSnapshotPropagationConfig, ...
    ] = (),
) -> None:
    self._auth_snapshot_signer = auth_snapshot_signer
    self._auth_snapshot_propagation_config = (
        effective_auth_snapshot_propagation_config(
            auth_snapshot_propagation_configs
        )
    )
    self._application_context = None

set_application_context(application_context)

Inject the application context used to read request-scoped auth state.

Source code in core/spakky-event/src/spakky/event/auth_propagation.py

@override
def set_application_context(self, application_context: IApplicationContext) -> None:
    """Inject the application context used to read request-scoped auth state."""
    self._application_context = application_context

inject(headers)

Add signed snapshot metadata when outbound propagation is enabled.

Source code in core/spakky-event/src/spakky/event/auth_propagation.py

def inject(self, headers: dict[str, str]) -> None:
    """Add signed snapshot metadata when outbound propagation is enabled."""
    self._remove_raw_bearer_headers(headers)
    if not self._auth_snapshot_propagation_config.enabled:
        return
    auth_context = self._current_auth_context()
    if auth_context is None:
        return
    snapshot = self._required_signer().sign_snapshot(
        SnapshotSignRequest(auth_context=auth_context)
    )
    headers[AUTH_CONTEXT_SNAPSHOT_METADATA_KEY] = (
        snapshot.base64url_canonical_json()
    )

Mediator

Event mediator implementations.

This module provides in-process mediator implementations that combine Consumer and Dispatcher interfaces. Mediators manage handler registration and event dispatching within the same bounded context.

EventMediator()

Bases: IEventConsumer, IEventDispatcher

In-process synchronous event mediator combining consumer and dispatcher roles.

Initialize an empty handler registry.

Source code in core/spakky-event/src/spakky/event/mediator/domain_event_mediator.py

def __init__(self) -> None:
    """Initialize an empty handler registry."""
    self._handlers = {}

register(event, handler)

Register a handler callback for the given event type.

Source code in core/spakky-event/src/spakky/event/mediator/domain_event_mediator.py

@override
def register(
    self,
    event: type[AbstractEvent],
    handler: EventHandlerCallback[
        Any
    ],  # Any: callback generic is erased by event-keyed registry
) -> None:
    """Register a handler callback for the given event type."""
    if event not in self._handlers:
        self._handlers[event] = []
    self._handlers[event].append(handler)
    logger.debug(f"Registered handler for {event.__name__}")

dispatch(event)

Dispatch an event to all registered handlers.

Source code in core/spakky-event/src/spakky/event/mediator/domain_event_mediator.py

@override
def dispatch(self, event: AbstractEvent) -> None:
    """Dispatch an event to all registered handlers."""
    event_type = type(event)
    handlers = self._handlers.get(event_type, [])

    if not handlers:
        logger.debug(f"No handlers registered for {event_type.__name__}")
        return

    for handler in handlers:
        try:
            handler(event)
        except Exception as e:
            logger.error(
                f"Handler {handler} failed for {event_type.__name__}: {e}",
                exc_info=True,
            )

AsyncEventMediator()

Bases: IAsyncEventConsumer, IAsyncEventDispatcher

In-process asynchronous event mediator combining consumer and dispatcher roles.

Initialize an empty async handler registry.

Source code in core/spakky-event/src/spakky/event/mediator/domain_event_mediator.py

def __init__(self) -> None:
    """Initialize an empty async handler registry."""
    self._handlers = {}

register(event, handler)

Register an async handler callback for the given event type.

Source code in core/spakky-event/src/spakky/event/mediator/domain_event_mediator.py

@override
def register(
    self,
    event: type[AbstractEvent],
    handler: AsyncEventHandlerCallback[
        Any
    ],  # Any: callback generic is erased by event-keyed registry
) -> None:
    """Register an async handler callback for the given event type."""
    if event not in self._handlers:
        self._handlers[event] = []
    self._handlers[event].append(handler)
    logger.debug(f"Registered async handler for {event.__name__}")

dispatch(event) async

Dispatch an event to all registered async handlers.

Source code in core/spakky-event/src/spakky/event/mediator/domain_event_mediator.py

@override
async def dispatch(self, event: AbstractEvent) -> None:
    """Dispatch an event to all registered async handlers."""
    event_type = type(event)
    handlers = self._handlers.get(event_type, [])

    if not handlers:
        logger.debug(f"No handlers registered for {event_type.__name__}")
        return

    for handler in handlers:
        try:
            await handler(event)
        except Exception as e:
            logger.error(
                f"Handler {handler} failed for {event_type.__name__}: {e}",
                exc_info=True,
            )

Publisher

Event publisher implementations.

This module provides event publishers that route events by type: - AbstractDomainEvent → EventMediator (in-process dispatch) - AbstractIntegrationEvent → IEventBus (external transport)

EventPublisher(dispatcher, bus)

Bases: IEventPublisher

Routes events by type: domain events to dispatcher, integration events to bus.

Initialize with dispatcher and bus dependencies.

Source code in core/spakky-event/src/spakky/event/publisher/domain_event_publisher.py

def __init__(
    self,
    dispatcher: IEventDispatcher,
    bus: IEventBus,
) -> None:
    """Initialize with dispatcher and bus dependencies."""
    self._dispatcher = dispatcher
    self._bus = bus

publish(event)

Route an event to the appropriate handler based on its type.

Source code in core/spakky-event/src/spakky/event/publisher/domain_event_publisher.py

@override
def publish(self, event: AbstractEvent) -> None:
    """Route an event to the appropriate handler based on its type."""
    match event:
        case AbstractDomainEvent():
            self._dispatcher.dispatch(event)
        case AbstractIntegrationEvent():
            self._bus.send(event)
        case _:  # pragma: no cover - 방어적 분기 (정상 흐름 불가)
            raise UnknownEventTypeError(type(event))

AsyncEventPublisher(dispatcher, bus)

Bases: IAsyncEventPublisher

Async counterpart that routes events by type.

Initialize with async dispatcher and bus dependencies.

Source code in core/spakky-event/src/spakky/event/publisher/domain_event_publisher.py

def __init__(
    self,
    dispatcher: IAsyncEventDispatcher,
    bus: IAsyncEventBus,
) -> None:
    """Initialize with async dispatcher and bus dependencies."""
    self._dispatcher = dispatcher
    self._bus = bus

publish(event) async

Route an event to the appropriate async handler based on its type.

Source code in core/spakky-event/src/spakky/event/publisher/domain_event_publisher.py

@override
async def publish(self, event: AbstractEvent) -> None:
    """Route an event to the appropriate async handler based on its type."""
    match event:
        case AbstractDomainEvent():
            await self._dispatcher.dispatch(event)
        case AbstractIntegrationEvent():
            await self._bus.send(event)
        case _:  # pragma: no cover - 방어적 분기 (정상 흐름 불가)
            raise UnknownEventTypeError(type(event))

Aspect

Transactional event publishing aspect for automatic domain event publishing.

AsyncTransactionalEventPublishingAspect(collector, publisher)

Bases: IAsyncAspect

Source code in core/spakky-event/src/spakky/event/aspects/transactional_event_publishing.py

def __init__(
    self,
    collector: AggregateCollector,
    publisher: IAsyncEventPublisher,
) -> None:
    self._collector = collector
    self._publisher = publisher

after_returning_async(result) async

Publish domain events from collected aggregates after successful commit.

Source code in core/spakky-event/src/spakky/event/aspects/transactional_event_publishing.py

@AfterReturning(lambda x: Transactional.exists(x) and iscoroutinefunction(x))
@override
async def after_returning_async(self, result: Any) -> None:
    """Publish domain events from collected aggregates after successful commit."""
    for aggregate in self._collector.all():
        for event in aggregate.events:
            await self._publisher.publish(event)
        aggregate.clear_events()

after_async() async

Clear the aggregate collector after transaction completion.

Source code in core/spakky-event/src/spakky/event/aspects/transactional_event_publishing.py

@After(lambda x: Transactional.exists(x) and iscoroutinefunction(x))
@override
async def after_async(self) -> None:
    """Clear the aggregate collector after transaction completion."""
    self._collector.clear()

TransactionalEventPublishingAspect(collector, publisher)

Bases: IAspect

Source code in core/spakky-event/src/spakky/event/aspects/transactional_event_publishing.py

def __init__(
    self,
    collector: AggregateCollector,
    publisher: IEventPublisher,
) -> None:
    self._collector = collector
    self._publisher = publisher

after_returning(result)

Publish domain events from collected aggregates after successful commit.

Source code in core/spakky-event/src/spakky/event/aspects/transactional_event_publishing.py

@AfterReturning(lambda x: Transactional.exists(x) and not iscoroutinefunction(x))
@override
def after_returning(self, result: Any) -> None:
    """Publish domain events from collected aggregates after successful commit."""
    for aggregate in self._collector.all():
        for event in aggregate.events:
            self._publisher.publish(event)
        aggregate.clear_events()

after()

Clear the aggregate collector after transaction completion.

Source code in core/spakky-event/src/spakky/event/aspects/transactional_event_publishing.py

@After(lambda x: Transactional.exists(x) and not iscoroutinefunction(x))
@override
def after(self) -> None:
    """Clear the aggregate collector after transaction completion."""
    self._collector.clear()

스테레오타입

EventHandler stereotype and event routing decorators.

This module provides @EventHandler stereotype and @on_event decorator for organizing event-driven architectures.

EventHandlerMethod = Callable[[Any, EventT_contra], None | Awaitable[None]]

Type alias for event handler callback functions.

EventRoute(event_type) dataclass

Bases: FunctionAnnotation

Annotation for marking methods as event handlers.

Associates a method with a specific domain event type.

event_type instance-attribute

The domain event type this handler processes.

__call__(obj)

Apply event route annotation to method.

Parameters:

Name	Type	Description	Default
obj	EventHandlerMethod[EventT_contra]	The method to annotate.	required

Returns:

Type	Description
EventHandlerMethod[EventT_contra]	The annotated method.

Source code in core/spakky-event/src/spakky/event/stereotype/event_handler.py

def __call__(
    self, obj: EventHandlerMethod[EventT_contra]
) -> EventHandlerMethod[EventT_contra]:
    """Apply event route annotation to method.

    Args:
        obj: The method to annotate.

    Returns:
        The annotated method.
    """
    return super().__call__(obj)

EventHandler(*, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

Stereotype for event handler classes.

EventHandlers contain methods decorated with @on_event that process domain events asynchronously.

on_event(event_type)

Decorator for marking methods as event handlers.

Parameters:

Name	Type	Description	Default
event_type	type[EventT_contra]	The domain event type to handle.	required

Returns:

Type	Description
Callable[[EventHandlerMethod[EventT_contra]], EventHandlerMethod[EventT_contra]]	Decorator function that applies EventRoute annotation.

Example

@EventHandler() class UserEventHandler: @on_event(UserCreatedEvent) async def handle_user_created(self, event: UserCreatedEvent) -> None: # Handle event pass

Source code in core/spakky-event/src/spakky/event/stereotype/event_handler.py

def on_event[EventT_contra: AbstractEvent](
    event_type: type[EventT_contra],
) -> Callable[
    [EventHandlerMethod[EventT_contra]],
    EventHandlerMethod[EventT_contra],
]:
    """Decorator for marking methods as event handlers.

    Args:
        event_type: The domain event type to handle.

    Returns:
        Decorator function that applies EventRoute annotation.

    Example:
        @EventHandler()
        class UserEventHandler:
            @on_event(UserCreatedEvent)
            async def handle_user_created(self, event: UserCreatedEvent) -> None:
                # Handle event
                pass
    """

    def wrapper(
        method: EventHandlerMethod[EventT_contra],
    ) -> EventHandlerMethod[EventT_contra]:
        return EventRoute(event_type)(method)

    return wrapper

후처리기

Event handler registration post-processor.

EventHandlerRegistrationPostProcessor

Bases: IPostProcessor, IContainerAware

Scans @EventHandler Pods and registers their @on_event methods with consumers.

set_container(container)

Receive the container reference via IContainerAware.

Source code in core/spakky-event/src/spakky/event/post_processor.py

@override
def set_container(self, container: IContainer) -> None:
    """Receive the container reference via IContainerAware."""
    self.__container = container

post_process(pod)

Register event handler methods with the appropriate consumer.

Source code in core/spakky-event/src/spakky/event/post_processor.py

@override
def post_process(self, pod: object) -> object:
    """Register event handler methods with the appropriate consumer."""
    pod_type = type(pod)

    if not EventHandler.exists(pod_type):
        return pod

    sync_consumer = self.__container.get(IEventConsumer)
    async_consumer = self.__container.get(IAsyncEventConsumer)

    for name, method in getmembers(pod, predicate=ismethod):
        route = EventRoute[AbstractEvent].get_or_none(method)
        if route is None:
            continue
        if not issubclass(route.event_type, AbstractDomainEvent):
            continue

        event_type = route.event_type

        if iscoroutinefunction(method):
            async_consumer.register(event_type, method)
            logger.debug(
                f"Registered async handler {pod_type.__name__}.{name} "
                f"for {event_type.__name__}"
            )
        else:
            sync_consumer.register(event_type, method)
            logger.debug(
                f"Registered sync handler {pod_type.__name__}.{name} "
                f"for {event_type.__name__}"
            )

    return pod

에러

AbstractSpakkyEventError

Bases: AbstractSpakkyFrameworkError, ABC

Base error for event system operations.

InvalidMessageError

Bases: AbstractSpakkyEventError

Raised when a message received is invalid or malformed.

AuthSnapshotPropagationSignerUnavailableError

Bases: AbstractSpakkyEventError

Raised when signed snapshot propagation lacks a signer provider.

AuthSnapshotPropagationContextUnavailableError

Bases: AbstractSpakkyEventError

Raised when signed snapshot propagation cannot read ApplicationContext.

EventTransportNotRunningError

Bases: AbstractSpakkyEventError

Raised when a transport is used outside the application lifecycle.

Transports whose broker client is opened at application start and closed at application stop raise this instead of publishing into a closed client, so that callers can tell a shutdown window apart from a delivery failure.

EventDeliveryRejectedError(reasons)

Bases: AbstractSpakkyEventError

Raised when a transport attributes a permanent rejection to a record.

A client may report the rejection synchronously from send() or later through a delivery callback consumed by flush(). Both paths use this error so callers can distinguish record-specific failure from a transport-wide outage.

Source code in core/spakky-event/src/spakky/event/error.py

def __init__(self, reasons: list[str]) -> None:
    self.reasons = reasons
    super().__init__()

UnknownEventTypeError(event_type)

Bases: AbstractSpakkyEventError

Raised when an event type is neither domain nor integration.

Source code in core/spakky-event/src/spakky/event/error.py

def __init__(self, event_type: type) -> None:
    self.event_type = event_type
    super().__init__()