Spakky Framework 용어 사전¶

Spakky 문서와 코드에서 반복해서 등장하는 용어를 실제 API 이름과 함께 설명합니다.

이 문서는 Spakky Framework에서 사용하는 핵심 용어를 정의합니다.

코어 개념¶

Pod¶

컨테이너가 관리하는 컴포넌트 단위. @Pod 데코레이터로 클래스나 함수를 표시하면 ApplicationContext가 인스턴스 생명주기와 의존성 주입을 담당합니다.

from spakky.core.pod.annotations.pod import Pod

@Pod()
class UserService:
    def __init__(self, repository: UserRepository) -> None:
        self.repository = repository

스코프 (Scope):

SINGLETON — 애플리케이션 전체에서 하나의 인스턴스 공유 (기본값)
PROTOTYPE — 요청할 때마다 새 인스턴스 생성
CONTEXT — 요청/컨텍스트 범위 내에서 인스턴스 공유

ApplicationContext¶

Pod 인스턴스와 생명주기를 관리하는 핵심 컨테이너. 의존성 주입, 서비스 시작/종료, 이벤트 루프 관리를 담당합니다.

from spakky.core.application.application_context import ApplicationContext

context = ApplicationContext()
context.add(UserService)
context.start()

Optional DI (`get_or_none`)¶

컨테이너에 등록되어 있지 않을 수 있는 Pod를 안전하게 조회하는 패턴입니다. ApplicationContext.get_or_none(type_)은 Pod가 등록되어 있으면 인스턴스를 반환하고, 없으면 None을 반환합니다. get()과 달리 NoSuchPodError를 발생시키지 않습니다.

from spakky.core.pod.interfaces.container import IContainer

propagator = container.get_or_none(ITracePropagator)
if propagator is not None:
    propagator.inject(carrier)

이 패턴은 플러그인 간 선택적 의존성에 사용됩니다. 예를 들어, spakky-fastapi의 AddBuiltInMiddlewaresPostProcessor는 get_or_none(ITracePropagator)로 propagator를 조회하고, 있으면 TracingMiddleware를 등록합니다.

Collection DI¶

같은 타입 또는 interface를 구현하는 모든 Pod 후보를 하나의 collection으로 주입하는 패턴입니다. 생성자 파라미터에 list[T], tuple[T, ...], dict[str, T]를 선언하면 ApplicationContext가 매칭 후보를 Pod name 기준의 안정적인 순서로 주입합니다. dict[str, T]는 Pod name을 key로 사용합니다.

@Pod()
class NotificationFanout:
    def __init__(self, senders: dict[str, IEmailSender]) -> None:
        self.senders = senders

지원하지 않는 collection 형태는 UnsupportedCollectionDependencyTypeError를 발생시킵니다. 필수 collection 의존성에 매칭 후보가 없으면 Pod 인스턴스화가 실패하므로, 없을 수 있는 후보 집합은 optional 의존성이나 기본값으로 표현합니다.

SpakkyApplication¶

애플리케이션 부트스트랩 진입점. 컴포넌트 스캔, 플러그인 로딩, 컨테이너 설정을 위한 fluent API를 제공합니다.

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

app = SpakkyApplication(ApplicationContext())
app.scan()  # 현재 패키지의 Pod 자동 스캔

DiscoveryManifest¶

SpakkyApplication.scan()의 discovery 결과를 재사용하기 위한 선택형 JSON artifact입니다. enable_discovery_manifest()로 활성화하며, schema version, Python version, scan 대상, exclude pattern, source file mtime/size fingerprint가 일치할 때만 hit로 사용됩니다. decision은 miss, hit, stale_schema, stale_input 중 하나로 시작 진단에 기록됩니다.

Actuator¶

애플리케이션의 health, readiness, liveness, info 상태를 전송 방식과 독립적인 모델로 집계하는 상태 확인 표면입니다. spakky-actuator는 probe와 info contributor를 ActuatorExtensionRegistry에 모으고, ActuatorAggregationService가 endpoint별 결과를 평가합니다.

from spakky.actuator import AbstractHealthProbe
from spakky.core.pod.annotations.pod import Pod


@Pod()
class DatabaseProbe(AbstractHealthProbe):
    ...

Redis Cache Backend¶

spakky-cache의 backend-neutral ICache[T] 계약을 Redis 저장소로 구현하는 플러그인입니다. spakky-redis는 RedisCache[T], RedisCacheConfig, Actuator health/info contributor를 제공합니다.

from spakky.cache import cacheable
from spakky.core.stereotype.usecase import UseCase


