spakky-saga

Spakky Framework를 위한 분산 트랜잭션 사가 오케스트레이션입니다. @Saga와 @saga_step으로 보상 기반 workflow를 DI-managed application component로 선언합니다.

설치

pip install spakky-saga

주요 기능

• @Saga stereotype: saga orchestrator class를 표시합니다(@Pod 확장).
• AbstractSaga[SagaDataT]: flow() 추상 메서드와 execute() entry point를 가진 generic base class입니다.
• @saga_step decorator: instance method에서 >>, &, | 연산자를 활성화하는 typed descriptor입니다.
• Flow DSL: 연산자 또는 builder 함수(saga_flow, step, parallel)로 선언적 구성
• Error strategy: Compensate(기본값), Skip, ExponentialBackoff을 포함한 Retry(max_attempts, backoff, then)을 제공합니다.
• Timeout: step별 timeout=과 saga별 SagaFlow.timeout() 지원
• 병렬 실행: asyncio.gather 기반 group(& operator / parallel())
• 보상: 실패 시 commit된 step을 역순으로 자동 rollback
• AuthContextSnapshot 전파: AbstractSagaData.auth_context_snapshot signed envelope를 실행, retry, parallel, compensation 동안 유지하고 protected saga step을 enforcement합니다.
• SagaResult[T]: status, data, failed_step, error, history, elapsed를 담는 non-throwing result
• 구조화 로깅: [saga=... step=... status=... elapsed=...ms] format

빠른 시작

SagaData 정의

from spakky.saga import AbstractSagaData


class OrderSagaData(AbstractSagaData):
    order_id: int
    customer_id: int
    ticket_id: int | None = None

Saga 정의

from spakky.saga import AbstractSaga, Saga, SagaFlow, saga_step


@Saga()
class CreateOrderSaga(AbstractSaga[OrderSagaData]):
    @saga_step
    async def issue_ticket(self, data: OrderSagaData) -> OrderSagaData:
        ...

    @saga_step
    async def cancel_ticket(self, data: OrderSagaData) -> None:
        ...

    @saga_step
    async def reserve_stock(self, data: OrderSagaData) -> OrderSagaData:
        ...

    def flow(self) -> SagaFlow[OrderSagaData]:
        return SagaFlow(
            items=(
                self.issue_ticket >> self.cancel_ticket,  # Transaction
                self.reserve_stock,                        # SagaStep (no compensation)
            )
        )

실행

result = await saga.execute(OrderSagaData(order_id=1, customer_id=42))
if result.status is SagaStatus.COMPLETED:
    ...

Protected Saga Step

from spakky.auth import protected, require_scope
from spakky.saga import SagaAuthExecutionContext


@Saga()
class ProtectedOrderSaga(AbstractSaga[OrderSagaData]):
    @saga_step
    @protected
    @require_scope("orders:write")
    async def issue_ticket(self, data: OrderSagaData) -> OrderSagaData:
        ...


result = await saga.execute(
    OrderSagaData(order_id=1, customer_id=42, auth_context_snapshot="signed-envelope"),
    auth_context=SagaAuthExecutionContext(
        snapshot_verifier=snapshot_verifier,
        scope_checker=scope_checker,
    ),
)

Protected step은 raw bearer token을 받지 않고 signed AuthContextSnapshot envelope를 검증합니다. missing, invalid, expired snapshot은 CHALLENGE decision으로 step failure가 되며, provider unavailable은 ERROR decision으로 기존 saga failure/compensation policy를 따릅니다.

Builder 함수 대안

from spakky.saga import Retry, parallel, saga_flow, step


flow = saga_flow(
    step(issue_ticket_fn, compensate=cancel_ticket_fn),
    parallel(reserve_stock_fn, charge_payment_fn),
    step(confirm_order_fn, on_error=Retry(max_attempts=3)),
)

Flow 연산자

Error Strategy

API 레퍼런스

Stereotype / Base

Flow 타입

Builder

실행

Result 타입

에러

관련 문서

• ADR-0007 — architecture decision record
• spakky-domain — AbstractSagaData의 부모인 AbstractDomainModel 제공

개발 검증

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

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

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

라이선스

MIT License