temporalio
temporalio is the official Python SDK for Temporal, a durable workflow system that records application decisions so long-running processes can survive crashes, retries, deploys, and long waits. Python classes define deterministic workflows, ordinary functions perform side effects as retryable activities, workers poll named task queues, and clients start or interact with executions. The package is only the SDK: a Temporal server and continuously running workers are required for useful work.
Temporal's Python SDK earns its complexity when workflows must be durable, inspectable, and externally interactive. Do not install it for ordinary jobs unless your team is ready to run the service, workers, replay discipline, and deterministic programming model.
Use it if
- A business process must survive worker restarts and continue for hours, days, or longer
- You need durable timers, retries, signals, queries, and child workflows without building a state machine database
- You can operate Temporal Cloud or a self-hosted Temporal service plus dedicated workers
- You need workflow histories that can be replayed to test compatibility before deploying code changes
- A database row and short background job are enough; Temporal adds a server, namespaces, task queues, workers, history, and operational training
- Your workflow code must perform arbitrary network I/O, randomness, threading, or global mutation directly; the README requires deterministic workflow code and moves side effects into activities
- You cannot keep old workflow behavior replay-compatible during deploys; the SDK's replayer throws on nondeterministic changes against stored histories
- You expect `async def` activities to tolerate blocking libraries; the README warns that blocking the event-loop thread can stop other Temporal processing
- You need a simple periodic scheduler or fire-and-forget queue rather than durable orchestration; Celery, Dramatiq, or RQ has far less infrastructure and conceptual weight
Setup reality
`python -m pip install temporalio` installs the SDK on Python 3.10 or later, and the README tells users to update pip because an older installer may miss the correct wheel. That is only the client side. Local development still needs a Temporal server on port 7233, while production needs a reachable service, a namespace, TLS or API credentials, retention and visibility choices, and separately deployed workers polling the exact task queue used by clients. A workflow will start but make no progress when no compatible worker is polling. Workflow files run in a sandbox by default and must be deterministic: no network access, subprocesses, threads, randomness, set iteration, or mutable globals. Third-party imports commonly need `workflow.unsafe.imports_passed_through()`, while real side effects belong in activities. Synchronous activities require an executor sized for worker concurrency; asynchronous activities must never call blocking code. Long activities need both a heartbeat timeout in workflow code and regular `activity.heartbeat()` calls to receive cancellation and resume retries from saved details. Payloads must remain serializable and compatible; the README recommends one dataclass or Pydantic model argument because adding positional parameters breaks callers. Pydantic support requires its optional integration and does not support Pydantic v1. Every workflow-code deploy should replay representative stored histories, and schema or behavior changes may need versioning rather than an ordinary edit.
Patterns
Define a retryable activitydefine-activity
from temporalio import activity
@activity.defn
def charge_customer(order_id: str) -> str:
return payments.charge(order_id)Side effects belong in activities, not workflows; make the external operation idempotent because retries can repeat it.
Define a deterministic workflowdefine-workflow
from datetime import timedelta
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from activities import charge_customer
@workflow.defn
class CheckoutWorkflow:
@workflow.run
async def run(self, order_id: str) -> str:
return await workflow.execute_activity(
charge_customer, order_id, start_to_close_timeout=timedelta(minutes=2)
)Keep workflow code deterministic and import side-effecting third-party modules through the sandbox pass-through block only when appropriate.
Connect a client to a namespaceconnect-client
from temporalio.client import Client
client = await Client.connect(
'temporal.example.com:7233',
namespace='payments',
tls=True,
)Production clusters normally require TLS and credentials; a client connection does not create the namespace or start a worker.
Run workflows and synchronous activitiesrun-worker
from concurrent.futures import ThreadPoolExecutor
from temporalio.worker import Worker
with ThreadPoolExecutor(max_workers=20) as executor:
worker = Worker(
client,
task_queue='payments',
workflows=[CheckoutWorkflow],
activities=[charge_customer],
activity_executor=executor,
)
await worker.run()Synchronous activities require an executor, and its worker count should be at least the configured activity concurrency or the SDK warns.
Start a workflow and wait for its resultexecute-workflow
result = await client.execute_workflow(
CheckoutWorkflow.run,
'order-123',
id='checkout-order-123',
task_queue='payments',
)Workflow IDs and task queues are required; choose an ID reuse policy that matches whether duplicate business operations are valid.
Start now and await the result laterstart-workflow-handle
handle = await client.start_workflow(
CheckoutWorkflow.run,
'order-123',
id='checkout-order-123',
task_queue='payments',
)
# perform other work
result = await handle.result()Use `start_workflow` when you need the handle for signals, queries, cancellation, or delayed result collection.
Change and inspect workflow statesignal-and-query
from temporalio import workflow
@workflow.defn
class ApprovalWorkflow:
def __init__(self):
self.approved = False
@workflow.run
async def run(self) -> bool:
await workflow.wait_condition(lambda: self.approved)
return True
@workflow.signal
def approve(self) -> None:
self.approved = True
@workflow.query
def status(self) -> bool:
return self.approved
await handle.signal(ApprovalWorkflow.approve)
approved = await handle.query(ApprovalWorkflow.status)Signals can mutate state but return no result; queries return data and must not mutate workflow state.
Send a validated update that returns a resultexecute-update
from temporalio import workflow
@workflow.defn
class LimitWorkflow:
def __init__(self) -> None:
self.limit = 0
@workflow.run
async def run(self) -> int:
await workflow.wait_condition(lambda: self.limit > 0)
return self.limit
@workflow.update
def set_limit(self, value: int) -> int:
old = self.limit
self.limit = value
return old
@set_limit.validator
def validate_limit(self, value: int) -> None:
if value < 0:
raise ValueError('limit must be non-negative')
previous = await handle.execute_update(LimitWorkflow.set_limit, 25)Validators must be synchronous and must not mutate state; throwing there rejects the update before history records its acceptance.
Set activity timeout and retry policyconfigure-activity-retry
from datetime import timedelta
from temporalio.common import RetryPolicy
result = await workflow.execute_activity(
send_invoice,
invoice,
start_to_close_timeout=timedelta(seconds=30),
retry_policy=RetryPolicy(
initial_interval=timedelta(seconds=1),
maximum_attempts=5,
),
)Retrying can repeat side effects; use stable idempotency keys and do not treat a timeout as proof that the remote call failed.
Heartbeat a long-running activityheartbeat-activity
from datetime import timedelta
from temporalio import activity, workflow
@activity.defn
def import_rows(rows: list[str]) -> int:
start = next(iter(activity.info().heartbeat_details), 0)
for index, row in enumerate(rows[start:], start=start):
write_row(row)
activity.heartbeat(index + 1)
return len(rows)
@workflow.defn
class ImportWorkflow:
@workflow.run
async def run(self, rows: list[str]) -> int:
return await workflow.execute_activity(
import_rows, rows,
start_to_close_timeout=timedelta(hours=1),
heartbeat_timeout=timedelta(seconds=20),
)Cancellation delivery for a non-local activity depends on both a heartbeat timeout and regular heartbeats.
Test a timer without waitingtest-with-time-skipping
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
async with await WorkflowEnvironment.start_time_skipping() as env:
async with Worker(env.client, task_queue='test', workflows=[ReminderWorkflow]):
result = await env.client.execute_workflow(
ReminderWorkflow.run, id='reminder-1', task_queue='test'
)
assert result == 'sent'Automatic time skipping advances while the test waits on a workflow result; activity code still runs in real time.
Check a deployed history for nondeterminismreplay-workflow-history
from temporalio.client import WorkflowHistory
from temporalio.worker import Replayer
async def verify_history(workflow_id: str, history_json: str) -> None:
history = WorkflowHistory.from_json(workflow_id, history_json)
replayer = Replayer(workflows=[CheckoutWorkflow])
await replayer.replay_workflow(history)Run replay against representative production histories before deploying workflow-code changes; a mismatch raises an error.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| celery | PyPI | You need a familiar distributed task queue and scheduler, not replayed durable workflows |
| dramatiq | PyPI | You want a smaller actor queue with retries on Redis or RabbitMQ |
| prefect | PyPI | Your main use case is observable Python data-flow orchestration |
| rq | PyPI | A straightforward Redis-backed background job queue is sufficient |