Spakky Framework

Python 3.12+를 위한 Spring-inspired DI/IoC 프레임워크

타입 안전한 의존성 주입, AOP, 이벤트 기반 아키텍처를 Python답게.

---

30초 만에 시작하기

1. 설치

pip install spakky

2. 첫 번째 애플리케이션

from spakky.core.pod.annotations.pod import Pod
from spakky.core.application.application import SpakkyApplication
from spakky.core.application.application_context import ApplicationContext
import apps  # 여러분의 패키지

@Pod()
class GreetingService:
    def greet(self, name: str) -> str:
        return f"Hello, {name}!"

app = (
    SpakkyApplication(ApplicationContext())
    .scan(apps)
    .start()
)

service = app.container.get(type_=GreetingService)
print(service.greet("World"))  # Hello, World!

이것만으로 DI 컨테이너에 GreetingService가 싱글톤으로 등록되고, 타입 기반으로 조회할 수 있습니다.

---

주요 기능

의존성 주입

@Pod 데코레이터 하나로 자동 등록. 생성자 주입으로 의존성을 자동 해결합니다.

@Pod()
class UserRepository:
    def find_by_id(self, user_id: UUID) -> User: ...

@Pod()
class UserService:
    _repo: UserRepository  # 자동 주입

    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    def get_user(self, user_id: UUID) -> User:
        return self._repo.find_by_id(user_id)

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

횡단 관심사를 깔끔하게 분리. Before, After, Around 어드바이스를 지원합니다.

from dataclasses import dataclass
from typing import Any

from spakky.core.aop.aspect import Aspect
from spakky.core.aop.interfaces.aspect import IAspect
from spakky.core.aop.pointcut import Around
from spakky.core.common.annotation import FunctionAnnotation
from spakky.core.common.types import Func

@dataclass
class Timed(FunctionAnnotation):
    """실행 시간 측정 어노테이션"""

