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 계약을 검증합니다. execute()를 생략하면 framework runner가 RunAgentInput을 받는 runner-backed execute()를 합성합니다. FastAPI WebSocket, CLI, SSE 같은 Spakky-native inbound adapter는 AgentYield를 transport별 payload로 변환하고, protocol adapter는 AgentRunner.run_events()의 AgentEvent를 사용합니다.

RunAgentInput¶

spakky.agent.inbound.RunAgentInput은 runner-backed Agent 실행을 시작하거나 재개하는 inbound contract입니다. state_id는 run/state id, instruction은 이번 model request의 사용자 지시, conversation_id는 멀티턴 thread id, parent_run_id는 delegation parent를 나타냅니다. resume=True는 durable signal queue에 들어온 승인 결정 같은 외부 입력을 반영해 paused run을 재개할 때 사용합니다.

message_history가 있으면 caller가 history를 직접 제공하는 stateless 경로이고, 없으면 runner가 optional ITaskStore에서 effective_conversation_id로 persisted history를 읽습니다.

ITaskStore¶

spakky.agent.interfaces.task_store.ITaskStore는 core Agent transcript를 conversation_id로 저장하는 server-side session port입니다. ConversationTurn은 user 또는 assistant 발화만 담고, 다음 run의 model request history로 재생됩니다. A2A protocol Task snapshot 저장은 spakky-a2a의 IA2ATaskRepository/SpakkyA2ATaskStore가 담당하므로 core transcript store와 섞지 않습니다.

AgentEvent¶

spakky.agent.event.AgentEvent는 AG-UI, A2A 같은 protocol adapter가 소비하는 protocol-neutral event union입니다. MessageDeltaEvent, ReasoningDeltaEvent, ToolCallStartEvent, ToolCallArgsDeltaEvent, ToolCallEndEvent, ToolCallResultEvent, RunStartedEvent, RunPausedEvent, RunFinishedEvent, StepStartedEvent, StepFinishedEvent, StateSnapshotEvent, StateDeltaEvent, ArtifactEvent가 포함됩니다.

모든 event는 AgentEventAttribution(agent_id, run_id, conversation_id, parent_run_id)를 통해 어떤 agent/run/conversation에서 나왔는지 보존합니다. AG-UI는 이를 runId와 threadId로, A2A는 task id와 context id로 투영합니다.

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은 없습니다.

AgentTeammate¶

AgentExecutionSpec.teammates에 들어가는 delegation 선언입니다. AgentTeammate는 name과 정확히 하나의 binding을 갖습니다. 로컬 teammate는 pod=SomeAgent, 원격 teammate는 card_url="https://.../.well-known/agent-card.json"로 선언합니다. 둘 다 없거나 둘 다 있으면 AgentDefinitionError입니다.

AG-UI Adapter¶

spakky-agui가 제공하는 UI streaming protocol adapter입니다. AgentRunner.run_events()의 AgentEvent를 AG-UI BaseEvent로 투영하고, FastAPI SSE endpoint(add_agui_endpoint), HTTP streaming endpoint(add_agui_http_stream_endpoint), WebSocket endpoint (add_agui_websocket_endpoint), stdio helper를 제공합니다. AG-UI에는 전용 approval event가 없으므로 RunPausedEvent는 hitl_approval deferred tool call로 표현됩니다.

A2A Adapter¶

spakky-a2a가 제공하는 Agent-to-Agent protocol adapter입니다. @A2ACompatible marker와 post-processors는 @Agent spec/tool/teammate에서 AgentCard를 파생하고, JSON-RPC, HTTP+JSON REST, gRPC transport에 선언형으로 연결합니다. A2AAgentDelegate는 원격 AgentCard를 사용해 teammate call을 보내고 remote stream을 child AgentEvent로 parent run에 합류시킵니다.

MCP Adapter¶

spakky-mcp가 제공하는 Model Context Protocol adapter입니다. MCPClient.open_runner()는 run마다 선택된 외부 MCP server tools를 lazy mcp_search_tools/mcp_call_tool 경로로 AgentToolCatalog에 합류시킨 AgentRunner를 엽니다. MCP server는 agent가 아니라 외부 도구 카탈로그를 표준 MCP 프로토콜로 제공하는 별도 tool host입니다.

@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¶

일관성 경계를 관리하는 엔터티의 진입점. 도메인 이벤트를 수집하며, 발행은 AggregateCollector와 TransactionalEventPublishingAspect가 트랜잭션 커밋 후 수행합니다. 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를 스캔하여 AbstractDomainEvent를 받는 @on_event 메서드를 in-process Consumer에 자동 등록합니다. AbstractIntegrationEvent는 broker transport 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 구현)

파티션 키 (partition key)¶

Integration Event가 어느 브로커 파티션으로 갈지 결정하는 값입니다. AbstractIntegrationEvent.partition_key property로 노출되며, 기본값 None은 "키 없음"을 의미하여 브로커가 라운드로빈으로 분산합니다.

같은 파티션 키를 가진 이벤트는 항상 같은 파티션으로 갑니다 — Kafka가 보장하는 순서는 파티션 안에서만 성립하므로, 파티션 키는 순서 보장의 전제 조건입니다. 보통 aggregate id를 키로 사용합니다.

Outbox를 경유하는 경로에서는 릴레이가 키 단위 순서를 함께 지킵니다. Transport가 영구적이고 특정 레코드에 귀속 가능한 거부를 EventDeliveryRejectedError로 확정하면 그 키의 후속 메시지를 보류하고 retry 예산을 쓰지만, 연결·timeout·queue·그 밖의 transport 장애는 원래 예외 타입을 유지하며 그 장애 자체로 예산을 소모하지 않습니다. fetch_pending은 키를 통째로 claim하여 여러 릴레이 인스턴스가 같은 키를 병렬 발행하지 않으며, 영구 레코드 귀속 거부로 retry를 소진한 메시지는 발행 포기(abandoned) 처리되어 키를 다시 열어 줍니다 — 상세와 그 대가는 Kafka 가이드의 파티션 키 절 참조.