@UseCase()
class ProfileService:
    @cacheable(key="profile:{0}")
    def load_profile(self, user_id: str) -> str:
        return f"profile:{user_id}"

AuthContext¶

spakky-auth가 정의하는 request/context-scoped 인증 상태입니다. Inbound adapter나 snapshot verifier가 ApplicationContext context value에 저장하며, 보호된 boundary는 사용자 메서드 인자 대신 이 context를 읽어 인증 주체, tenant, role, scope, safe claim을 평가합니다.

from spakky.auth import AuthContext, AuthSubject

auth_context = AuthContext(
    subject=AuthSubject(id="user:alice"),
    issuer="issuer:local",
    scopes=("documents:read",),
)

Auth Requirement Decorator¶

클래스나 메서드 boundary에 provider에 묶이지 않는 auth metadata를 붙이는 decorator입니다. @protected, @require_scope, @require_role, @require_permission, @require_policy, @require_relation은 AND semantics로 결합되며, 실제 decision은 등록된 provider port가 수행합니다.

from spakky.auth import protected, require_scope

@protected
@require_scope("documents:read")
def read_document(document_id: str) -> str:
    ...

Auth Provider Capability¶

provider plugin이 spakky.contributions.spakky.auth entry point로 선언하는 기능 단위입니다. spakky-auth startup validation은 보호된 boundary와 snapshot propagation이 요구하는 AUTHENTICATION, POLICY_EVALUATION, PERMISSION_CHECK, ROLE_CHECK, SCOPE_CHECK, RELATION_CHECK, SNAPSHOT_SIGN, SNAPSHOT_VERIFY capability가 정확히 하나의 provider에 의해 충족되는지 확인합니다. PASSWORD_HASH와 PASSWORD_VERIFY는 provider가 선언하고 사용할 수 있는 capability지만, 현재 startup validation은 password-only provider의 고유성을 startup requirement로 계산하지 않습니다.

AuthContextSnapshot¶

task, broker, event, saga 같은 비동기 경계에서 raw bearer token 대신 전파하는 provider에 묶이지 않는 인증 snapshot입니다. spakky-cryptography는 HMAC 기반 SNAPSHOT_SIGN/SNAPSHOT_VERIFY provider를 제공하고, consumer boundary는 snapshot을 검증한 뒤 AuthContext를 seed합니다.

OIDC Provider¶

spakky-oidc가 제공하는 bearer JWT 인증 provider입니다. Discovery document와 JWKS를 읽고 issuer, audience, signature, time claim을 검증한 뒤 선택된 safe claim만 AuthContext로 옮깁니다. Browser login, callback, session route는 애플리케이션이나 별도 서비스가 담당합니다.

OpenFGA Provider¶

spakky-openfga가 제공하는 관계 기반 인가 provider입니다. @require_relation 같은 요구사항을 OpenFGA check 요청으로 변환해 ALLOW/DENY/ERROR 결정을 반환합니다. Tuple write, authorization model migration, list resources 같은 관리 기능은 제공하지 않습니다.

Policy Document Evaluator¶

spakky-policy가 제공하는 policy document 평가기입니다. YAML, TOML, JSON으로 작성한 정책 문서를 typed model로 바꾼 뒤 PolicyDocumentEvaluator가 resource/action, permission, role, scope 요구사항을 평가합니다. 명시적인 deny가 우선하고, 일치하는 allow가 없으면 기본적으로 거부합니다.

Kafka Transport¶

spakky-kafka가 제공하는 Integration Event 전송 플러그인입니다. Event system의 IAsyncEventTransport 구현으로 동작하며, 비즈니스 코드는 Kafka client가 아니라 IAsyncEventPublisher 또는 IAsyncEventBus에 의존합니다.

RabbitMQ Transport¶

spakky-rabbitmq가 제공하는 Integration Event 전송 플러그인입니다. RabbitMQ consumer는 메시지 header에서 trace/auth snapshot을 복원하고, 등록된 event handler를 호출합니다. 보호된 handler의 ack/nack 정책은 RabbitMQ auth 설정으로 조정할 수 있습니다.

Stereotype 데코레이터¶

Pod의 역할을 명시하는 특화된 데코레이터입니다. 기능적으로 @Pod와 동일하지만 의도를 명확히 합니다.

@Configuration¶

설정 값을 담는 클래스 자체를 Pod로 등록하는 @Pod 특화 stereotype입니다. 환경변수 기반 설정은 보통 BaseSettings를 상속한 @Configuration 클래스로 선언합니다. 다른 Pod를 만드는 factory는 클래스 내부 메서드가 아니라 모듈 수준 @Pod() 함수로 둡니다.