@Aspect()
class TimingAspect(IAspect):
    @Around(Timed.exists)
    def around(self, joinpoint: Func, *args: Any, **kwargs: Any) -> Any:
        import time
        start = time.perf_counter()
        result = joinpoint(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{joinpoint.__name__}: {elapsed:.3f}s")
        return result

@Pod()
class DataService:
    @Timed()
    def heavy_computation(self) -> int:
        return sum(range(1_000_000))

DDD 빌딩 블록

Aggregate Root, Entity, Value Object, Domain Event를 기본 제공합니다.

from typing import Self
from uuid import UUID, uuid4

from spakky.core.common.mutability import mutable, immutable
from spakky.domain.models.aggregate_root import AbstractAggregateRoot
from spakky.domain.models.event import AbstractDomainEvent

@mutable
class Order(AbstractAggregateRoot[UUID]):
    customer_name: str
    total_amount: float

    @immutable
    class Created(AbstractDomainEvent):
        order_id: UUID
        total_amount: float

    @classmethod
    def next_id(cls) -> UUID:
        return uuid4()

    def validate(self) -> None:
        if self.total_amount < 0:
            raise ValueError("total_amount must be non-negative")

    @classmethod
    def create(cls, customer_name: str, total_amount: float) -> Self:
        order = cls(uid=cls.next_id(), customer_name=customer_name, total_amount=total_amount)
        order.add_event(Order.Created(order_id=order.uid, total_amount=total_amount))
        return order

이벤트 기반 아키텍처

도메인 이벤트와 통합 이벤트를 자동으로 라우팅합니다. @Transactional 메서드가 성공하면, 수집된 Aggregate의 이벤트가 자동으로 발행됩니다.

from spakky.core.stereotype.usecase import UseCase
from spakky.data.aspects.transactional import transactional
from spakky.event.stereotype.event_handler import EventHandler, on_event

@UseCase()
class CreateOrderUseCase:
    def __init__(self, order_repository: IOrderRepository) -> None:
        self._order_repository = order_repository

    @transactional
    async def execute(self, customer_name: str, total_amount: float) -> Order:
        order = Order.create(customer_name=customer_name, total_amount=total_amount)
        return await self._order_repository.save(order)
        # → Repository.save()가 AggregateCollector에 자동 등록
        # → 메서드 성공 후 Aspect가 order.events를 자동 발행

@EventHandler()
class OrderEventHandler:
    @on_event(Order.Created)
    async def on_order_created(self, event: Order.Created) -> None:
        print(f"주문 생성됨: {event.order_id}")

플러그인 시스템

FastAPI, SQLAlchemy, Celery 등을 플러그인으로 간편하게 통합합니다.

import spakky.plugins.fastapi
import spakky.plugins.sqlalchemy

app = (
    SpakkyApplication(ApplicationContext())
    .load_plugins(include={
        spakky.plugins.fastapi.PLUGIN_NAME,
        spakky.plugins.sqlalchemy.PLUGIN_NAME,
    })
    .scan(apps)
    .start()
)

---

튜토리얼

실제 코드를 따라하며 배울 수 있는 단계별 가이드:

가이드	배울 내용
DI와 Pod	@Pod, 생성자 주입, @Qualifier, @Primary, @Lazy
AOP	@Aspect, Before/After/Around, 커스텀 어노테이션
도메인 모델링	Entity, Value Object, Aggregate Root, Domain Event
이벤트 시스템	@EventHandler, @on_event, EventBus, 이벤트 발행
태스크 & 스케줄링	@task, @schedule, Crontab
FastAPI 통합	@ApiController, HTTP 라우팅, WebSocket
보안	JWT, 암호화, 패스워드 해싱
CLI 애플리케이션	@Controller, Typer 명령어
데이터베이스	Transaction, Repository, ORM
구조화 로깅	@logged, LogContext, 포맷 설정
Celery 태스크	태스크 디스패치, 비동기 실행
RabbitMQ 통합	Integration Event 발행/수신, Exchange 라우팅
Kafka 통합	Integration Event 발행/수신, SASL 인증
Transactional Outbox	at-least-once 전달, Relay, OutboxEventBus
분산 트레이싱	TraceContext, Propagator, W3C traceparent
OpenTelemetry 통합	OTel SDK 브릿지, OTLP exporter, Propagator 자동 교체
Actuator 상태 확인	Health, Readiness, Liveness, Info, probe 확장
사가 오케스트레이션	SagaFlow, SagaStep, 보상 기반 롤백, ErrorStrategy
애플리케이션 데이터 캐시	CacheHit, CacheMiss, @cacheable, Redis 백엔드
gRPC 통합	@GrpcController, @rpc, code-first 프로토콜 생성

---

패키지 구조

코어

패키지	설명
spakky	DI 컨테이너, AOP, 부트스트랩
spakky-domain	DDD 빌딩 블록
spakky-data	Repository, Transaction 추상화
spakky-event	인프로세스 이벤트
spakky-task	태스크 추상화 (스케줄링, 디스패치)
spakky-tracing	분산 트레이싱 추상화
spakky-outbox	Outbox 패턴
spakky-saga	사가 오케스트레이션
spakky-actuator	Actuator 상태/정보 계약
spakky-cache	애플리케이션 데이터 캐시 계약

플러그인

패키지	설명
spakky-logging	구조화된 로깅
spakky-fastapi	FastAPI 통합
spakky-rabbitmq	RabbitMQ 통합
spakky-security	보안 (JWT, 인증)
spakky-typer	Typer CLI 통합
spakky-kafka	Kafka 통합
spakky-sqlalchemy	SQLAlchemy 통합
spakky-celery	Celery 통합
spakky-opentelemetry	OpenTelemetry 브릿지
spakky-grpc	gRPC 통합
spakky-redis	Redis 캐시 백엔드

의존 방향

flowchart LR
  subgraph Plugins[Plugins]
    direction TB
    rabbitmq[spakky-rabbitmq]
    kafka[spakky-kafka]
    sqlalchemy[spakky-sqlalchemy]
    fastapi[spakky-fastapi]
    typer[spakky-typer]
    logging[spakky-logging]
    security[spakky-security]
    celery[spakky-celery]
    opentelemetry[spakky-opentelemetry]
    grpc[spakky-grpc]
    redis[spakky-redis]
  end

  subgraph Core[Core]
    direction TB
    outbox[spakky-outbox] --> event[spakky-event]
    outbox --> tracing_dep[spakky-tracing]
    event --> data[spakky-data]
    event --> tracing_dep
    event --> domain[spakky-domain]
    data --> domain
    domain --> core[spakky]
    task[spakky-task] --> core
    tracing_dep --> core
    cache[spakky-cache] --> core
    saga[spakky-saga] --> domain
    saga --> core
  end

  rabbitmq -. RabbitMQ .-> event
  rabbitmq -- tracing --> tracing_dep
  kafka -. Kafka .-> event
  kafka -- tracing --> tracing_dep
  sqlalchemy -. ORM .-> data
  sqlalchemy -- Outbox --> outbox
  fastapi -. FastAPI .-> core
  fastapi -- tracing --> tracing_dep
  typer -. Typer .-> core
  logging -. 로깅 .-> core
  security -. 인증 .-> core
  celery -. Celery .-> task
  celery -- tracing --> tracing_dep
  opentelemetry -. OTel .-> core
  opentelemetry -- OTel --> tracing_dep
  opentelemetry -. logging .-> logging
  grpc -. gRPC .-> core
  grpc -- tracing --> tracing_dep
  redis -. Redis .-> cache