Temporal WorkflowTemporal 워크플로우

Workflow: 비즈니스 프로세스 정의 (결정론적 코드)
Activity: 외부 서비스 호출, 부작용 있는 작업
Worker: Workflow + Activity 실행 프로세스
Task Queue: Worker에 작업 배분
Temporal Server: 워크플로우 상태 영속 관리

내결함성 원리:
  1. 모든 Workflow 실행 이력을 이벤트 소싱으로 저장
  2. Worker 재시작 시 이벤트 재실행(replay)으로 상태 복구
  3. Activity 실패 시 자동 재시도 (설정 가능)

import asyncio
from datetime import timedelta
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.worker import Worker

# Activity: 외부 서비스 호출 (실패 가능)
@activity.defn
async def charge_payment(amount: float, user_id: str) -> str:
    # 결제 API 호출 (실패 시 Temporal이 자동 재시도)
    result = await payment_api.charge(user_id, amount)
    return result.transaction_id

@activity.defn
async def send_email(to: str, subject: str) -> None:
    await email_service.send(to, subject)

# Workflow: 비즈니스 프로세스 (결정론적)
@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order: dict) -> str:
        # 결제 처리 (재시도 설정)
        tx_id = await workflow.execute_activity(
            charge_payment,
            args=[order["amount"], order["user_id"]],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=3)
        )

        # 이메일 발송
        await workflow.execute_activity(
            send_email,
            args=[order["email"], f"주문 완료: {tx_id}"],
            start_to_close_timeout=timedelta(seconds=10)
        )

        return tx_id

# 실행
async def main():
    client = await Client.connect("localhost:7233")

    # Worker 시작
    async with Worker(client, task_queue="orders", workflows=[OrderWorkflow], activities=[charge_payment, send_email]):
        # Workflow 실행
        result = await client.execute_workflow(
            OrderWorkflow.run,
            {"amount": 99.99, "user_id": "u123", "email": "user@example.com"},
            id="order-1001",
            task_queue="orders"
        )
        print(f"완료: {result}")