from spakky.core.stereotype.configuration import Configuration
from pydantic_settings import BaseSettings

@Configuration()
class DatabaseConfig(BaseSettings):
    pool_size: int = 10
    connection_url: str = "postgresql://localhost/app"

@Controller¶

외부 요청을 처리하는 클래스 (HTTP, CLI 등).

from spakky.core.stereotype.controller import Controller

@Controller()
class UserController:
    def __init__(self, service: UserService) -> None:
        self.service = service

@UseCase¶

비즈니스 로직을 캡슐화하는 애플리케이션 서비스.

from spakky.core.stereotype.usecase import UseCase

@UseCase()
class CreateUserUseCase:
    def execute(self, command: CreateUserCommand) -> User:
        ...

@Agent¶

LLM 기반 orchestration을 수행하는 application workflow component입니다. @UseCase와 동격인 @Pod 계열 stereotype이며, inbound adapter에서 호출되고 constructor DI로 model/workspace/shell/git/repository 같은 outbound port를 주입받습니다.

@Agent의 execute()는 AgentYield stream을 반환할 수 있습니다. token, progress, tool, evidence, approval, final, error, cancel event는 FastAPI WebSocket, CLI, SSE 같은 inbound adapter가 transport별 payload로 변환합니다.

IAgentModel¶

spakky-agent가 소유하는 model outbound port입니다. spakky-vllm은 이 port의 첫 공식 provider plugin으로, OpenAI-compatible vLLM HTTP/SSE 응답을 공통 ModelResponse와 ModelStreamEvent로 변환합니다.

Agent Persistence Contribution¶

Durable Agent 실행에 필요한 IAgentStateRepository, IAgentSignalRepository, IAgentEvidenceRepository 구현을 provider plugin이 기여하는 방식입니다. SQLAlchemy 구현은 spakky.contributions.spakky.agent entry point로 제공되며, 운영용 in-memory fallback은 없습니다.

@Repository¶

데이터 접근 계층. 영속성 저장소와의 상호작용을 추상화합니다.

from spakky.data.stereotype.repository import Repository

@Repository()
class UserRepository:
    def find_by_id(self, id: UUID) -> User | None:
        ...

@EventHandler¶

이벤트를 처리하는 클래스. @on_event 데코레이터와 함께 사용합니다.

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

@EventHandler()
class UserEventHandler:
    @on_event(UserCreatedEvent)
    async def handle(self, event: UserCreatedEvent) -> None:
        ...

AOP (관점 지향 프로그래밍)¶

Aspect¶

횡단 관심사 (로깅, 트랜잭션 등)를 모듈화하는 컴포넌트. IAspect 또는 IAsyncAspect 인터페이스를 구현합니다.

from spakky.core.aop.aspect import Aspect
from spakky.core.aop.interfaces.aspect import IAspect

@Aspect()
class LoggingAspect(IAspect):
    def before(self, *args, **kwargs) -> None:
        print("Method called")

Advice¶

Aspect가 특정 시점에 실행하는 액션:

Before — 메서드 실행 전
AfterReturning — 메서드 정상 반환 후
AfterRaising — 메서드 예외 발생 후
After — 메서드 실행 후 (결과와 무관)
Around — 메서드 실행을 감싸서 제어

Pointcut¶

Advice가 적용될 메서드를 선택하는 조건자 함수.

from spakky.core.aop.pointcut import Around

def is_service_method(method) -> bool:
    return hasattr(method, "__service__")

@Around(pointcut=is_service_method)
def around(self, joinpoint, *args, **kwargs):
    return joinpoint(*args, **kwargs)

JoinPoint¶

Aspect가 개입할 수 있는 프로그램 실행 지점. Around advice에서 다음 호출을 제어할 때 사용합니다.

도메인 모델 (spakky-domain)¶

Entity¶

고유 식별자로 구분되는 도메인 객체. AbstractEntity를 상속합니다.

from spakky.domain.models.entity import AbstractEntity

@mutable
class User(AbstractEntity[UUID]):
    name: str
    email: str

ValueObject¶

식별자 없이 속성값으로만 동등성을 판단하는 불변 객체. AbstractValueObject를 상속합니다.

from spakky.domain.models.value_object import AbstractValueObject

@immutable
class Email(AbstractValueObject):
    value: str

AggregateRoot¶

