temporal-python

Temporal Python SDK（temporalio）

Python で、耐久性とフォールトトレランスを備えた分散アプリケーションを構築できます。Temporal は、プロセスのクラッシュ、ネットワーク障害、サーバーの停止時にもワークフローの完了を保証します。

インストール

pip install temporalio

Python >= 3.9 が必要です。SDK は、重い依存関係のない単一のパッケージです。

コアコンセプト

• Workflow：アクティビティやほかのワークフローをオーケストレーションする、耐久性のある関数（クラスとして定義します）。決定論的である必要があります。ランダム処理、実際の I/O、システム時刻の使用は禁止です。
• Activity：副作用（API 呼び出し、データベースクエリ、ファイル I/O）を実行する通常の関数（同期または非同期）です。
• Worker：タスクキューをポーリングし、ワークフローやアクティビティを実行するプロセスです。
• Client：Temporal サーバーに接続し、ワークフローの開始、シグナル送信、クエリ、キャンセルを行います。
• Task Queue：ワーカーに処理を振り分ける名前付きのキューです。ワークフローとアクティビティはタスクキューに割り当てられます。

プロジェクト構成

my_temporal_app/
    workflows.py       # Workflow definitions
    activities.py      # Activity definitions
    worker.py          # Worker startup
    client.py          # Client code (start/signal/query workflows)
    models.py          # Shared dataclasses for inputs/outputs
    tests/
        conftest.py    # Test fixtures (WorkflowEnvironment)
        test_workflows.py
        test_activities.py

インポート

# Workflow module -- used inside workflow definitions
from temporalio import workflow

# Activity module -- used inside activity definitions
from temporalio import activity

# Client -- used to connect to Temporal and interact with workflows
from temporalio.client import Client

# Worker -- used to run workflows and activities
from temporalio.worker import Worker

# Common types -- RetryPolicy, SearchAttributes, etc.
from temporalio.common import RetryPolicy

# Exceptions
from temporalio.exceptions import (
    ApplicationError,       # Raise from activities/workflows for business errors
    ActivityError,          # Caught in workflows when an activity fails
    ChildWorkflowError,     # Caught when a child workflow fails
    CancelledError,         # Caught when workflow/activity is cancelled
    FailureError,           # Base class for all failure errors
)

# Testing
from temporalio.testing import WorkflowEnvironment, ActivityEnvironment

ワークフローサンドボックスでのインポート

ワークフローのコードは、非決定的な呼び出しをインターセプトするサンドボックス内で実行されます。ワークフローファイル内でサードパーティ製モジュールや独自のアクティビティ／モデルモジュールをインポートする場合は、パススルーパターンを使用します。

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from my_app.activities import my_activity
    from my_app.models import MyInput, MyOutput

これにより、型参照またはアクティビティ参照としてのみ使用されるインポートにサンドボックスが干渉するのを防げます。

データモデル

Temporal では、アクティビティやワークフローの引数に複数のパラメーターを使うのではなく、単一の dataclass パラメーターを使うことを強く推奨しています。これにより、後方互換性を保ったままフィールドを追加できます。

from dataclasses import dataclass
from typing import Optional

@dataclass
class OrderInput:
    order_id: str
    customer_name: str
    amount: float
    currency: str = "USD"

@dataclass
class OrderResult:
    order_id: str
    status: str
    confirmation_number: Optional[str] = None

アクティビティの定義

非同期アクティビティ（ワーカーのイベントループ内で実行）

from temporalio import activity

@activity.defn
async def process_payment(input: PaymentInput) -> PaymentResult:
    activity.logger.info("Processing payment for order %s", input.order_id)
    result = await payment_gateway.charge(input.amount, input.card_token)
    return PaymentResult(transaction_id=result.id, status="completed")

同期アクティビティ（ワーカー上でスレッドプールエグゼキューターが必要）

import time
from temporalio import activity

@activity.defn
def send_email(input: EmailInput) -> None:
    activity.logger.info("Sending email to %s", input.recipient)
    # Blocking I/O is fine in sync activities
    smtp_client.send(to=input.recipient, subject=input.subject, body=input.body)

ハートビートを送信するアクティビティ（長時間実行するアクティビティ向け）

import time
from temporalio import activity

@activity.defn
def process_large_dataset(input: DatasetInput) -> DatasetResult:
    rows_processed = 0
    for batch in read_batches(input.file_path):
        # Heartbeat to tell the server this activity is still alive.
        # If heartbeat_timeout passes without a heartbeat, the server
        # considers the activity failed and retries it.
        activity.heartbeat(rows_processed)
        process_batch(batch)
        rows_processed += len(batch)
    return DatasetResult(total_rows=rows_processed)

アクティビティが再試行されたときに、最後のハートビートの詳細を取得できます。

@activity.defn
def resumable_activity(input: ProcessInput) -> ProcessResult:
    # On retry, pick up from where we left off
    start_index = 0
    if activity.info().heartbeat_details:
        start_index = activity.info().heartbeat_details[0]

    for i in range(start_index, len(input.items)):
        activity.heartbeat(i)
        process_item(input.items[i])

    return ProcessResult(processed=len(input.items))

クラスメソッドとしてのアクティビティ（依存性注入向け）

from temporalio import activity

class MyActivities:
    def __init__(self, db_client: DatabaseClient) -> None:
        self.db_client = db_client

    @activity.defn
    async def fetch_record(self, input: FetchInput) -> Record:
        return await self.db_client.get(input.record_id)

    @activity.defn
    async def save_record(self, input: SaveInput) -> None:
        await self.db_client.save(input.record)

