Spakky FastAPI

Spakky Framework를 위한 FastAPI 통합 플러그인입니다. 플러그인이 제공하거나 사용자가 등록한 FastAPI 앱에 @ApiController route, context middleware, lifespan binding, actuator endpoint를 연결합니다.

설치

pip install spakky-fastapi

Spakky extras로도 설치할 수 있습니다.

pip install spakky[fastapi]

주요 기능

• 자동 route 등록: @ApiController 클래스에서 route를 등록합니다.
• 모든 HTTP 메서드: GET, POST, PUT, DELETE, PATCH, HEAD, OPTIONS, WebSocket
• OpenAPI 통합: tag와 documentation 자동 설정
• 에러 처리 middleware: debug mode를 포함한 built-in exception handling
• Context 관리: request-scoped 의존성 주입 지원
• Auth 경계 통합: spakky-auth 보호 handler용 AuthContext seeding
• Actuator endpoint: spakky-actuator 로드 시 선택적 /actuator/* HTTP endpoint 제공

사용법

기본 Controller

from spakky.plugins.fastapi.stereotypes.api_controller import ApiController
from spakky.plugins.fastapi.routes import get, post

@ApiController("/users", tags=["users"])
class UserController:
    def __init__(self, user_service: UserService) -> None:
        self.user_service = user_service

    @get("/{user_id}")
    async def get_user(self, user_id: int) -> User:
        return await self.user_service.get(user_id)

    @post("")
    async def create_user(self, request: CreateUserRequest) -> User:
        return await self.user_service.create(request)

사용 가능한 route decorator

from spakky.plugins.fastapi.routes import (
    get,
    post,
    put,
    delete,
    patch,
    head,
    options,
    websocket,
)

@ApiController("/api")
class MyController:
    @get("/items")
    async def list_items(self) -> list[Item]:
        ...

    @post("/items")
    async def create_item(self, item: CreateItemRequest) -> Item:
        ...

    @put("/items/{item_id}")
    async def update_item(self, item_id: int, item: UpdateItemRequest) -> Item:
        ...

    @delete("/items/{item_id}")
    async def delete_item(self, item_id: int) -> None:
        ...

    @websocket("/ws")
    async def websocket_endpoint(self, websocket: WebSocket) -> None:
        await websocket.accept()
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")

FastAPI 앱 접근

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

import apps
import spakky.plugins.fastapi


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

fast_api: FastAPI = application.container.get(FastAPI)

spakky-fastapi는 기본 FastAPI 앱을 Pod로 제공합니다. 앱 title, description, version, debug는 SPAKKY_FASTAPI_* 환경변수로 설정합니다.

TestClient 테스트

from fastapi.testclient import TestClient

def test_get_user(application: SpakkyApplication) -> None:
    fast_api = application.container.get(FastAPI)
    client = TestClient(fast_api)
    response = client.get("/users/1")
    assert response.status_code == 200

분산 트레이싱

spakky-tracing은 필수 의존성으로 자동 설치됩니다. TracingMiddleware가 자동으로 등록되어 모든 HTTP 요청에 대해 TraceContext를 전파합니다.

• 수신 요청의 traceparent 헤더에서 TraceContext를 추출하여 자식 스팬을 생성합니다
• 헤더가 없으면 새로운 루트 트레이스를 시작합니다
• 요청 완료 후 TraceContext를 자동으로 정리합니다

Actuator endpoint

spakky-actuator 플러그인을 함께 로드하면 FastAPI 앱에 표준 actuator route가 등록됩니다. HTTP payload는 transport-neutral core result shape를 그대로 반영합니다.

FastAPI actuator route는 기본적으로 인증을 자동 적용하지 않습니다. 운영 환경에서는 내부망, API 게이트웨이, 리버스 프록시 허용 목록, 또는 별도의 접근 제어 계층 뒤에서만 노출하세요. 신뢰 경계 밖으로 actuator traffic을 허용하기 전에는 필요 없는 endpoint를 끄고 SPAKKY_ACTUATOR_INCLUDE_DETAILS=false를 설정하세요.

Endpoint 노출과 base path는 플러그인이 등록하는 FastAPIActuatorConfig @Configuration Pod가 읽습니다. Component detail 노출은 spakky.actuator.ActuatorConfig가 읽습니다. readiness는 traffic/work readiness용이며, liveness는 process-local check로 남아 외부 의존성 실패와 독립적이어야 합니다.

export SPAKKY_ACTUATOR_INCLUDE_DETAILS=false
export SPAKKY_FASTAPI_ACTUATOR_BASE_PATH=/internal/actuator
export SPAKKY_FASTAPI_ACTUATOR_READINESS_ENABLED=false

FastAPI adapter는 플러그인별 상세 check를 자동 등록하지 않습니다. 데이터베이스, broker, worker readiness가 actuator 출력에 영향을 줘야 한다면 애플리케이션에 spakky.actuator.AbstractHealthProbe Pod를 등록하세요.

Auth 경계 통합

spakky-fastapi는 HTTP/WebSocket 경계에서 spakky-auth의 AuthContext를 seed합니다. FastAPI wrapper가 Spakky request context를 clear한 뒤 credential 전달체를 추출하고, 사용자 handler 호출 전에 인증 provider로 AuthContext를 저장합니다. 보호된 handler는 인증만을 위해 FastAPI Request, WebSocket, 또는 AuthContext 파라미터를 선언할 필요가 없습니다.

• HTTP는 Authorization: Bearer <token>을 지원합니다.
• WebSocket은 Authorization: Bearer <token>과 access_token=<token> connection query를 지원합니다.
• HTTP auth failure는 CHALLENGE 401, DENY 403, ERROR 500으로 매핑됩니다.
• WebSocket auth failure는 보호된 handler 호출 전에 connection close로 처리됩니다.

구성 요소

설정

기본 앱 설정은 환경변수로 조정합니다:

export SPAKKY_FASTAPI_TITLE="My API"
export SPAKKY_FASTAPI_VERSION="1.0.0"
export SPAKKY_FASTAPI_DEBUG=false

커스텀 FastAPI 인스턴스가 필요하면 플러그인 로드 전에 Pod로 등록합니다. 이 경우 플러그인은 기본 앱을 중복 등록하지 않고, 등록된 앱에 route, middleware, lifespan hook을 연결합니다:

from fastapi import FastAPI
from spakky.core.application.application import SpakkyApplication
from spakky.core.application.application_context import ApplicationContext
from spakky.core.pod.annotations.pod import Pod

import spakky.plugins.fastapi


@Pod()
def custom_fastapi() -> FastAPI:
    return FastAPI(
        title="My API",
        description="Custom API configuration",
        version="1.0.0",
    )

application = (
    SpakkyApplication(ApplicationContext())
    .add(custom_fastapi)
    .load_plugins(include={spakky.plugins.fastapi.PLUGIN_NAME})
    .start()
)

개발 검증

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

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

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

라이선스

MIT