temporalio__skill-temporal-developer/references/python/python.md

Temporal Python SDK Reference

Overview

The Temporal Python SDK (temporalio) provides a fully async, type-safe approach to building durable workflows. Python 3.9+ required. Workflows run in a sandbox by default for determinism protection.

Quick Start

activities/greet.py - Activity definitions (separate file for performance):

from temporalio import activity

@activity.defn
async def greet(name: str) -> str:
    return f"Hello, {name}!"

workflows/greeting.py - Workflow definition (import activities through sandbox):

from datetime import timedelta
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities.greet import greet

@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            greet, name, schedule_to_close_timeout=timedelta(seconds=30)
        )

worker.py - Worker setup (imports both):

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker

from activities.greet import greet
from workflows.greeting import GreetingWorkflow

async def main():
    client = await Client.connect("localhost:7233")
    async with Worker(client, task_queue="greeting-queue",
                      workflows=[GreetingWorkflow], activities=[greet]):
        result = await client.execute_workflow(
            GreetingWorkflow.run, "World",
            id="greeting-workflow", task_queue="greeting-queue"
        )
        print(result)

asyncio.run(main())

Key Concepts

Workflow Definition

• Use @workflow.defn decorator on class
• Use @workflow.run on the entry point method
• Must be async (async def)
• Use @workflow.signal, @workflow.query, @workflow.update for handlers

Activity Definition

• Use @activity.defn decorator
• Can be sync or async functions
• Default to sync activities - safer and easier to debug
• Sync activities need activity_executor (ThreadPoolExecutor)
• Async activities require async-safe libraries throughout (e.g., aiohttp not requests)

See sync-vs-async.md for detailed guidance on choosing between sync and async.

Worker Setup

• Connect client, create Worker with workflows and activities
• Use async with Worker(...) context manager
• Activities can specify custom executor

Why Determinism Matters: History Replay

Temporal achieves durability through history replay. When a Worker restarts or recovers from a failure, it re-executes the Workflow code from the beginning, but instead of re-running Activities, it uses the results stored in the Event History.

This is why Workflow code must be deterministic:

• During replay, the SDK compares Commands generated by your code against the Events in history
• If the sequence differs (non-determinism), the Worker cannot restore state
• Non-determinism causes NondeterminismError and blocks Workflow progress

The Python SDK's sandbox provides automatic protection against many common non-deterministic operations, but understanding replay helps you write correct code.

See determinism.md for detailed rules and safe alternatives.

Determinism Rules Summary

Safe alternatives:

• workflow.now() instead of datetime.now()
• workflow.random() instead of random
• asyncio.sleep() works (sandbox handles it)
• workflow.uuid4() for UUIDs
• workflow.wait() instead of asyncio.wait() (deterministic ordering)
• workflow.as_completed() instead of asyncio.as_completed() (deterministic ordering)

Pass-through for libraries:

with workflow.unsafe.imports_passed_through():
    import pydantic

Replay-Aware Logging

Use workflow.logger inside Workflows for replay-safe logging that avoids duplicate log messages during replay:

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        workflow.logger.info("Starting workflow")  # Only logs on initial execution
        result = await workflow.execute_activity(...)
        workflow.logger.info(f"Activity result: {result}")
        return result

Use activity.logger in activities for context-aware logging with automatic correlation:

@activity.defn
async def my_activity() -> str:
    activity.logger.info("Processing activity")  # Includes activity context
    return "done"

File Organization Best Practice

Keep Workflow definitions in separate files from Activity definitions. The Python SDK sandbox reloads Workflow definition files on every execution for determinism protection. Minimizing file contents improves Worker performance.

my_temporal_app/
├── workflows/
│   └── greeting.py      # Only Workflow classes
├── activities/
│   └── translate.py     # Only Activity functions/classes
├── worker.py            # Worker setup, imports both
└── starter.py           # Client code to start workflows

In the Workflow file, import Activities through the sandbox:

# workflows/greeting.py
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities.translate import TranslateActivities

Common Pitfalls

1. Using datetime.now() in workflows - Use workflow.now() instead
2. Blocking in async activities - Use sync activities or async-safe libraries only
3. Missing executor for sync activities - Add activity_executor=ThreadPoolExecutor()
4. Forgetting to heartbeat - Long activities need activity.heartbeat()
5. Using gevent - Incompatible with SDK
6. Using print() in workflows - Use workflow.logger instead for replay-safe logging
7. Mixing Workflows and Activities in same file - Causes unnecessary reloads, hurts performance

Additional Resources

Reference Files

• determinism.md - Sandbox behavior, safe alternatives, pass-through pattern, history replay
• sync-vs-async.md - Sync vs async activities, event loop blocking, executor configuration
• error-handling.md - ApplicationError, retry policies, non-retryable errors, idempotency
• testing.md - WorkflowEnvironment, time-skipping, activity mocking
• patterns.md - Signals, queries, child workflows, saga pattern
• observability.md - Logging, metrics, tracing, Search Attributes
• advanced-features.md - Continue-as-new, updates, schedules, dynamic workflows, and more
• data-handling.md - Data converters, Pydantic, payload encryption
• versioning.md - Patching API, workflow type versioning, Worker Versioning