ワーカーへの登録時にクラスをインスタンス化し、バインド済みメソッドを渡します。

my_activities = MyActivities(db_client)
worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activities.fetch_record, my_activities.save_record],
)

ワークフローから呼び出すときは、workflow.execute_activity_method を使用します。

result = await workflow.execute_activity_method(
    MyActivities.fetch_record,
    FetchInput(record_id="123"),
    start_to_close_timeout=timedelta(seconds=10),
)

カスタムアクティビティ名

@activity.defn(name="custom-activity-name")
async def my_activity(input: MyInput) -> MyOutput:
    ...

動的アクティビティ（未登録のアクティビティタイプをすべて処理）

from typing import Sequence
from temporalio.common import RawValue

@activity.defn(dynamic=True)
async def dynamic_activity(args: Sequence[RawValue]) -> str:
    arg1 = activity.payload_converter().from_payload(args[0].payload, MyDataClass)
    activity_type = activity.info().activity_type
    return f"Dynamic activity {activity_type} received: {arg1}"

非同期アクティビティの完了（アクティビティ関数の外部から完了）

from temporalio import activity

class MyProcessor:
    def __init__(self, client: Client) -> None:
        self.client = client

    @activity.defn
    async def start_processing(self, input: ProcessInput) -> str:
        task_token = activity.info().task_token
        # Hand off to external system, passing the task_token
        await self.external_queue.send(task_token, input)
        # Signal that this activity will be completed externally
        activity.raise_complete_async()

    async def complete_from_external(self, task_token: bytes, result: str) -> None:
        handle = self.client.get_async_activity_handle(task_token=task_token)
        await handle.complete(result)

    async def fail_from_external(self, task_token: bytes, error: Exception) -> None:
        handle = self.client.get_async_activity_handle(task_token=task_token)
        await handle.fail(error)

    async def heartbeat_from_external(self, task_token: bytes) -> None:
        handle = self.client.get_async_activity_handle(task_token=task_token)
        await handle.heartbeat()

アクティビティ情報

アクティビティ内でメタデータにアクセスします。

@activity.defn
async def my_activity(input: MyInput) -> MyOutput:
    info = activity.info()
    info.activity_id          # Unique activity ID
    info.activity_type        # Activity type name
    info.attempt              # Current attempt number (starts at 1)
    info.workflow_id          # Parent workflow ID
    info.workflow_run_id      # Parent workflow run ID
    info.task_token           # Token for async completion
    info.heartbeat_details    # Details from last heartbeat (on retry)
    info.scheduled_time       # When the activity was scheduled
    ...

ワークフローの定義

基本的なワークフロー

from datetime import timedelta
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from my_app.activities import process_order
    from my_app.models import OrderInput, OrderResult

@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, input: OrderInput) -> OrderResult:
        workflow.logger.info("Processing order %s", input.order_id)
        result = await workflow.execute_activity(
            process_order,
            input,
            start_to_close_timeout=timedelta(seconds=30),
        )
        return result

ワークフローのルール（決定性）

@workflow.run 内、およびすべてのシグナル／クエリ／アップデートハンドラー内では、次の操作を行ってはいけません。

• datetime.now()、time.time()、random、uuid4() を使用すること。代わりに workflow.now()、workflow.random()、workflow.uuid4() を使用してください。
• I/O（ネットワーク、ファイル、データベース）を実行すること。代わりにアクティビティを使用してください。
• スレッドを使用すること。asyncio のプリミティブのみを使用してください。
• 非決定的なライブラリ関数を呼び出すこと。

使用できるもの：

• asyncio.sleep()。耐久性のあるタイマーになります。
• asyncio.gather()。アクティビティを並列実行します。
• asyncio.Lock()。ハンドラーの実行が交互に行われてもワークフローの共有状態を保護します。
• workflow.wait_condition()。ブール条件が true になるまで待機します。

シグナル（実行中のワークフローにデータを送信）

from dataclasses import dataclass
from typing import List, Optional

@dataclass
class ApprovalInput:
    approver: str
    approved: bool

@workflow.defn
class ApprovalWorkflow:
    def __init__(self) -> None:
        self._approved: Optional[bool] = None
        self._approver: Optional[str] = None

    @workflow.run
    async def run(self, input: OrderInput) -> str:
        # Wait until we receive the approval signal
        await workflow.wait_condition(lambda: self._approved is not None)
        if self._approved:
            return f"Order {input.order_id} approved by {self._approver}"
        else:
            return f"Order {input.order_id} rejected by {self._approver}"

    @workflow.signal
    def approve(self, input: ApprovalInput) -> None:
        self._approved = input.approved
        self._approver = input.approver

クエリ（ワークフローの状態を変更せずに読み取る）

@workflow.defn
class OrderWorkflow:
    def __init__(self) -> None:
        self._status = "pending"
        self._items_processed = 0

    @workflow.run
    async def run(self, input: OrderInput) -> OrderResult:
        self._status = "processing"
        # ... do work ...
        self._status = "completed"
        return result

    @workflow.query
    def get_status(self) -> str:
        return self._status

    @workflow.query
    def get_progress(self) -> int:
        return self._items_processed

クエリハンドラーは、ワークフローの状態を変更したり、アクティビティを呼び出したりしてはいけません。読み取り専用です。

アップデート（データを送信して応答を取得。任意で検証も可能）

from temporalio.exceptions import ApplicationError