일관성 경계를 관리하는 엔터티의 진입점. 도메인 이벤트를 수집하고 발행합니다. AbstractAggregateRoot를 상속합니다.

from spakky.domain.models.aggregate_root import AbstractAggregateRoot

@mutable
class Order(AbstractAggregateRoot[UUID]):
    items: list[OrderItem]

    def add_item(self, item: OrderItem) -> None:
        self.items.append(item)
        self.add_event(ItemAddedEvent(order_id=self.uid, item=item))

이벤트 시스템 (spakky-event)¶

설계 배경 및 대안 분석은 ADR-0001을 참조하세요.

DomainEvent¶

하나의 바운디드 컨텍스트 내에서 발생하는 도메인 상태 변경. AbstractDomainEvent를 상속합니다.

from spakky.domain.models.event import AbstractDomainEvent

@immutable
class UserCreatedEvent(AbstractDomainEvent):
    user_id: UUID
    email: str

IntegrationEvent¶

바운디드 컨텍스트 간 또는 서비스 간 통신에 사용되는 이벤트. AbstractIntegrationEvent를 상속합니다.

from spakky.domain.models.event import AbstractIntegrationEvent

@immutable
class OrderPlacedEvent(AbstractIntegrationEvent):
    order_id: UUID
    total_amount: Decimal

동사 규칙¶

이벤트 시스템 전체에서 동사를 다음과 같이 구분합니다:

동사	의미	사용 레이어
`publish`	이벤트를 시스템에 발행 (호출자가 경로를 모름)	EventPublisher
`send`	Integration Event를 외부로 전송	EventBus, EventTransport
`dispatch`	등록된 핸들러에 인프로세스 전달	Dispatcher, Mediator
`register`	이벤트 타입에 핸들러 콜백 등록	Consumer

이벤트 인터페이스¶

설계 배경은 ADR-0001 참조.

역할	인터페이스 (sync / async)	설명
발행 진입점	`IEventPublisher` / `IAsyncEventPublisher`	타입 기반 라우팅
인프로세스 전달	`IEventDispatcher` / `IAsyncEventDispatcher`	핸들러에 dispatch
핸들러 등록	`IEventConsumer` / `IAsyncEventConsumer`	콜백 등록
외부 전송 진입점	`IEventBus` / `IAsyncEventBus`	Outbox seam
실제 메시지 전송	`IEventTransport` / `IAsyncEventTransport`	Kafka/RabbitMQ

주요 구현체:

구현체	역할
`EventMediator`	Consumer + Dispatcher 통합 (인프로세스)
`EventPublisher`	`match event:` 타입 라우터
`DirectEventBus`	기본 EventBus → Transport 위임
`KafkaEventTransport`	Kafka Transport 구현
`RabbitMQEventTransport`	RabbitMQ Transport 구현

Consumer와 EventHandler¶

Consumer — 핸들러를 등록하는 인터페이스 (register(event_type, callback))
EventHandler — 이벤트를 처리하는 클래스 스테레오타입 (@EventHandler + @on_event)

EventHandlerRegistrationPostProcessor가 @EventHandler Pod를 스캔하여 @on_event 메서드를 Consumer에 자동 등록합니다.

EventPublisher¶

이벤트를 발행하는 단일 진입점:

IEventPublisher / IAsyncEventPublisher — publish(event: AbstractEvent) → 타입 기반 라우팅
AbstractDomainEvent → EventMediator (인프로세스 dispatch)
AbstractIntegrationEvent → IEventBus (외부 전송)

EventBus와 EventTransport¶

Integration Event 전송을 2단 인터페이스로 분리:

EventBus (IEventBus) — Integration Event 발행 진입점. Outbox seam 역할
EventTransport (IEventTransport) — 실제 메시지 브로커 전송 (Kafka/RabbitMQ 구현)

태스크 시스템 (spakky-task)¶

설계 배경은 ADR-0003 참조.

@TaskHandler¶

태스크 핸들러 클래스를 마크하는 스테레오타입. @task 및 @schedule 메서드를 그룹화합니다.

from spakky.task import TaskHandler, task, schedule
from datetime import timedelta

@TaskHandler()
class EmailTaskHandler:
    @task
    def send_email(self, to: str) -> None: ...

    @schedule(interval=timedelta(hours=1))
    def cleanup(self) -> None: ...

@task¶

메서드를 온디맨드 디스패치 대상으로 마크하는 데코레이터. 플러그인의 AOP Aspect가 호출을 가로채 태스크 큐로 전달합니다.

@schedule¶

