General Temporal

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

# 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.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()

# 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())

# 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())

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
    ),
)

@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)

@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}

@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 ...

@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
)

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]))

# 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"

# Use workflow.patched() to change workflow logic without breaking running workflows
@workflow.run
async def run(self, url: str) -> str:
    if workflow.patched("use-v2-processor"):
        # New code path — for workflows started after this deploy
        result = await workflow.execute_activity(
            process_data_v2, url,
            start_to_close_timeout=timedelta(minutes=5),
        )
    else:
        # Old code path — for workflows already in flight
        result = await workflow.execute_activity(
            process_data, url,
            start_to_close_timeout=timedelta(minutes=5),
        )
    return result