@workflow.defn
class ShoppingCartWorkflow:
    def __init__(self) -> None:
        self._items: dict[str, int] = {}
        self._checked_out = False

    @workflow.run
    async def run(self) -> dict:
        await workflow.wait_condition(lambda: self._checked_out)
        return self._items

    @workflow.update
    async def add_item(self, item_id: str, quantity: int) -> dict:
        self._items[item_id] = self._items.get(item_id, 0) + quantity
        return self._items

    @add_item.validator
    def validate_add_item(self, item_id: str, quantity: int) -> None:
        if self._checked_out:
            raise ApplicationError("Cannot add items after checkout")
        if quantity <= 0:
            raise ApplicationError("Quantity must be positive")

    @workflow.update
    async def checkout(self) -> dict:
        self._checked_out = True
        return self._items

アップデートバリデーターはアップデートハンドラーの前に実行され、例外を発生させてアップデートを拒否できます。バリデーターはワークフローの状態を変更してはいけません。

タイムアウト付きの条件待機

@workflow.run
async def run(self, input: OrderInput) -> str:
    try:
        # Wait up to 24 hours for approval
        await workflow.wait_condition(
            lambda: self._approved is not None,
            timeout=timedelta(hours=24),
        )
        return "approved" if self._approved else "rejected"
    except asyncio.TimeoutError:
        return "timed_out"

すべてのハンドラーの完了を待つ

ワークフローを完了する前に、すべてのシグナルハンドラーとアップデートハンドラーが完了していることを確認します。

@workflow.run
async def run(self, input: MyInput) -> str:
    # ... main workflow logic ...

    # Wait for any in-progress signal/update handlers to complete
    await workflow.wait_condition(workflow.all_handlers_finished)
    return "done"

タイマー（耐久性のあるスリープ）

@workflow.run
async def run(self, input: ReminderInput) -> None:
    # This sleep is durable -- if the worker crashes, the timer
    # continues on the server and fires when it expires
    await asyncio.sleep(3600)  # Wait 1 hour
    await workflow.execute_activity(
        send_reminder,
        input,
        start_to_close_timeout=timedelta(seconds=10),
    )

アクティビティの並列実行

@workflow.run
async def run(self, users: list[str]) -> list[str]:
    # Run multiple activities concurrently using asyncio.gather
    results = await asyncio.gather(
        *[
            workflow.execute_activity(
                greet_user,
                user,
                start_to_close_timeout=timedelta(seconds=5),
            )
            for user in users
        ]
    )
    return list(results)

子ワークフロー

@workflow.defn
class ParentWorkflow:
    @workflow.run
    async def run(self, input: ParentInput) -> str:
        # Execute and wait for child workflow result
        child_result = await workflow.execute_child_workflow(
            ChildWorkflow.run,
            ChildInput(data=input.data),
            id=f"child-{input.parent_id}",
        )
        return f"Parent got: {child_result}"

@workflow.defn
class ChildWorkflow:
    @workflow.run
    async def run(self, input: ChildInput) -> str:
        return await workflow.execute_activity(
            process_data,
            input,
            start_to_close_timeout=timedelta(seconds=30),
        )

結果を待たずに子ワークフローを開始します。

child_handle = await workflow.start_child_workflow(
    ChildWorkflow.run,
    ChildInput(data=input.data),
    id=f"child-{input.parent_id}",
)
# Can signal the child
await child_handle.signal(ChildWorkflow.some_signal, signal_data)
# Wait for result later
result = await child_handle

親のクローズポリシーによって、親の完了時に子ワークフローがどうなるかが決まります。

from temporalio.workflow import ParentClosePolicy

child_result = await workflow.execute_child_workflow(
    ChildWorkflow.run,
    input,
    id="child-id",
    parent_close_policy=ParentClosePolicy.ABANDON,  # Child keeps running
    # Other options: TERMINATE (default), REQUEST_CANCEL
)

Continue-As-New（長時間実行／エンティティワークフロー向け）

ワークフローにはイベント履歴が蓄積されます。無期限または非常に長期間実行されるワークフローでは、論理状態を維持したまま履歴をリセットするために、continue-as-new を使用します。

@workflow.defn
class EntityWorkflow:
    @workflow.run
    async def run(self, state: EntityState) -> EntityResult:
        while not state.is_done:
            await workflow.wait_condition(
                lambda: state.has_pending_work or state.is_done,
                timeout=timedelta(minutes=10),
            )
            if state.has_pending_work:
                await self.process_work(state)

            # Check if Temporal suggests continuing as new
            if workflow.info().is_continue_as_new_suggested():
                # Finish any in-progress handlers first
                await workflow.wait_condition(workflow.all_handlers_finished)
                workflow.continue_as_new(state)

        return EntityResult(...)

workflow.continue_as_new() は内部例外を発生させて現在のワークフロー実行を停止し、指定された引数を渡して、同じワークフロー ID で新しい実行を開始します。

ワークフローの初期化（実行前に入力にアクセス）

@dataclass
class MyWorkflowInput:
    name: str

@workflow.defn
class MyWorkflow:
    @workflow.init
    def __init__(self, input: MyWorkflowInput) -> None:
        # Set up state before run() is called.
        # Same input is passed to both __init__ and run.
        self.greeting_prefix = f"Hello, {input.name}"

    @workflow.run
    async def run(self, input: MyWorkflowInput) -> str:
        return self.greeting_prefix

外部ワークフローへのシグナル送信

