Spakky Event

Spakky Framework를 위한 이벤트 처리 stereotype입니다.

설치

pip install spakky-event

주요 기능

• @EventHandler: 이벤트 handler 클래스용 stereotype
• @on_event: 메서드를 이벤트 handler로 표시하는 decorator
• 타입 안전성: 이벤트 타입에 대한 타입 힌트 지원
• 비동기 지원: 이벤트 처리에 async/await 네이티브 지원

빠른 시작

이벤트 정의

이벤트는 spakky-domain의 이벤트 기반 클래스 중 하나를 상속해야 합니다.

• bounded context 내부 이벤트(in-process domain event)에는 AbstractDomainEvent를 사용합니다.
• 경계를 넘는 이벤트(message broker integration)에는 AbstractIntegrationEvent를 사용합니다.

메시지 브로커 플러그인(RabbitMQ/Kafka)을 사용할 때는 이벤트를 AbstractIntegrationEvent로 정의하세요.

from spakky.core.common.mutability import immutable
from spakky.domain.models.event import AbstractIntegrationEvent


@immutable
class UserCreatedEvent(AbstractIntegrationEvent):
    user_id: str
    email: str


@immutable
class UserDeletedEvent(AbstractIntegrationEvent):
    user_id: str

EventHandler 생성

@EventHandler stereotype과 @on_event decorator를 함께 사용합니다.

from spakky.event.stereotype.event_handler import EventHandler, on_event


@EventHandler()
class UserEventHandler:
    def __init__(self, notification_service: NotificationService) -> None:
        self.notification_service = notification_service

    @on_event(UserCreatedEvent)
    async def on_user_created(self, event: UserCreatedEvent) -> None:
        await self.notification_service.send_welcome_email(event.email)

    @on_event(UserDeletedEvent)
    async def on_user_deleted(self, event: UserDeletedEvent) -> None:
        await self.notification_service.send_goodbye_email(event.user_id)

메시지 브로커 통합

@EventHandler stereotype은 Spakky의 message broker 플러그인과 함께 동작합니다.

RabbitMQ

pip install spakky-rabbitmq

from spakky.core.application.application import SpakkyApplication
from spakky.core.application.application_context import ApplicationContext

app = (
    SpakkyApplication(ApplicationContext())
    .load_plugins()  # Loads spakky-rabbitmq plugin
    .scan()
    .start()
)

Kafka

pip install spakky-kafka

from spakky.core.application.application import SpakkyApplication
from spakky.core.application.application_context import ApplicationContext

app = (
    SpakkyApplication(ApplicationContext())
    .load_plugins()  # Loads spakky-kafka plugin
    .scan()
    .start()
)

API 레퍼런스

스테레오타입

인터페이스

구현체

전파 metadata

DirectEventBus와 AsyncDirectEventBus는 IEventTransport.send(event_name, payload, headers) public signature를 유지하면서 outbound header를 구성합니다.

Authorization: Bearer ... 같은 raw bearer credential은 event header로 전파하지 않습니다.

타입

에러

관련 패키지

In-process 도메인 이벤트 발행

bounded context 내부 이벤트(DomainEvents)에는 in-process publisher를 사용합니다.

from spakky.core.application.application import SpakkyApplication
from spakky.core.application.application_context import ApplicationContext
from spakky.event import IAsyncEventPublisher

# 애플리케이션 부트스트랩(load_plugins가 event component를 자동 등록)
app = (
    SpakkyApplication(ApplicationContext())
    .load_plugins()
    .scan()
    .start()
)

# 컨테이너에서 publisher 조회
publisher = app.container.get(IAsyncEventPublisher)
await publisher.publish(UserCreatedEvent(user_id="123", email="test@example.com"))

아키텍처(ISP 준수)

in-process event system은 Interface Segregation Principle을 따릅니다.

• Consumer: event handler 등록(register() 메서드)
• Dispatcher: event를 handler로 dispatch(dispatch() 메서드)
• Mediator: 두 인터페이스를 단일 구현체로 결합
• Publisher: Consumer가 아니라 Dispatcher에만 의존

flowchart TD
    publisher[Publisher]
    dispatcher[Dispatcher]
    mediator["Mediator<br/>Consumer + Dispatcher"]
    handler["EventHandler<br/>@on_event"]

    publisher --> dispatcher
    dispatcher --> mediator
    mediator --> handler

개발 검증

패키지 단위 검증은 해당 패키지 디렉토리에서 실행합니다.

uv run ruff format .
uv run ruff check .
uv run pyrefly check
uv run pytest

pytest는 각 패키지 pyproject.toml의 coverage 설정을 사용합니다.

라이선스

MIT License