Temporal Workflows — Python Patterns
Temporal is a durable execution engine. Every workflow step is recorded as an immutable event; if the worker crashes, Temporal replays history to resume exactly where it left off.
When to Activate
- Structuring a new Temporal workflow and its activities
- Debugging non-determinism errors, replay failures, or signal issues
- Adding retries, timeouts, or error handling to activities
- Managing state across workflow turns without losing it on crash
- Implementing human-in-the-loop or long-running multi-step pipelines
- Writing or wiring a Temporal worker
Core Concepts
Event Sourcing / Replay
Temporal records every decision as an immutable event before executing it:
Event 1: WorkflowStarted
Event 2: ActivityScheduled (fetch_data, url)
Event 3: ActivityCompleted (fetch_data, url) → "result"
Event 4: SignalReceived (approve)
Event 5: ActivityScheduled (process_data, ...)
← worker crashes here
On restart, Temporal replays events 1–4. For completed activities it returns the recorded result — no real I/O. Execution resumes at Event 5 for real.
Determinism rule: Workflow code must produce the same decisions on every replay.
- ✅ Call activities for all I/O, random values, current time
- ❌ Never use
random, time.time(), datetime.now(), httpx, or file reads in workflow code - ❌ Never import I/O libraries at module level in workflow files
Workflow vs Activity
| Workflow | Activity | |
|---|
| Purpose | Orchestration, decisions, state | Real I/O — HTTP, DB, LLM, file | |
| I/O allowed | ❌ Must be deterministic | ✅ Unrestricted | |
| Retried by Temporal | Workflow tasks retry on exception | Yes, via RetryPolicy | |
| Current time | workflow.now() only | datetime.now() fine | |
Minimal Workflow
python
# workflow.py
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy
from activities import fetch_data, process_data # imported for type reference only
@workflow.defn
class MyWorkflow:
@workflow.run
async def run(self, url: str) -> str:
# All I/O goes through execute_activity — never call directly
raw = await workflow.execute_activity(
fetch_data,
url,
start_to_close_timeout=timedelta(minutes=2),
retry_policy=RetryPolicy(maximum_attempts=3),
)
result = await workflow.execute_activity(
process_data,
raw,
start_to_close_timeout=timedelta(minutes=5),
)
return result
Activities
Activities are the only place with real I/O. Keep them focused — one network call or DB operation per activity.
python
# activities.py
import httpx
from temporalio import activity
@activity.defn
async def fetch_data(url: str) -> str:
async with httpx.AsyncClient(timeout=30) as client:
response = await client.get(url)
response.raise_for_status() # non-2xx → exception → Temporal retries
return response.text
@activity.defn
async def process_data(raw: str) -> str:
# CPU-bound or DB work here
return raw.strip().upper()
Activity design rules:
- Accept and return JSON-serializable types (str, int, dict, list, Pydantic models)
- Raise exceptions freely — Temporal catches and retries per
RetryPolicy - Make activities idempotent — they may run more than once on retry
- Keep activities short — long-running ones need heartbeats
Worker
python
# run_worker.py
import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from workflow import MyWorkflow
from activities import fetch_data, process_data
async def main():
client = await Client.connect("localhost:7233")
worker = Worker(
client,
task_queue="my-task-queue",
workflows=[MyWorkflow],
activities=[fetch_data, process_data],
)
print("Worker started")
await worker.run()
if __name__ == "__main__":
asyncio.run(main())
Starting a Workflow
python
# client.py
import asyncio
from temporalio.client import Client
from workflow import MyWorkflow
async def main():
client = await Client.connect("localhost:7233")
# Start and wait for result
result = await client.execute_workflow(
MyWorkflow.run,
"https://example.com/data",
id="my-workflow-id-001", # unique per workflow instance
task_queue="my-task-queue",
)
print(result)
# Start without waiting (fire and forget)
handle = await client.start_workflow(
MyWorkflow.run,
"https://example.com/data",
id="my-workflow-id-002",
task_queue="my-task-queue",
)
# Later: result = await handle.result()
asyncio.run(main())
Retries and Timeouts
python
from datetime import timedelta
from temporalio.common import RetryPolicy
# Full retry config
result = await workflow.execute_activity(
fetch_data,
url,
# How long one attempt can run
start_to_close_timeout=timedelta(minutes=2),
# How long all attempts combined can run
schedule_to_close_timeout=timedelta(minutes=10),
retry_policy=RetryPolicy(
initial_interval=timedelta(seconds=1), # first retry after 1s
backoff_coefficient=2.0, # doubles each retry
maximum_interval=timedelta(seconds=30), # cap at 30s
maximum_attempts=5, # 5 total attempts, then raise
non_retryable_error_types=["ValueError"], # don't retry these
),
)
| Timeout | Scope | Use for | |
|---|
start_to_close_timeout | Single attempt | Normal activity duration limit | |
schedule_to_close_timeout | All attempts | Hard deadline across all retries | |
schedule_to_start_timeout | Queue wait time | Detect stuck workers | |
Signals and Queries
python
@workflow.defn
class ApprovalWorkflow:
def __init__(self):
self._approved = False
self._status = "pending"
@workflow.run
async def run(self, item_id: str) -> str:
# Block until approved (or timeout)
await workflow.wait_condition(
lambda: self._approved,
timeout=timedelta(hours=24), # give up after 24h
)
return await workflow.execute_activity(
process_item, item_id,
start_to_close_timeout=timedelta(minutes=5),
)
@workflow.signal
async def approve(self) -> None:
self._approved = True
self._status = "approved"
@workflow.signal
async def reject(self, reason: str) -> None:
self._status = f"rejected: {reason}"
raise Exception(f"Rejected: {reason}")
@workflow.query
def status(self) -> str:
return self._status
# Send a signal from a client
handle = client.get_workflow_handle("approval-workflow-id")
await handle.signal(ApprovalWorkflow.approve)
# Query current state without interrupting
status = await handle.query(ApprovalWorkflow.status)
State Management
Workflows are stateful by design — instance variables persist across signals and replay.
python
@workflow.defn
class BatchWorkflow:
def __init__(self):
self._results: list[str] = []
self._errors: list[str] = []
@workflow.run
async def run(self, urls: list[str]) -> dict:
for url in urls:
try:
result = await workflow.execute_activity(
fetch_data, url,
start_to_close_timeout=timedelta(minutes=2),
retry_policy=RetryPolicy(maximum_attempts=2),
)
self._results.append(result)
except Exception as e:
self._errors.append(f"{url}: {e}")
return {"results": self._results, "errors": self._errors}
For state that must survive worker replacement (long-running workflows across deployments), persist it in an external store (Postgres, Redis) via an activity and reload it on startup.
python
@workflow.run
async def run(self, workflow_id: str) -> str:
# Load persisted state at the start of each run
state = await workflow.execute_activity(
load_state, workflow_id,
start_to_close_timeout=timedelta(seconds=10),
)
# ... do work, update state via save_state activity ...
Long-Running Activities (Heartbeats)
Activities that take longer than start_to_close_timeout must send heartbeats — otherwise Temporal assumes the worker is dead and retries.
python
@activity.defn
async def process_large_file(file_path: str) -> str:
lines = open(file_path).readlines()
results = []
for i, line in enumerate(lines):
result = expensive_operation(line)
results.append(result)
# Heartbeat every 100 lines — keeps the activity alive
if i % 100 == 0:
activity.heartbeat(f"processed {i}/{len(lines)} lines")
return "\n".join(results)
# In workflow — set heartbeat_timeout shorter than start_to_close_timeout
await workflow.execute_activity(
process_large_file,
file_path,
start_to_close_timeout=timedelta(hours=1),
heartbeat_timeout=timedelta(seconds=30), # fail if no heartbeat in 30s
)
Child Workflows
python
from temporalio.workflow import ChildWorkflowHandle
@workflow.defn
class ParentWorkflow:
@workflow.run
async def run(self, items: list[str]) -> list[str]:
# Launch child workflows concurrently
handles: list[ChildWorkflowHandle] = []
for item in items:
handle = await workflow.start_child_workflow(
ChildWorkflow.run,
item,
id=f"child-{item}",
task_queue="my-task-queue",
)
handles.append(handle)
# Wait for all to complete
return list(await asyncio.gather(*[h.result() for h in handles]))
Testing
python
# test_workflow.py
import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from workflow import MyWorkflow
from activities import fetch_data, process_data
@pytest.mark.asyncio
async def test_my_workflow():
async with await WorkflowEnvironment.start_time_skipping() as env:
async with Worker(
env.client,
task_queue="test-queue",
workflows=[MyWorkflow],
activities=[fetch_data, process_data],
):
result = await env.client.execute_workflow(
MyWorkflow.run,
"https://example.com",
id="test-workflow-1",
task_queue="test-queue",
)
assert result == "EXPECTED OUTPUT"
# Mock activities for unit testing the workflow logic
from unittest.mock import AsyncMock
@pytest.mark.asyncio
async def test_workflow_with_mocked_activities():
async with await WorkflowEnvironment.start_time_skipping() as env:
mock_fetch = AsyncMock(return_value="raw data")
mock_process = AsyncMock(return_value="processed")
async with Worker(
env.client,
task_queue="test-queue",
workflows=[MyWorkflow],
activities=[mock_fetch, mock_process],
):
result = await env.client.execute_workflow(
MyWorkflow.run, "https://example.com",
id="test-2", task_queue="test-queue",
)
assert result == "processed"
Common Errors