@workflow.defn
class WorkflowB:
    @workflow.run
    async def run(self) -> None:
        # Get a handle to another running workflow and signal it
        handle = workflow.get_external_workflow_handle_for(
            WorkflowA.run, "workflow-a-id"
        )
        await handle.signal(WorkflowA.some_signal, "signal data")

動的なシグナルハンドラーとクエリハンドラー

from typing import Sequence
from temporalio.common import RawValue

@workflow.defn
class FlexibleWorkflow:
    @workflow.signal(dynamic=True)
    async def dynamic_signal(self, name: str, args: Sequence[RawValue]) -> None:
        # Called for any signal that does not match a named handler
        payload = workflow.payload_converter().from_payload(args[0].payload, str)
        workflow.logger.info(f"Received dynamic signal '{name}': {payload}")

    @workflow.query(dynamic=True)
    def dynamic_query(self, name: str, args: Sequence[RawValue]) -> str:
        return f"Dynamic query '{name}' handled"

動的ワークフロー（キャッチオール）

@workflow.defn(dynamic=True)
class DynamicWorkflow:
    @workflow.run
    async def run(self, args: Sequence[RawValue]) -> str:
        name = workflow.payload_converter().from_payload(args[0].payload, str)
        return f"Hello, {name}!"

ハンドラーの安全性を確保する AsyncIO ロック

複数のシグナルハンドラーまたは更新ハンドラーが同じ状態を変更する可能性があり、await ポイントを含む場合は、インターリーブを防ぐために asyncio.Lock を使用します。

@workflow.defn
class SafeWorkflow:
    def __init__(self) -> None:
        self.lock = asyncio.Lock()
        self.balance = 0

    @workflow.update
    async def transfer(self, amount: float) -> float:
        async with self.lock:
            # Check balance
            if self.balance < amount:
                raise ApplicationError("Insufficient funds")
            # This await yields control, but the lock prevents interleaving
            await workflow.execute_activity(
                record_transfer,
                amount,
                start_to_close_timeout=timedelta(seconds=10),
            )
            self.balance -= amount
            return self.balance

ワークフローのユーティリティ関数

# Inside a workflow:
workflow.info()                    # WorkflowInfo with id, run_id, task_queue, etc.
workflow.now()                     # Current time (deterministic)
workflow.random()                  # Random instance (deterministic, seeded)
workflow.uuid4()                   # Deterministic UUID generation
workflow.logger                    # Logger that includes workflow context
workflow.memo()                    # Access workflow memo
workflow.all_handlers_finished()   # True when all handlers are done
workflow.info().is_continue_as_new_suggested()  # Server recommends continue-as-new
workflow.info().get_current_history_length()     # Current event history size

ワーカーのセットアップ

基本的なワーカー

import asyncio
from concurrent.futures import ThreadPoolExecutor
from temporalio.client import Client
from temporalio.worker import Worker

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

    worker = Worker(
        client,
        task_queue="my-task-queue",
        workflows=[MyWorkflow],
        activities=[activity_a, activity_b],
        # Required for synchronous (non-async) activities:
        activity_executor=ThreadPoolExecutor(10),
    )
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

コンテキストマネージャーとしてのワーカー

async with Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activity],
    activity_executor=ThreadPoolExecutor(10),
) as worker:
    # Worker is running, do other things...
    await some_other_task()
# Worker shuts down gracefully when exiting the context

アクティビティメソッドクラスを使用するワーカー

db_client = DatabaseClient(connection_string)
my_activities = MyActivities(db_client)

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activities.fetch_record, my_activities.save_record],
)

ワーカーの主なパラメーター

Worker(
    client,
    task_queue="my-task-queue",
    workflows=[...],
    activities=[...],
    activity_executor=ThreadPoolExecutor(10),   # For sync activities
    max_concurrent_activities=100,               # Limit concurrent activities
    max_concurrent_workflow_tasks=100,           # Limit concurrent workflow tasks
    max_cached_workflows=100,                    # Workflow cache size
)

クライアントの使用方法

Temporal への接続

from temporalio.client import Client

# Local development
client = await Client.connect("localhost:7233")

# With namespace
client = await Client.connect("localhost:7233", namespace="my-namespace")

# Temporal Cloud with mTLS
client = await Client.connect(
    "my-namespace.a1b2c.tmprl.cloud:7233",
    namespace="my-namespace.a1b2c",
    tls=TLSConfig(
        client_cert=Path("client.pem").read_bytes(),
        client_private_key=Path("client.key").read_bytes(),
    ),
)

ワークフローの開始

# Start and wait for result
result = await client.execute_workflow(
    MyWorkflow.run,
    MyInput(data="hello"),
    id="my-workflow-id",
    task_queue="my-task-queue",
)

# Start without waiting (returns handle)
handle = await client.start_workflow(
    MyWorkflow.run,
    MyInput(data="hello"),
    id="my-workflow-id",
    task_queue="my-task-queue",
)
# Get result later
result = await handle.result()

ワークフローハンドルの操作

# Get a handle to an existing workflow
handle = client.get_workflow_handle("my-workflow-id")

# Signal the workflow
await handle.signal(MyWorkflow.my_signal, SignalInput(data="value"))

# Query the workflow
status = await handle.query(MyWorkflow.get_status)

# Execute an update (send + wait for result)
update_result = await handle.execute_update(
    MyWorkflow.my_update, UpdateInput(data="value")
)

# Cancel the workflow
await handle.cancel()

# Terminate the workflow (immediate, no cleanup)
await handle.terminate("reason for termination")

# Describe the workflow (get execution info)
description = await handle.describe()
print(description.status)  # WorkflowExecutionStatus.RUNNING, COMPLETED, etc.