메서드를 정기 실행 대상으로 마크하는 데코레이터. interval, at, crontab 중 정확히 하나를 지정해야 합니다.

파라미터	타입	설명
`interval`	`timedelta`	고정 간격 실행
`at`	`time`	매일 특정 시각 실행
`crontab`	`Crontab`	Cron 기반 스케줄

Crontab¶

Python 네이티브 타입 기반 cron 명세 값 객체. 문자열 대신 Weekday/Month IntEnum을 사용합니다.

from spakky.task import Crontab, Weekday, Month

# 매주 월요일 09:00
Crontab(weekday=Weekday.MONDAY, hour=9)

# 매년 1월 1일 자정
Crontab(month=Month.JANUARY, day=1)

필드 순서: month → day → weekday → hour → minute (내림차순 시간 척도)

서비스 생명주기¶

IService¶

시작/종료 생명주기를 가진 동기 서비스 인터페이스.

import threading
from spakky.core.service.interfaces.service import IService

class BackgroundWorker(IService):
    def set_stop_event(self, stop_event: threading.Event) -> None: ...
    def start(self) -> None: ...
    def stop(self) -> None: ...

IAsyncService¶

시작/종료 생명주기를 가진 비동기 서비스 인터페이스.

from asyncio import locks
from spakky.core.service.interfaces.service import IAsyncService

class AsyncWorker(IAsyncService):
    def set_stop_event(self, stop_event: locks.Event) -> None: ...
    async def start_async(self) -> None: ...
    async def stop_async(self) -> None: ...

플러그인 시스템¶

Plugin¶

프레임워크 확장을 식별하는 불변 객체. 이름으로 구분됩니다.

from spakky.core.application.plugin import Plugin

FASTAPI_PLUGIN = Plugin(name="spakky-fastapi")

플러그인 로딩:

app.load_plugins(include={FASTAPI_PLUGIN})

데이터 계층 (spakky-data)¶

@Transactional¶

메서드에 트랜잭션 경계를 적용하는 Aspect.

AggregateCollector¶

AggregateRoot에서 발생한 도메인 이벤트를 수집하여 트랜잭션 커밋 시 발행하는 컴포넌트.

트랜잭셔널 Outbox (spakky-outbox)¶

설계 배경은 ADR-0002, ADR-0006 참조.

OutboxEventBus¶

@Primary로 기본 IEventBus를 대체하여, Integration Event를 브로커 대신 Outbox 테이블에 저장합니다. 비즈니스 데이터와 같은 트랜잭션 내에서 원자적으로 기록되므로 at-least-once 전달을 보장합니다.

from spakky.outbox.bus.outbox_event_bus import OutboxEventBus, AsyncOutboxEventBus

OutboxMessage¶

영속성에 독립적인 Outbox 메시지 모델. id, event_name, payload, headers, created_at, published_at, retry_count, claimed_at 필드를 가집니다.

from spakky.outbox.common.message import OutboxMessage

IOutboxStorage¶

Outbox 메시지의 CRUD를 담당하는 포트 인터페이스. spakky-sqlalchemy는 spakky.contributions.spakky.outbox contribution으로 SQLAlchemy 구현체와 Outbox table을 제공합니다.

from spakky.outbox.ports.storage import IOutboxStorage, IAsyncOutboxStorage

OutboxRelayBackgroundService¶

백그라운드 서비스로 Outbox 테이블을 폴링하여 미전송 메시지를 IEventTransport로 전송합니다.

구조화 로깅 (spakky-logging)¶

@logged¶

메서드에 자동 로깅을 적용하는 어노테이션. 인자, 반환값, 실행 시간을 자동 기록합니다. enable_masking, masking_keys, slow_threshold_ms, max_result_length, log_args, log_result 파라미터를 지원합니다.

from spakky.plugins.logging import logged

@logged()
def create_user(self, name: str) -> str: ...

LogContext¶

contextvars 기반 컨텍스트 전파. bind(), unbind(), clear(), get(), scope() 메서드로 요청 ID 등 공통 정보를 모든 로그에 자동 주입합니다.

from spakky.plugins.logging import LogContext

LogContext.bind(request_id="abc-123")

LoggingConfig¶

로깅 설정 @Configuration Pod. 환경변수 접두사 SPAKKY_LOGGING__로 구성합니다. level, format (TEXT/JSON/PRETTY), mask_keys, slow_threshold_ms 등을 설정합니다.

암호화 유틸리티 (spakky-cryptography)¶

HMAC¶

