spakky-outbox

Spakky Framework를 위한 트랜잭셔널 Outbox 패턴 패키지입니다. IEventBus를 @Primary outbox bus로 교체해 이벤트 저장과 외부 전송을 분리합니다.

설치

pip install spakky-outbox "spakky-sqlalchemy[outbox]"

참고: spakky-outbox는 core abstraction을 제공하고, spakky-sqlalchemy는 spakky.contributions.spakky.outbox contribution으로 Outbox storage 구현체를 제공합니다. spakky-sqlalchemy 단독 설치는 spakky-outbox를 끌어오지 않습니다.

주요 기능

• 트랜잭셔널 Outbox: 이벤트를 비즈니스 데이터와 원자적으로 저장
• 자동 relay: background relay가 이벤트를 외부 transport(Kafka, RabbitMQ)로 발행
• 재시도 지원: transport가 영구적이고 특정 레코드에 귀속되는 거부를 확정한 경우에만 설정 가능한 한도 내에서 재시도하며, connection·timeout·queue 등 transport 자체 장애는 메시지별 retry 예산을 소모하지 않음
• 다중 인스턴스 안전성: 원자적 claim으로 중복 발행 방지
• 파티션 키 단위 순서 보전: 파티션 키를 통째로 claim하고, 영구 거부되어 재시도 대기 중인 키의 후속 메시지를 보류
• 발행 포기 상태: 재시도를 소진한 메시지를 abandoned 처리하여 키가 무기한 막히지 않게 함

사용법

1. 애플리케이션에서 플러그인 로드

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

app = (
    SpakkyApplication(ApplicationContext())
    .load_plugins()  # outbox와 sqlalchemy plugin을 자동 로드
    .scan()
    .start()
)

2. Use case에서 이벤트 발행

IAsyncEventPublisher로 발행된 이벤트는 자동으로 라우팅됩니다.

• AbstractDomainEvent → in-process dispatch
• AbstractIntegrationEvent → IEventBus(Outbox가 @Primary로 가로챔)

from spakky.core.common.mutability import immutable
from spakky.core.stereotype.usecase import UseCase
from spakky.data.aspects.transactional import Transactional
from spakky.domain.models.event import AbstractIntegrationEvent
from spakky.event.event_publisher import IAsyncEventPublisher


@immutable
class OrderCreatedEvent(AbstractIntegrationEvent):
    order_id: int
    customer_id: int


@UseCase()
class CreateOrderUseCase:
    def __init__(self, event_publisher: IAsyncEventPublisher) -> None:
        self._event_publisher = event_publisher

    @Transactional()
    async def execute(self, command: CreateOrderCommand) -> Order:
        order = Order.create(...)
        # 이벤트는 order와 같은 transaction에 저장됨
        await self._event_publisher.publish(
            OrderCreatedEvent(order_id=order.id, customer_id=command.customer_id)
        )
        return order

3. 환경변수로 설정

구성 요소

OutboxEventBus와 AsyncOutboxEventBus는 메시지를 저장할 때 DirectEventBus와 같은 outbound metadata 정책을 사용합니다. 기존 tracing header는 보존하고, auth snapshot 전파가 활성화된 request-scope에서는 spakky.auth.context_snapshot에 signed AuthContextSnapshot envelope를 저장합니다. Raw bearer token은 Outbox headers에 저장하지 않습니다.

커스텀 Storage 구현

custom storage backend를 구현하려면 다음 interface를 구현합니다.

from spakky.outbox.ports.storage import IAsyncOutboxStorage
from spakky.outbox.common.message import OutboxMessage

class MyCustomStorage(IAsyncOutboxStorage):
    async def save(self, message: OutboxMessage) -> None:
        # 현재 transaction 안에서 저장
        ...

    async def fetch_pending(self, limit: int, max_retry: int) -> list[OutboxMessage]:
        # pending message를 atomic claim 후 반환.
        # 파티션 키가 있는 메시지는 그 키에서 아직 발행되지도 포기되지도 않은
        # 가장 오래된 메시지를 함께 claim할 때만 넘긴다 — 그러지 않으면 릴레이
        # 인스턴스 둘이 같은 키를 병렬 발행하여 키가 보장하려던 순서를 잃는다.
        ...

    async def mark_published(self, message_id: UUID) -> None:
        ...

    async def increment_retry(self, message_id: UUID) -> None:
        ...

    async def mark_abandoned(self, message_id: UUID) -> None:
        # 재시도를 소진한 메시지를 발행 대기열에서 빼되 레코드는 남긴다.
        # 이 상태가 없으면 그 메시지가 같은 키의 후속 메시지를 영구히 막는다.
        ...

개발 검증

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

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

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

라이선스

MIT License