# Get the result (will raise WorkflowFailureError if workflow failed)
from temporalio.client import WorkflowFailureError
try:
    result = await handle.result()
except WorkflowFailureError as e:
    print(f"Workflow failed: {e}")

シグナルと同時に開始

ワークフローを開始し、シグナルをアトミックに送信します。ワークフローがすでに存在する場合は、シグナルのみを送信します。

handle = await client.start_workflow(
    MyWorkflow.run,
    MyInput(data="hello"),
    id="my-workflow-id",
    task_queue="my-task-queue",
    start_signal="my_signal",
    start_signal_args=[SignalInput(data="initial")],
)

遅延して開始

handle = await client.start_workflow(
    MyWorkflow.run,
    MyInput(data="hello"),
    id="my-workflow-id",
    task_queue="my-task-queue",
    start_delay=timedelta(hours=1),
)

ワークフローの再試行ポリシー（クライアントから指定）

from temporalio.common import RetryPolicy

handle = await client.start_workflow(
    MyWorkflow.run,
    input_data,
    id="my-workflow-id",
    task_queue="my-task-queue",
    retry_policy=RetryPolicy(
        initial_interval=timedelta(seconds=1),
        backoff_coefficient=2.0,
        maximum_interval=timedelta(minutes=5),
        maximum_attempts=10,
    ),
)

ワークフローのタイムアウト

handle = await client.start_workflow(
    MyWorkflow.run,
    input_data,
    id="my-workflow-id",
    task_queue="my-task-queue",
    execution_timeout=timedelta(hours=24),    # Total time across all runs
    run_timeout=timedelta(hours=1),           # Single run (before continue-as-new)
    task_timeout=timedelta(seconds=30),       # Single workflow task
)

ワークフローの一覧表示

async for workflow_exec in client.list_workflows(
    "WorkflowType = 'MyWorkflow' AND ExecutionStatus = 'Running'"
):
    print(f"Workflow: {workflow_exec.id}, Status: {workflow_exec.status}")

スケジュール

from temporalio.client import (
    Schedule,
    ScheduleActionStartWorkflow,
    ScheduleSpec,
    ScheduleIntervalSpec,
    ScheduleState,
)

await client.create_schedule(
    "my-schedule-id",
    Schedule(
        action=ScheduleActionStartWorkflow(
            MyWorkflow.run,
            MyInput(data="scheduled"),
            id="scheduled-workflow",
            task_queue="my-task-queue",
        ),
        spec=ScheduleSpec(
            intervals=[ScheduleIntervalSpec(every=timedelta(minutes=5))]
        ),
        state=ScheduleState(note="Runs every 5 minutes"),
    ),
)

# Manage schedule
handle = client.get_schedule_handle("my-schedule-id")
await handle.pause(note="Pausing for maintenance")
await handle.trigger()  # Run immediately
await handle.delete()
desc = await handle.describe()

再試行ポリシー

RetryPolicy は、アクティビティ（およびワークフロー）の失敗時の再試行方法を制御します。

from temporalio.common import RetryPolicy
from datetime import timedelta

retry_policy = RetryPolicy(
    initial_interval=timedelta(seconds=1),     # First retry delay
    backoff_coefficient=2.0,                    # Multiplier for subsequent delays
    maximum_interval=timedelta(seconds=60),     # Cap on retry delay
    maximum_attempts=5,                         # Max total attempts (0 = unlimited)
    non_retryable_error_types=["InvalidInput"], # Error types that skip retries
)

再試行ポリシーを指定したアクティビティ

result = await workflow.execute_activity(
    call_external_api,
    input_data,
    start_to_close_timeout=timedelta(seconds=30),
    retry_policy=RetryPolicy(
        initial_interval=timedelta(seconds=1),
        backoff_coefficient=2.0,
        maximum_interval=timedelta(seconds=30),
        maximum_attempts=5,
    ),
)

アクティビティのタイムアウト

result = await workflow.execute_activity(
    my_activity,
    input_data,
    # Time from activity scheduled to completed (including retries + queue time)
    schedule_to_close_timeout=timedelta(minutes=5),
    # Time from activity picked up by worker to completed (single attempt)
    start_to_close_timeout=timedelta(seconds=30),
    # Time waiting in queue before a worker picks it up
    schedule_to_start_timeout=timedelta(seconds=60),
    # Max time between heartbeats before activity is considered failed
    heartbeat_timeout=timedelta(seconds=10),
)

schedule_to_close_timeout または start_to_close_timeout の少なくとも一方を設定する必要があります。

次回の再試行間隔をカスタマイズ

アクティビティ内から再試行間隔を上書きします。

from temporalio.exceptions import ApplicationError

@activity.defn
async def my_activity(input: MyInput) -> MyOutput:
    try:
        return await do_work(input)
    except Exception as e:
        attempt = activity.info().attempt
        raise ApplicationError(
            f"Failed on attempt {attempt}",
            next_retry_delay=timedelta(seconds=3 * attempt),
        ) from e

エラー処理

ApplicationError（ビジネスロジックのエラー）

from temporalio.exceptions import ApplicationError

# Retryable error (default)
raise ApplicationError("Temporary failure, will retry")

# Non-retryable error (will not be retried regardless of retry policy)
raise ApplicationError(
    "Invalid input: card number is malformed",
    type="InvalidInput",
    non_retryable=True,
)

# Error with custom type (can be matched by non_retryable_error_types)
raise ApplicationError(
    "Insufficient funds",
    type="InsufficientFunds",
)