공유 Key로 문자열을 서명하고 검증하는 유틸리티. HMACType은 HS224, HS256, HS384, HS512 알고리즘을 표현합니다.

from spakky.plugins.cryptography.hmac_signer import HMAC, HMACType
from spakky.plugins.cryptography.key import Key

key = Key(size=32)
signature = HMAC.sign_text(key, HMACType.HS256, "message")
is_valid = HMAC.verify(key, HMACType.HS256, "message", signature)

Password Encoder¶

패스워드 해시를 생성하고 입력 패스워드를 검증하는 인코더 계열. Argon2PasswordEncoder, BcryptPasswordEncoder, ScryptPasswordEncoder, Pbkdf2PasswordEncoder가 encode()와 challenge(password) 계약을 제공합니다.

from spakky.plugins.cryptography.password.argon2 import Argon2PasswordEncoder

hashed = Argon2PasswordEncoder(password="secret").encode()
is_valid = Argon2PasswordEncoder(password_hash=hashed).challenge("secret")

CLI 통합 (spakky-typer)¶

@CliController¶

Typer 하위 명령 그룹으로 등록되는 CLI 컨트롤러 스테레오타입. group_name을 생략하면 클래스명을 kebab-case로 변환한 그룹명이 사용됩니다.

from spakky.plugins.typer.stereotypes.cli_controller import CliController

@CliController("users")
class UserCliController:
    ...

@command¶

@CliController 메서드를 Typer 명령으로 표시하는 데코레이터. name, help, short_help, hidden, deprecated 등 Typer 명령 옵션을 보존하고 TyperCLIPostProcessor가 컨테이너 초기화 중 등록합니다.

from spakky.plugins.typer.stereotypes.cli_controller import command

@command("create")
def create_user(self, name: str, email: str) -> None:
    ...

TyperCLIPostProcessor¶

@CliController Pod에서 @command 메서드를 스캔해 Typer 앱에 하위 명령 그룹을 추가하는 후처리기. 명령 실행 시 컨텍스트 스코프를 비우고 컨테이너에서 컨트롤러 인스턴스를 다시 조회합니다.

Celery 통합 (spakky-celery)¶

Celery 태스크 디스패치¶

@task 메서드 호출을 AOP로 가로채 Celery send_task() 호출로 변환하는 디스패치 패턴. 워커 컨텍스트 내부에서는 재디스패치하지 않고 원래 메서드를 직접 실행합니다.

from spakky.task.stereotype.task_handler import TaskHandler, task

@TaskHandler()
class EmailTaskHandler:
    @task
    def send_email(self, to: str) -> None: ...

CeleryTaskResult¶

Celery의 AsyncResult를 spakky-task의 태스크 결과 계약에 맞게 감싸는 결과 객체. 디스패치된 태스크의 task_id, 블로킹 get(), 비동기 get_async() 조회를 제공합니다.

Celery Beat 스케줄¶

@schedule 메서드를 Celery Beat beat_schedule 항목으로 등록하는 스케줄 통합. interval, at, Crontab 라우트를 Celery schedule 또는 crontab 객체로 변환합니다.

from datetime import timedelta

from spakky.task.stereotype.schedule import schedule

@schedule(interval=timedelta(minutes=30))
def health_check(self) -> None: ...

사가 시스템 (spakky-saga)¶

@Saga¶

사가 오케스트레이터 클래스를 마크하는 스테레오타입. @Pod의 서브클래스이므로 패키지 스캔만으로 DI 컨테이너에 자동 등록됩니다.

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

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

    def flow(self) -> SagaFlow[OrderSagaData]:
        return saga_flow(self.create_order)

AbstractSaga¶

사가의 기본 추상 클래스. flow()를 구현하여 사가 흐름을 선언하고, execute(data)로 실행합니다. 내부적으로 run_saga_flow에 클래스명을 saga_name으로 전달하여 구조화 로그에 포함합니다.

@saga_step¶

사가 step 메서드를 _SagaStepDescriptor로 감싸는 데코레이터. 인스턴스 접근 시 bound 메서드가 SagaStep으로 승격되어 >>, &, | 연산자가 타입 안전하게 동작합니다.

from spakky.saga import saga_step

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

SagaStep, Transaction, Parallel¶

DSL의 기본 빌딩 블록.

타입	역할	생성 방법
`SagaStep[T]`	단일 action (compensate 없음)	`step(action)`, `@saga_step` 메서드
`Transaction[T]`	action + compensate 쌍	`step(action, compensate=...)`, `action >> compensate`
`Parallel[T]`	동시 실행 그룹 (최소 2개)	`parallel(...)`, `a & b`

