Kontiki, Python용 async 마이크로서비스 프레임워크

Source: Dev.to

Kontiki는 aio‑pika와 asyncio를 통해 AMQP (RabbitMQ) 위에 구축된 파이썬 마이크로‑서비스 프레임워크입니다.
Nameko에 익숙한 팀이라면 전체적인 철학이 친숙하게 느껴질 것입니다:

• RabbitMQ 위에 구축된 메시지‑드리븐 서비스
• RPC와 이벤트를 일급 원시값으로 제공

Kontiki는 같은 “패밀리”의 프레임워크에 속하면서도 다음에 중점을 둡니다:

• asyncio‑네이티브 구현
• 위임‑지향 서비스 구조
• 구성‑기반 실행기

이 프로젝트는 오픈‑소스(Apache‑2.0)이며 코드는 GitHub에서 확인할 수 있습니다:
github.com/kontiki-org/kontiki

서비스 구조

Kontiki는 서비스 클래스를 진입점 레이어로 유지하고 도메인 로직을 델리게이트(ServiceDelegate)로 이동하며 setup/start/stop 라이프사이클 훅을 사용하도록 권장합니다.

# delegate.py
from kontiki.delegate import ServiceDelegate
from kontiki.messaging import Messenger, on_event, rpc

class MyDelegate(ServiceDelegate):
    async def setup(self):
        # initialize from self.container.config
        pass

    async def start(self):
        # optional: start background tasks / open connections
        pass

    async def stop(self):
        # optional: stop background tasks / close connections
        pass

    def do_something(self, x: int) -> int:
        return x * 2

# service.py
class MyService:
    name = "compute-api"          # optional; defaults to class name
    delegate = MyDelegate()
    messenger = Messenger()       # publish events / call other services

    @rpc
    async def compute(self, x: int):
        return self.delegate.do_something(x)

    @on_event("thing_happened")
    async def on_thing(self, payload):
        await self.messenger.publish("thing_processed", {"payload": payload})

서비스 실행

서비스를 cli.run(...)을 통해 CLI 엔트리 포인트로 노출하고 스크립트로 등록합니다.

pyproject.toml에 엔트리 포인트를 추가합니다 (Poetry 예시):

구성 파일 하나(또는 여러 개)와 함께 서비스를 실행합니다.

이벤트 (Fire‑and‑Forget)

소비자는 @on_event("event_type") 로 구독하고, 퍼블리셔는 Messenger.publish(...) 로 이벤트를 발행합니다.

# Server‑side (service)
class UserEventsService:
    @on_event("user.created")
    async def on_user_created(self, payload: dict):
        ...

# Client‑side (stand‑alone script)
async def publish_event_from_script():
    async with Messenger(standalone=True) as messenger:
        await messenger.publish("user.created", {"id": "u_123"})

# Client‑side (from another service)
class BillingService:
    messenger = Messenger()

    @rpc
    async def create_invoice(self, user_id: str):
        await self.messenger.publish("user.created", {"id": user_id})
        return "ok"

RPC (요청/응답)

Kontiki RPC는 AMQP를 통한 동기식 요청/응답 방식입니다.

• Server side – @rpc 로 핸들러를 노출합니다 (rpc_error(code, message) 를 반환할 수 있음).
• Client side – RpcProxy 를 사용해 원격 메서드를 호출합니다.

# Server‑side (service)
class RpcService:
    @rpc
    async def rpc_example(self, feature: str):
        if feature == "bad_input":
            return rpc_error("USER_INPUT_ERROR", "Invalid feature value")
        return "ok"

# Client‑side (stand‑alone script)
async def call_rpc_from_script():
    async with Messenger(standalone=True) as messenger:
        rpc_service = RpcProxy(messenger, service_name="RpcService")
        return await rpc_service.rpc_example("bad_input")

# Client‑side (from another service)
class ApiGatewayService:
    messenger = Messenger()

    @rpc
    async def compute(self, feature: str):
        rpc_service = RpcProxy(self.messenger, service_name="RpcService")
        return await rpc_service.rpc_example(feature)

HTTP 엔드포인트

Kontiki는 @http(...)를 사용하여 HTTP 라우트를 직접 노출할 수 있습니다.
선택적으로 Pydantic 모델로 요청 본문을 검증하고 OpenAPI/Swagger UI를 생성할 수 있습니다.

class SimpleHttpService:
    http_error_handlers = {HttpExampleError: (400, "Example error occurred")}

    @http("/health", "GET", version="v1", response_model=HelloResponse)
    async def health(self, request):
        return HelloResponse(message="ok").model_dump()

주기적 작업

@task(interval=..., immediate=...)를 사용하여 서비스 루프 내에서 주기적인 비동기 작업을 실행합니다.

class TaskService:
    @task(interval=10, immediate=True)
    async def task_example(self):
        ...

구성

서비스는 여러 YAML 파일을 사용하여 반복 가능한 --config 옵션으로 시작할 수 있습니다 (예: 공유 base.yaml와 환경별 env.yaml).

# base.yaml
kontiki:
  amqp:
    url: amqp://guest:guest@localhost/

app:
  cache:
    ttl_seconds: 60

# env.yaml
app:
  cache:
    enabled: true

my_service --config base.yaml --config env.yaml

구성 값은 일반적으로 setup()에서 점(.) 기반 경로를 사용하여 로드됩니다:

from kontiki.configuration import get_parameter
from kontiki.delegate import ServiceDelegate

class MyDelegate(ServiceDelegate):
    async def setup(self):
        self.cache_ttl_seconds = get_parameter(
            self.container.config,
            "app.cache.ttl_seconds",
            60,
        )

Health & Degraded‑Mode Checks

When the Kontiki registry service is running, services can register and expose operational state:

• Heartbeats at a configurable interval
• Degraded mode checks (custom logic)
• Optional event/exception tracking for visibility

from pathlib import Path
from kontiki.decorators import degraded_on

class MyService:
    @degraded_on
    def is_degraded(self) -> bool:
        required_dir = Path("/var/lib/my-service/uploads")
        return not required_dir.exists()

For an operator‑friendly view, the sibling repository kontiki-tui provides a terminal UI to inspect and operate the Kontiki ecosystem.

Source: …

TL;DR

Kontiki는 RabbitMQ 위에서 asyncio‑native, message‑driven 마이크로‑서비스에 적합한 선택이며, 이벤트와 RPC를 위한 일류 프리미티브를 제공하면서 저수준 AMQP 연결, 라이프사이클 관리 및 설정 처리를 추상화합니다.

모든 서비스에 대한 합리적인 파이프라인

이 문서는 간략한 개요를 제공합니다. Kontiki는 핵심 기능을 중심으로 다음과 같은 고급 옵션도 제공합니다:

• 세션
• 브로드캐스트 전송
• 메시지 헤더
• 오류/재시도 동작
• 그 외 다수…

레포지토리에는 실행 가능한 예제가 풍부하게 포함되어 있습니다. RabbitMQ를 로컬에서 실행하고 Makefile 타깃을 통해 대부분의 예제를 실행할 수 있습니다. 예:

make run-amqp
make run-

(원하는 run-* 명령으로 “ 를 교체하세요.)