ワークフローでのアクティビティエラーの捕捉

from temporalio.exceptions import ActivityError, ApplicationError

@workflow.run
async def run(self, input: OrderInput) -> OrderResult:
    try:
        result = await workflow.execute_activity(
            charge_payment,
            input,
            start_to_close_timeout=timedelta(seconds=30),
        )
    except ActivityError as e:
        # e.cause contains the original error (usually ApplicationError)
        workflow.logger.error(f"Payment failed: {e.cause}")
        # Handle the failure...
        raise ApplicationError(
            f"Order failed: {e.cause}",
            type="OrderFailed",
        )

子ワークフローのエラーの捕捉

from temporalio.exceptions import ChildWorkflowError

try:
    result = await workflow.execute_child_workflow(
        ChildWorkflow.run,
        input_data,
        id="child-id",
    )
except ChildWorkflowError as e:
    workflow.logger.error(f"Child workflow failed: {e.cause}")

冪等性キー

ワークフローとアクティビティのメタデータを使用して、外部呼び出し用の冪等性キーを作成します。

@activity.defn
async def charge_payment(input: PaymentInput) -> str:
    info = activity.info()
    idempotency_key = f"{info.workflow_run_id}-{info.activity_id}"
    return await payment_service.charge(
        amount=input.amount,
        idempotency_key=idempotency_key,
    )

キャンセル

クライアントからワークフローをキャンセル

handle = client.get_workflow_handle("my-workflow-id")
await handle.cancel()

アクティビティでのキャンセル処理

キャンセルを受け取るには、アクティビティでハートビートを送信する必要があります。キャンセルされると、asyncio.CancelledError（非同期）または temporalio.exceptions.CancelledError（同期）が発生します。

@activity.defn
async def cancellable_activity(input: MyInput) -> None:
    try:
        while True:
            activity.heartbeat()
            await asyncio.sleep(1)
            await do_some_work()
    except asyncio.CancelledError:
        # Perform cleanup
        await cleanup()
        raise  # Must re-raise for proper cancellation

@activity.defn
def sync_cancellable_activity(input: MyInput) -> None:
    try:
        while True:
            activity.heartbeat()
            time.sleep(1)
    except CancelledError:
        cleanup()
        raise  # Must re-raise

クリーンアップを伴うワークフローでのキャンセル処理

@workflow.defn
class WorkflowWithCleanup:
    @workflow.run
    async def run(self) -> None:
        try:
            await workflow.execute_activity(
                long_running_activity,
                start_to_close_timeout=timedelta(hours=1),
                heartbeat_timeout=timedelta(seconds=10),
            )
        finally:
            # Cleanup runs even on cancellation
            await workflow.execute_activity(
                cleanup_activity,
                start_to_close_timeout=timedelta(seconds=30),
            )

ワークフローからアクティビティをキャンセル

@workflow.run
async def run(self) -> None:
    activity_handle = workflow.start_activity(
        long_running_activity,
        start_to_close_timeout=timedelta(hours=1),
        heartbeat_timeout=timedelta(seconds=10),
    )
    # Wait, then cancel
    await asyncio.sleep(60)
    activity_handle.cancel()

バージョン管理（パッチ適用）

既存の実行が継続している間にワークフローのロジックを安全に変更するには、パッチを適用します。

ステップ 1：パッチを追加する

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> None:
        if workflow.patched("my-change-id"):
            # New code path
            result = await workflow.execute_activity(
                new_activity,
                schedule_to_close_timeout=timedelta(minutes=5),
            )
        else:
            # Old code path (for in-flight workflows)
            result = await workflow.execute_activity(
                old_activity,
                schedule_to_close_timeout=timedelta(minutes=5),
            )

ステップ 2：パッチを非推奨にする（古い実行がすべて完了した後）

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> None:
        workflow.deprecate_patch("my-change-id")
        result = await workflow.execute_activity(
            new_activity,
            schedule_to_close_timeout=timedelta(minutes=5),
        )

ステップ 3：非推奨化を解除する（さらに一巡した後）

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> None:
        result = await workflow.execute_activity(
            new_activity,
            schedule_to_close_timeout=timedelta(minutes=5),
        )

テスト

pytest を使用したテストのセットアップ（conftest.py）

import pytest
import pytest_asyncio
from temporalio.client import Client
from temporalio.testing import WorkflowEnvironment

@pytest_asyncio.fixture(scope="session")
async def env():
    # start_local() starts a real lightweight Temporal server
    # start_time_skipping() uses a test server that auto-advances time
    env = await WorkflowEnvironment.start_local()
    yield env
    await env.shutdown()

@pytest_asyncio.fixture
async def client(env: WorkflowEnvironment) -> Client:
    return env.client

ワークフローのエンドツーエンドテスト

import uuid
from concurrent.futures import ThreadPoolExecutor
from temporalio.worker import Worker

async def test_order_workflow(client: Client):
    task_queue = str(uuid.uuid4())

    async with Worker(
        client,
        task_queue=task_queue,
        workflows=[OrderWorkflow],
        activities=[process_order, send_confirmation],
        activity_executor=ThreadPoolExecutor(5),
    ):
        result = await client.execute_workflow(
            OrderWorkflow.run,
            OrderInput(order_id="123", customer_name="Alice", amount=99.99),
            id=str(uuid.uuid4()),
            task_queue=task_queue,
        )
        assert result.status == "completed"
        assert result.confirmation_number is not None

モックアクティビティを使用したテスト