SagaFlow¶

최상위 흐름 정의. saga_flow(*items)로 생성하며, .timeout(duration)·.on_compensation_failure(handler)로 실행 옵션을 덧붙입니다. >> (action + compensate 쌍), & (병렬), | (에러 전략) 연산자를 지원합니다.

from datetime import timedelta

from spakky.saga import parallel, saga_flow, step

flow = saga_flow(
    step(saga.create_order, compensate=saga.cancel_order),
    parallel(
        step(saga.reserve_stock, compensate=saga.release_stock),
        step(saga.process_payment, compensate=saga.refund_payment),
    ),
).timeout(timedelta(seconds=30))

ErrorStrategy¶

사가 step 실패 시 적용할 전략:

전략	설명
`Compensate()`	역순 보상 후 FAILED 반환 (기본값)
`Skip()`	실패를 무시하고 다음 step으로 진행
`Retry(max_attempts, backoff, then)`	재시도 후 `then` 전략 적용
`ExponentialBackoff(base=1.0)`	`Retry.backoff`용 지수 백오프 (`base * 2^(attempt-1)`)

parallel() 그룹 내부의 step은 v1에서 기본 Compensate 외 on_error를 지정할 수 없습니다.

타임아웃¶

수준	지정 방법	동작
step	`step(..., timeout=timedelta(...))`	초과 시 `SagaStepTimeoutError`가 `on_error` 전략을 거침
saga	`SagaFlow.timeout(duration)`	초과 시 `SagaStatus.TIMED_OUT`으로 종료하고 commit된 step을 역순 보상

run_saga_flow¶

SagaFlow를 실행하는 얇은 엔트리. AbstractSaga.execute가 내부적으로 호출합니다.

from spakky.saga import run_saga_flow

result = await run_saga_flow(flow, data, saga_name="OrderSaga")

SagaResult¶

사가 실행 결과. status (SagaStatus), data, failed_step, error, history (tuple[StepRecord, ...]), elapsed 필드를 가집니다. 예외는 발생시키지 않습니다 (보상 실패 시 SagaCompensationFailedError는 예외).

StepRecord / StepStatus¶

각 step의 실행 기록. status는 COMMITTED / FAILED / COMPENSATED 중 하나입니다.

SagaStatus¶

사가 전체 상태 열거형: STARTED, RUNNING, COMPENSATING, COMPLETED, FAILED, TIMED_OUT.

AbstractSagaData¶

사가 비즈니스 데이터 모델의 기본 클래스. @immutable + AbstractDomainModel을 확장하며, saga_id: UUID 필드가 기본 제공됩니다. 엔진 상태는 포함하지 않습니다.

gRPC 시스템 (spakky-grpc)¶

@GrpcController¶

gRPC 서비스 컨트롤러 클래스를 마크하는 스테레오타입. @Controller의 서브클래스입니다.

from spakky.plugins.grpc.stereotypes.grpc_controller import GrpcController

@GrpcController(package="example.user", service_name="UserService")
class UserServiceController:
    ...

@rpc¶

메서드를 gRPC RPC 엔드포인트로 마크하는 데코레이터. RpcMethodType을 지정하여 스트리밍 모드를 선택할 수 있습니다.

from spakky.plugins.grpc.decorators.rpc import rpc, RpcMethodType

@rpc(method_type=RpcMethodType.UNARY)
async def get_user(self, request: GetUserRequest) -> GetUserResponse:
    ...

RpcMethodType¶

gRPC 메서드 유형을 나타내는 열거형:

값	설명
`UNARY`	단일 요청, 단일 응답
`SERVER_STREAMING`	단일 요청, 스트림 응답
`CLIENT_STREAMING`	스트림 요청, 단일 응답
`BIDI_STREAMING`	양방향 스트리밍

ProtoField¶

Pydantic BaseModel 필드에 protobuf 메타데이터를 부착하는 어노테이션입니다. code-first protobuf descriptor 생성 시 DescriptorBuilder가 model_fields에서 이 metadata를 읽습니다.

from typing import Annotated

from pydantic import BaseModel
from spakky.plugins.grpc.annotations.field import ProtoField


class GetUserRequest(BaseModel):
    user_id: Annotated[str, ProtoField(number=1)]

DescriptorRegistry¶

protobuf descriptor를 캐싱하고 관리하는 레지스트리. DescriptorBuilder가 Python 타입에서 descriptor를 자동 생성합니다. spakky-grpc 플러그인이 기본 Pod로 등록하므로 사용자는 listener 주소 같은 애플리케이션 설정만 제공합니다.

