general-temporal

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

# 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