@activity.defn(name="...") を使用して、実際のアクティビティと同じ名前のモックアクティビティを定義します。

from temporalio import activity

@activity.defn(name="process_order")
async def mock_process_order(input: OrderInput) -> OrderResult:
    return OrderResult(
        order_id=input.order_id,
        status="completed",
        confirmation_number="MOCK-123",
    )

async def test_workflow_with_mock(client: Client):
    task_queue = str(uuid.uuid4())
    async with Worker(
        client,
        task_queue=task_queue,
        workflows=[OrderWorkflow],
        activities=[mock_process_order],  # Use mock instead of real activity
    ):
        result = await client.execute_workflow(
            OrderWorkflow.run,
            OrderInput(order_id="123", customer_name="Alice", amount=99.99),
            id=str(uuid.uuid4()),
            task_queue=task_queue,
        )
        assert result.confirmation_number == "MOCK-123"

シグナルのテスト

from temporalio.client import WorkflowExecutionStatus

async def test_signal_workflow(client: Client):
    task_queue = str(uuid.uuid4())
    async with Worker(client, task_queue=task_queue, workflows=[ApprovalWorkflow]):
        handle = await client.start_workflow(
            ApprovalWorkflow.run,
            OrderInput(order_id="123", customer_name="Alice", amount=99.99),
            id=str(uuid.uuid4()),
            task_queue=task_queue,
        )

        # Verify running
        assert WorkflowExecutionStatus.RUNNING == (await handle.describe()).status

        # Send signal
        await handle.signal(
            ApprovalWorkflow.approve,
            ApprovalInput(approver="Bob", approved=True),
        )

        # Wait for result
        result = await handle.result()
        assert "approved" in result

アクティビティを単独でテスト

from temporalio.testing import ActivityEnvironment

async def test_activity_with_heartbeat():
    env = ActivityEnvironment()

    # Capture heartbeats
    heartbeats = []
    env.on_heartbeat = lambda *args: heartbeats.append(args[0])

    result = await env.run(process_large_dataset, DatasetInput(file_path="test.csv"))
    assert result.total_rows > 0
    assert len(heartbeats) > 0

時間スキップを使用したテスト

タイマー（asyncio.sleep）を使用するワークフローでは、時間スキップ環境を使用します。

@pytest_asyncio.fixture(scope="session")
async def env():
    env = await WorkflowEnvironment.start_time_skipping()
    yield env
    await env.shutdown()

async def test_reminder_workflow(client: Client):
    task_queue = str(uuid.uuid4())
    async with Worker(client, task_queue=task_queue, workflows=[ReminderWorkflow]):
        # This workflow sleeps for 24 hours, but time-skipping makes it instant
        result = await client.execute_workflow(
            ReminderWorkflow.run,
            ReminderInput(message="test"),
            id=str(uuid.uuid4()),
            task_queue=task_queue,
        )
        assert result == "reminder_sent"

リプレイテスト（非決定的な変更を検出）

from temporalio.worker import Replayer
from temporalio.workflow import WorkflowHistory

async def test_replay_from_history():
    replayer = Replayer(workflows=[MyWorkflow])
    # Replay from a JSON history file
    with open("workflow_history.json") as f:
        await replayer.replay_workflow(
            WorkflowHistory.from_json(f.read())
        )

async def test_replay_from_server(client: Client):
    workflows = client.list_workflows("WorkflowType = 'MyWorkflow'")
    histories = workflows.map_histories()
    replayer = Replayer(workflows=[MyWorkflow])
    await replayer.replay_workflows(histories, raise_on_replay_failure=False)

主なパターン

Saga／補償パターン

一連のアクティビティを実行し、いずれかのステップが失敗した場合は、完了済みのステップをロールバックします。

from dataclasses import dataclass, field
from typing import Any, Callable, List
from temporalio.exceptions import ActivityError, ApplicationError

@dataclass
class Compensation:
    activity: Any  # Activity function reference
    input: Any     # Input to pass to compensation activity

@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, input: OrderInput) -> OrderResult:
        compensations: List[Compensation] = []

        try:
            # Step 1: Reserve inventory
            await workflow.execute_activity(
                reserve_inventory,
                input,
                start_to_close_timeout=timedelta(seconds=30),
            )
            compensations.append(Compensation(
                activity=release_inventory,
                input=input,
            ))

            # Step 2: Charge payment
            payment = await workflow.execute_activity(
                charge_payment,
                PaymentInput(order_id=input.order_id, amount=input.amount),
                start_to_close_timeout=timedelta(seconds=30),
            )
            compensations.append(Compensation(
                activity=refund_payment,
                input=PaymentInput(order_id=input.order_id, amount=input.amount),
            ))

            # Step 3: Create shipment
            shipment = await workflow.execute_activity(
                create_shipment,
                ShipmentInput(order_id=input.order_id, address=input.address),
                start_to_close_timeout=timedelta(seconds=30),
            )

            return OrderResult(
                order_id=input.order_id,
                status="completed",
                confirmation_number=shipment.tracking_id,
            )

        except ActivityError as e:
            workflow.logger.error(f"Saga step failed: {e.cause}, compensating...")
            # Run compensations in reverse order
            for compensation in reversed(compensations):
                try:
                    await workflow.execute_activity(
                        compensation.activity,
                        compensation.input,
                        start_to_close_timeout=timedelta(seconds=30),
                    )
                except ActivityError as comp_err:
                    workflow.logger.error(
                        f"Compensation failed: {comp_err.cause}"
                    )
            raise ApplicationError(
                f"Order saga failed: {e.cause}",
                type="SagaFailed",
            )