전달 경로: 이벤트 → IEventBus → (OutboxMessage.partition_key 컬럼 경유) → IEventTransport.send(..., partition_key=...) → Kafka produce(key=...). RabbitMQ transport는 파티션 개념이 없어 이 값을 라우팅에 사용하지 않습니다.

태스크 시스템 (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¶

Repository 저장 경로에서 AggregateRoot 참조를 수집하는 컴포넌트. 트랜잭션 커밋 후 TransactionalEventPublishingAspect가 collector에 모인 AggregateRoot의 이벤트를 발행합니다.

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

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

OutboxEventBus¶

@Primary로 기본 IEventBus를 대체하여, Integration Event를 브로커 대신 Outbox 테이블에 저장합니다. 비즈니스 데이터와 같은 트랜잭션 내에서 원자적으로 기록하고, 브로커가 수락 가능한 레코드는 확인될 때까지 재전송하므로 성공 전달 경로는 중복 가능한 at-least-once 의미를 갖습니다. 영구 레코드 귀속 거부는 retry 소진 후 abandoned 처리되어 성공 전달이 0회일 수 있습니다.

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

OutboxMessage¶

영속성에 독립적인 Outbox 메시지 모델. id, event_name, payload, headers, partition_key, created_at, published_at, retry_count, claimed_at, abandoned_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을 역순 보상

v1 제약상 전체 timeout이 parallel() 그룹 실행 도중 만료되면, 이미 성공했지만 compensable 등록 전인 side effect는 보상되지 않을 수 있습니다. 순차 step이나 이미 완료된 parallel 그룹의 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 필드 번호를 명시적으로 지정할 때 사용하는 선택적 어노테이션입니다. ProtoField를 생략하면 DescriptorBuilder가 필드 이름의 SHA-256 해시에서 번호를 자동 부여합니다. 이름 기반이므로 필드 순서를 바꿔도 기존 번호는 변하지 않고, 필드를 추가해도 기존 번호가 보존됩니다(새 이름의 salt-0 해시가 기존 번호와 충돌하는 약 5.4억분의 1 확률의 경우만 예외). 예약 구간 19000–19999는 자동으로 건너뜁니다. 기존 .proto 계약과 번호를 맞추거나 특정 필드 번호를 무조건 고정해야 할 때 ProtoField(number=N)으로 번호를 직접 지정합니다.

from typing import Annotated

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


# 자동 번호 부여 (기본)
class GetUserRequest(BaseModel):
    user_id: str

# 번호 명시 (기존 계약 유지가 필요한 경우)
class GetUserRequestPinned(BaseModel):
    user_id: Annotated[str, ProtoField(number=1)]

DescriptorRegistry¶

protobuf descriptor를 캐싱하고 관리하는 레지스트리. DescriptorBuilder가 Python 타입에서 descriptor를 자동 생성합니다. spakky-grpc 플러그인이 기본 Pod로 등록하므로 사용자는 listener 주소 같은 애플리케이션 설정만 제공합니다. 등록된 서비스의 전체 이름 목록(service_names)은 서버 리플렉션이 광고할 대상이 됩니다.

GrpcServerSpec¶

grpc.aio.Server를 만들기 전까지 필요한 구성(핸들러, 인터셉터, bind 대상, 채널 옵션, 표준 서비스 등록 콜백)을 누적하는 명세. grpc.aio.server()가 호출 시점의 이벤트 루프에 묶이기 때문에, 실제 서버 생성은 서버를 구동할 루프에서 build()로 미뤄집니다. bind 대상은 주소와 TLS 자격증명을 함께 담으므로 평문 포트와 TLS 포트를 같은 목록에서 표현합니다.

GrpcClient¶

@GrpcController 선언 하나에서 호출 가능한 gRPC callable을 만들어 주는 클라이언트. 호출자가 메시지 모델을 복제하면 필드 이름 한 글자 차이로 필드 번호가 갈라지므로, 서버가 등록하는 것과 같은 컨트롤러 클래스에서 descriptor를 만들고 호출 대상도 메서드 참조로 지정합니다.

from spakky.plugins.grpc.client import GrpcClient

client = GrpcClient(channel, UserServiceController)
response = await client.unary_unary(UserServiceController.get_user)(
    GetUserRequest(user_id="1")
)

표준 서비스 (health·reflection)¶

애플리케이션 서비스와 함께 노출되는 gRPC 생태계 표준 서비스. grpc.health.v1.Health는 Kubernetes의 gRPC 네이티브 프로브가 호출하는 대상이고, 서버 리플렉션은 .proto 산출물이 없는 code-first 서비스를 외부 도구에서 조회하는 유일한 경로입니다. 둘 다 SPAKKY_GRPC_HEALTH_SERVICE_ENABLED·SPAKKY_GRPC_REFLECTION_SERVICE_ENABLED로 끌 수 있습니다.

descriptor 스냅샷¶

컨트롤러에서 생성되는 와이어 배치(메시지 → 필드 이름 → 번호·타입)를 결정론적 JSON으로 렌더링한 결과. spakky-grpc-descriptor-snapshot 명령으로 출력하며, 저장소에 커밋해 두고 CI에서 diff하면 필드 이름 변경이 만드는 와이어 파손을 배포 전에 잡습니다.

분산 트레이싱 (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 인스턴스 생성 중 발생하는 예외.