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

Python Gotchas

Python-specific mistakes and anti-patterns. See also Common Gotchas for language-agnostic concepts.

File Organization

Importing Activities into Workflow Files

The Problem: The Python sandbox reloads workflow files on every task. Importing heavy activity modules slows down workers.

# BAD - activities.py gets reloaded constantly
# workflows.py
from activities import my_activity

@workflow.defn
class MyWorkflow:
    pass

# GOOD - Pass-through import
# workflows.py
from temporalio import workflow

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

@workflow.defn
class MyWorkflow:
    pass

references/python/determinism-protection.md contains more info about the Python sandbox.

Mixing Workflows and Activities

# BAD - Everything in one file
# app.py
@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self):
        await workflow.execute_activity(my_activity, ...)

@activity.defn
async def my_activity():
    # Heavy imports, I/O, etc.
    pass

# GOOD - Separate files
# workflows.py
@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self):
        await workflow.execute_activity(my_activity, ...)

# activities.py
@activity.defn
async def my_activity():
    pass

Async vs Sync Activities

The Temporal Python SDK supports both async and sync activities. See references/python/sync-vs-async.md to understand which to choose. Below are important anti-patterns for both aysnc and sync activities.

Blocking in Async Activities

# BAD - Blocks the event loop
@activity.defn
async def process_file(path: str) -> str:
    with open(path) as f:  # Blocking I/O in async!
        return f.read()

# GOOD Option 1 - Use sync activity with executor
@activity.defn
def process_file(path: str) -> str:
    with open(path) as f:
        return f.read()

# Register with executor in worker
Worker(
    client,
    task_queue="my-queue",
    activities=[process_file],
    activity_executor=ThreadPoolExecutor(max_workers=10),
)

# GOOD Option 2 - Use async I/O
@activity.defn
async def process_file(path: str) -> str:
    async with aiofiles.open(path) as f:
        return await f.read()

Missing Executor for Sync Activities

# BAD - Sync activity REQUIRES executor
@activity.defn
def slow_computation(data: str) -> str:
    return heavy_cpu_work(data)

Worker(
    client,
    task_queue="my-queue",
    activities=[slow_computation],
    # Missing activity_executor! --> THIS IMMEDIATELY RAISES AN EXCEPTION!
)

# GOOD - Provide executor
Worker(
    client,
    task_queue="my-queue",
    activities=[slow_computation],
    activity_executor=ThreadPoolExecutor(max_workers=10),
)

Wrong Retry Classification

Example: Transient networks errors should be retried. Authentication errors should not be. See references/python/error-handling.md to understand how to classify errors.

Heartbeating

Forgetting to Heartbeat Long Activities

# BAD - No heartbeat, can't detect stuck activities
@activity.defn
async def process_large_file(path: str):
    async for chunk in read_chunks(path):
        process(chunk)  # Takes hours, no heartbeat

# GOOD - Regular heartbeats with progress
@activity.defn
async def process_large_file(path: str):
    async for i, chunk in enumerate(read_chunks(path)):
        activity.heartbeat(f"Processing chunk {i}")
        process(chunk)

Heartbeat Timeout Too Short

# BAD - Heartbeat timeout shorter than processing time
await workflow.execute_activity(
    process_chunk,
    start_to_close_timeout=timedelta(minutes=30),
    heartbeat_timeout=timedelta(seconds=10),  # Too short!
)

# GOOD - Heartbeat timeout allows for processing variance
await workflow.execute_activity(
    process_chunk,
    start_to_close_timeout=timedelta(minutes=30),
    heartbeat_timeout=timedelta(minutes=2),
)

Set heartbeat timeout as high as acceptable for your use case — each heartbeat counts as an action.

Cancellation

Not Handling Workflow Cancellation

# BAD - Cleanup doesn't run on cancellation
@workflow.defn
class BadWorkflow:
    @workflow.run
    async def run(self) -> None:
        await workflow.execute_activity(
            acquire_resource,
            start_to_close_timeout=timedelta(minutes=5),
        )
        await workflow.execute_activity(
            do_work,
            start_to_close_timeout=timedelta(minutes=5),
        )
        await workflow.execute_activity(
            release_resource,  # Never runs if cancelled!
            start_to_close_timeout=timedelta(minutes=5),
        )

# GOOD - Use try/finally for cleanup
@workflow.defn
class GoodWorkflow:
    @workflow.run
    async def run(self) -> None:
        await workflow.execute_activity(
            acquire_resource,
            start_to_close_timeout=timedelta(minutes=5),
        )
        try:
            await workflow.execute_activity(
                do_work,
                start_to_close_timeout=timedelta(minutes=5),
            )
        finally:
            # Runs even on cancellation
            await workflow.execute_activity(
                release_resource,
                start_to_close_timeout=timedelta(minutes=5),
            )

Not Handling Activity Cancellation

Activities must opt in to receive cancellation. This requires:

1. Heartbeating - Cancellation is delivered via heartbeat
2. Catching the cancellation exception - Exception is raised when heartbeat detects cancellation

Cancellation exceptions:

• Async activities: asyncio.CancelledError
• Sync threaded activities: temporalio.exceptions.CancelledError

# BAD - Activity ignores cancellation
@activity.defn
async def long_activity() -> None:
    await do_expensive_work()  # Runs to completion even if cancelled

# GOOD - Heartbeat and catch cancellation
@activity.defn
async def long_activity() -> None:
    try:
        for item in items:
            activity.heartbeat()
            await process(item)
    except asyncio.CancelledError:
        await cleanup()
        raise

Testing

Not Testing Failures

It is important to make sure workflows work as expected under failure paths in addition to happy paths. Please see references/python/testing.md for more info.

Not Testing Replay

Replay tests help you test that you do not have hidden sources of non-determinism bugs in your workflow code, and should be considered in addition to standard testing. Please see references/python/testing.md for more info.

Timers and Sleep

Using asyncio.sleep

# BAD: asyncio.sleep is not deterministic during replay
import asyncio

@workflow.defn
class BadWorkflow:
    @workflow.run
    async def run(self) -> None:
        await asyncio.sleep(60)  # Non-deterministic!

# GOOD: Use workflow.sleep for deterministic timers
from temporalio import workflow
from datetime import timedelta

@workflow.defn
class GoodWorkflow:
    @workflow.run
    async def run(self) -> None:
        await workflow.sleep(timedelta(seconds=60))  # Deterministic
        # Or with string duration:
        await workflow.sleep("1 minute")

Why this matters: asyncio.sleep uses the system clock, which differs between original execution and replay. workflow.sleep creates a durable timer in the event history, ensuring consistent behavior during replay.