長時間実行エンティティワークフロー

シグナルと更新を処理し、履歴の増大を防ぐために定期的に continue-as-new を実行する、無期限に実行されるワークフローです。

@dataclass
class EntityState:
    items: dict = field(default_factory=dict)
    is_shutdown: bool = False

@workflow.defn
class EntityWorkflow:
    @workflow.init
    def __init__(self, state: EntityState) -> None:
        self.state = state
        self.lock = asyncio.Lock()

    @workflow.run
    async def run(self, state: EntityState) -> EntityState:
        while not self.state.is_shutdown:
            try:
                await workflow.wait_condition(
                    lambda: self.state.is_shutdown or self.should_continue_as_new(),
                    timeout=timedelta(minutes=10),
                )
            except asyncio.TimeoutError:
                # Periodic maintenance
                await self.perform_maintenance()

            if self.should_continue_as_new():
                await workflow.wait_condition(workflow.all_handlers_finished)
                workflow.continue_as_new(self.state)

        await workflow.wait_condition(workflow.all_handlers_finished)
        return self.state

    @workflow.signal
    def shutdown(self) -> None:
        self.state.is_shutdown = True

    @workflow.update
    async def add_item(self, key: str, value: str) -> dict:
        async with self.lock:
            self.state.items[key] = value
            await workflow.execute_activity(
                persist_item,
                PersistInput(key=key, value=value),
                start_to_close_timeout=timedelta(seconds=10),
            )
            return self.state.items

    @workflow.query
    def get_items(self) -> dict:
        return self.state.items

    def should_continue_as_new(self) -> bool:
        return workflow.info().is_continue_as_new_suggested()

    async def perform_maintenance(self) -> None:
        async with self.lock:
            await workflow.execute_activity(
                run_maintenance,
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=1),
            )

ポーリングパターン（外部条件を待機）

@workflow.defn
class PollingWorkflow:
    @workflow.run
    async def run(self, input: PollInput) -> PollResult:
        for attempt in range(input.max_attempts):
            result = await workflow.execute_activity(
                check_status,
                input,
                start_to_close_timeout=timedelta(seconds=10),
            )
            if result.is_ready:
                return result
            # Durable sleep between polls
            await asyncio.sleep(input.poll_interval_seconds)

        raise ApplicationError("Polling timed out", type="PollTimeout")

ファンアウト／ファンイン

@workflow.run
async def run(self, items: list[ItemInput]) -> list[ItemResult]:
    # Fan out: start all activities concurrently
    results = await asyncio.gather(
        *[
            workflow.execute_activity(
                process_item,
                item,
                start_to_close_timeout=timedelta(seconds=30),
            )
            for item in items
        ]
    )
    # Fan in: aggregate results
    return list(results)

承認／人間参加型ループ

@workflow.defn
class ApprovalWorkflow:
    def __init__(self) -> None:
        self._decision: Optional[bool] = None

    @workflow.run
    async def run(self, input: ApprovalInput) -> str:
        # Notify reviewer
        await workflow.execute_activity(
            send_approval_request,
            input,
            start_to_close_timeout=timedelta(seconds=30),
        )

        # Wait for human decision with timeout
        try:
            await workflow.wait_condition(
                lambda: self._decision is not None,
                timeout=timedelta(hours=72),
            )
        except asyncio.TimeoutError:
            return "timed_out"

        if self._decision:
            await workflow.execute_activity(
                execute_approved_action,
                input,
                start_to_close_timeout=timedelta(seconds=30),
            )
            return "approved"
        else:
            return "rejected"

    @workflow.signal
    def decide(self, approved: bool) -> None:
        self._decision = approved

完全な動作例

すべての要素をまとめて示す、単一ファイルの完全な例です。

import asyncio
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import timedelta

from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.common import RetryPolicy
from temporalio.worker import Worker


# --- Models ---

@dataclass
class GreetingInput:
    name: str
    greeting: str = "Hello"


@dataclass
class GreetingResult:
    message: str
    attempts: int


# --- Activities ---

@activity.defn
def compose_greeting(input: GreetingInput) -> GreetingResult:
    activity.logger.info("Composing greeting for %s", input.name)
    return GreetingResult(
        message=f"{input.greeting}, {input.name}!",
        attempts=activity.info().attempt,
    )


# --- Workflows ---

@workflow.defn
class GreetingWorkflow:
    def __init__(self) -> None:
        self._greeting_count = 0
        self._last_greeting = ""

    @workflow.run
    async def run(self, input: GreetingInput) -> str:
        result = await workflow.execute_activity(
            compose_greeting,
            input,
            start_to_close_timeout=timedelta(seconds=10),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )
        self._greeting_count += 1
        self._last_greeting = result.message
        return result.message

    @workflow.query
    def get_count(self) -> int:
        return self._greeting_count

    @workflow.query
    def get_last_greeting(self) -> str:
        return self._last_greeting


# --- Main ---

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

    async with Worker(
        client,
        task_queue="greeting-task-queue",
        workflows=[GreetingWorkflow],
        activities=[compose_greeting],
        activity_executor=ThreadPoolExecutor(5),
    ):
        result = await client.execute_workflow(
            GreetingWorkflow.run,
            GreetingInput(name="World"),
            id="greeting-workflow-1",
            task_queue="greeting-task-queue",
        )
        print(f"Result: {result}")


if __name__ == "__main__":
    asyncio.run(main())