분산 트레이싱 (spakky-tracing)¶

설계 배경은 ADR-0004 참조.

TraceContext¶

W3C Trace Context Level 2 호환 컨텍스트 객체. trace_id, span_id, parent_span_id, trace_flags를 보유합니다. Python contextvars를 사용하여 asyncio 태스크 간 격리됩니다.

from spakky.tracing.context import TraceContext

ctx = TraceContext.new_root()    # 새 트레이스 시작
child = ctx.child()              # 자식 span 생성
TraceContext.set(child)          # 현재 컨텍스트에 설정
TraceContext.get()               # 현재 컨텍스트 조회
TraceContext.clear()             # 컨텍스트 초기화

ITracePropagator¶

서비스 경계에서 TraceContext를 헤더(carrier)에 주입/추출하는 인터페이스.

메서드	설명
`inject(carrier)`	현재 TraceContext를 carrier에 기록
`extract(carrier)`	carrier에서 TraceContext를 복원 (실패 시 `None`)
`fields()`	사용하는 헤더 필드명 목록

W3CTracePropagator¶

ITracePropagator의 기본 구현체. traceparent 헤더를 사용합니다.

OTelTracePropagator¶

ITracePropagator의 OpenTelemetry 구현체. spakky-opentelemetry 플러그인이 로드되면 OTelSetupPostProcessor가 기본 W3CTracePropagator를 이 구현체로 교체합니다. OpenTelemetry SDK의 TraceContextTextMapPropagator에 위임합니다.

from spakky.plugins.opentelemetry.propagator import OTelTracePropagator

OpenTelemetryConfig¶

spakky-opentelemetry 플러그인의 설정 클래스. 환경변수 접두사 SPAKKY_OTEL_로 구성합니다.

필드	환경변수	기본값
`service_name`	`SPAKKY_OTEL_SERVICE_NAME`	`"spakky-service"`
`exporter_type`	`SPAKKY_OTEL_EXPORTER_TYPE`	`ExporterType.OTLP`
`exporter_endpoint`	`SPAKKY_OTEL_EXPORTER_ENDPOINT`	`"http://localhost:4317"`
`sample_rate`	`SPAKKY_OTEL_SAMPLE_RATE`	`1.0`

LogContextBridge¶

spakky-opentelemetry의 로깅 통합 컴포넌트. 생성자에서 ILogContextBinder | None을 Optional DI로 주입받아, spakky-logging이 등록된 경우 TraceContext의 trace/span ID를 LogContext에 동기화합니다. ILogContextBinder가 컨테이너에 없으면 no-op으로 동작합니다.

from spakky.plugins.opentelemetry.bridge import LogContextBridge

traceparent¶

W3C 표준 분산 트레이싱 헤더. 형식: {version:2}-{trace_id:32}-{span_id:16}-{flags:2}

예시: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01

어노테이션¶

@Primary¶

동일 타입의 여러 Pod 중 기본 선택 대상으로 지정.

@Pod()
@Primary()
class DefaultUserRepository(IUserRepository):
    ...

@Qualifier¶

의존성 주입 시 특정 Pod를 이름으로 지정.

@Pod(name="cache")
class CacheUserRepository(IUserRepository):
    ...

@Pod()
class UserService:
    def __init__(self, repository: Annotated[IUserRepository, Qualifier(lambda p: p.name == "cache")]) -> None:
        ...

PodBinding¶

같은 interface를 구현하는 Pod 후보가 여러 개일 때 application config가 선택할 구현체를 명시하는 binding policy 값입니다. ApplicationContext.bind(), bind_to_name(), bind_to_type()으로 등록하며, Qualifier/name보다 낮고 @Primary보다 높은 우선순위로 단수 의존성을 선택합니다.

from spakky.core.pod.binding import PodBinding

context.bind(PodBinding(interface=IRepository, implementation_name="postgres"))

@Lazy¶

Pod 인스턴스화를 첫 사용 시점까지 지연.

@Order¶

Pod 처리 순서를 지정 (숫자가 낮을수록 우선).

@Tag¶

Pod를 태그로 그룹화하여 일괄 조회 가능.

에러 계층¶

AbstractSpakkyFrameworkError¶

모든 Spakky 프레임워크 예외의 기반 클래스.

PodAnnotationFailedError¶

Pod 어노테이션 처리 중 발생하는 예외.

PodInstantiationFailedError¶

Pod 인스턴스 생성 중 발생하는 예외.