spakky-celery

Spakky Framework를 위한 Celery 통합 플러그인입니다. @task/@schedule metadata를 Celery task dispatch와 beat schedule로 자동 연결합니다.

AOP 기반 자동 task dispatch와 주기 schedule 등록을 제공합니다. @task 또는 @schedule로 장식된 메서드는 명시적인 dispatcher 호출 없이 가로채져 Celery로 라우팅됩니다.

설치

pip install spakky-celery

필수 의존성: spakky-task, spakky-tracing(의존성으로 자동 설치)

주요 기능

• AOP 기반 dispatch: @task 메서드는 aspect가 가로채며 수동 dispatch() 호출이 필요 없습니다.
• Broker dispatch: 모든 @task 호출은 send_task()를 통해 Celery broker로 전송됩니다.
• Schedule 등록: @schedule 메서드는 Celery Beat에 자동 등록됩니다.
• Worker context 감지: context key로 worker 내부 재dispatch 방지
• 자동 등록: @TaskHandler pod를 스캔해 Celery task로 자동 등록합니다.
• 기본 Celery 앱: CeleryConfig 기반 Celery 앱을 Pod로 자동 등록합니다.
• 전체 설정: broker URL, serializer, timezone 등을 환경변수로 설정
• AuthContextSnapshot 전파: spakky-auth snapshot propagation이 활성화되면 raw bearer token 대신 signed snapshot을 task header에 담아 보호된 worker task를 fail-closed로 실행합니다.

설정

SPAKKY_CELERY__ 접두사로 환경변수를 설정합니다.

사용법

1. Task handler 정의

from datetime import time, timedelta

from spakky.task import TaskHandler, Crontab, Weekday, task, schedule


@TaskHandler()
class EmailTaskHandler:
    @task
    def send_email(self, to: str, subject: str, body: str) -> None:
        """Dispatched to Celery broker via send_task()."""
        send_smtp(to, subject, body)


@TaskHandler()
class MaintenanceHandler:
    @schedule(interval=timedelta(minutes=30))
    def health_check(self) -> None:
        """Registered as Celery Beat periodic task — runs every 30 minutes."""
        ...

    @schedule(at=time(3, 0))
    def daily_cleanup(self) -> None:
        """Runs daily at 03:00."""
        ...

    @schedule(crontab=Crontab(weekday=Weekday.MONDAY, hour=9))
    def weekly_report(self) -> None:
        """Runs every Monday at 09:00."""
        ...

2. 애플리케이션 부트스트랩

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

import spakky.plugins.celery

app = (
    SpakkyApplication(ApplicationContext())
    .load_plugins(include={spakky.plugins.celery.PLUGIN_NAME})
    .scan()
    .start()
)

커스텀 Celery 인스턴스를 써야 한다면 플러그인을 로드하기 전에 Celery 반환 Pod를 먼저 등록하세요.

3. Task 메서드 일반 호출

handler = app.container.get(EmailTaskHandler)

# broker로 dispatch됨: AOP가 가로채 send_task() 호출
handler.send_email("user@example.com", "Hello", "World")

AOP aspect가 호출을 가로채 Celery로 자동 라우팅합니다.

Dispatch 동작

구성 요소

에러

분산 트레이싱

spakky-tracing은 필수 의존성으로 자동 설치됩니다. ITracePropagator가 컨테이너에 등록되어 있으면 태스크 디스패치 시 TraceContext가 자동으로 전파됩니다.

• 디스패치 측: @task 호출 시 현재 TraceContext를 Celery 메시지 헤더에 주입합니다
• 워커 측: 수신 태스크에서 TraceContext를 추출하여 자식 스팬을 생성합니다
• 헤더가 없으면 새로운 루트 트레이스를 시작합니다

인증/인가 스냅샷 전파

SPAKKY_AUTH_SNAPSHOT_PROPAGATION_ENABLED=true와 IAuthContextSnapshotSigner / IAuthContextSnapshotVerifier provider가 등록되어 있으면, CeleryTaskDispatchAspect와 AsyncCeleryTaskDispatchAspect는 현재 request/context scope의 AuthContext를 signed AuthContextSnapshot으로 직렬화해 Celery task header spakky.auth.context_snapshot에 저장합니다.

Worker endpoint는 task 실행 시작 시 ApplicationContext.clear_context() 후 spakky.plugins.celery.task_context를 설정하고, 보호된 task metadata가 있으면 snapshot을 검증해 AuthContext를 다시 seed합니다. 이후 @protected, @require_scope, @require_role, @require_permission, @require_policy, @require_relation requirement를 worker boundary에서 fail-closed로 평가합니다.

Raw bearer token은 task/broker header로 전파하지 않습니다. 같은 worker process 안에서 재호출되는 task는 기존 Celery task context를 감지해 재dispatch하지 않고 직접 실행합니다.

관련 패키지

• spakky-task: core task 추상화(@TaskHandler, @task, @schedule, Crontab)
• spakky-rabbitmq: RabbitMQ event transport(Celery broker로도 사용 가능)
• spakky-tracing: 분산 트레이싱 추상화(필수, context propagation 활성화)

개발 검증

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

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

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

라이선스

MIT License