Getting started with Temporal in Python: Workflow that never fails.

A short guide for backend engineers who want reliable, long-running workflows.

The Problem

You have a background job with 5 steps.

Your server crashes at step 3.

The job restarts from step 1.

Temporal fixes this. It saves progress after every step. If anything crashes, it picks up exactly where it stopped.

What Is Temporal?

Temporal is a workflow engine.

You write workflows in plain Python. Temporal runs them reliably — with retries, timeouts, and durable state — without you having to manage any of that.

It has four main pieces:

• Activity — one function, one job (DB call, API call, file read)
• Workflow — the blueprint that calls activities in order
• Worker — your Python process that executes the workflow
• Task Queue — a named channel connecting workflows to workers

Writing Activities

An activity is just an async function with @activity.defn.

One rule: import Django models inside the function body, not at the top of the file.

Temporal loads your activity file before Django is fully ready. Top-level ORM imports will crash.

from temporalio import activity

@activity.defn
async def fetch_order(order_id: int) -> dict:
    from asgiref.sync import sync_to_async
    from orders.models import Order

    def _get():
        order = Order.objects.get(pk=order_id)
        return {"id": order.id, "status": order.status}

    return await sync_to_async(_get)()

@activity.defn
async def process_order(order_id: int) -> str:
    # do the actual processing
    return "processed"

Writing a Workflow

A workflow calls activities in sequence. No direct I/O here — no database, no API calls, nothing. Everything goes through activities.

One more rule: wrap activity imports with workflow.unsafe.imports_passed_through().

Temporal wraps workflow code in a sandbox. Without this, your imports get blocked and cause subtle errors.

from temporalio import workflow
from temporalio.common import RetryPolicy
from datetime import timedelta

with workflow.unsafe.imports_passed_through():
    from myapp.activities import fetch_order, process_order

@workflow.defn
class OrderWorkflow:
    def __init__(self):
        self.progress = "Starting"

    @workflow.run
    async def run(self, order_id: int) -> str:
        retry = RetryPolicy(maximum_attempts=3)

        self.progress = "Fetching order"
        order = await workflow.execute_activity(
            fetch_order,
            args=[order_id],
            start_to_close_timeout=timedelta(minutes=2),
            retry_policy=retry,
        )

        self.progress = "Processing order"
        result = await workflow.execute_activity(
            process_order,
            args=[order["id"]],
            start_to_close_timeout=timedelta(minutes=10),
            retry_policy=retry,
        )

        self.progress = "Done"
        return result

Running a Worker

The worker connects to Temporal and listens on a queue.

Register every workflow and every activity you use — if you forget one, the workflow will hang waiting for a worker that can handle it.

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from myapp.workflows import OrderWorkflow
from myapp.activities import fetch_order, process_order

async def main():
    client = await Client.connect("localhost:7233")

    worker = Worker(
        client=client,
        task_queue="orders",
        workflows=[OrderWorkflow],
        activities=[fetch_order, process_order],
    )
    await worker.run()

asyncio.run(main())

Triggering a Workflow

Start a workflow from your Django app and get back a workflow ID.

import asyncio, time
from temporalio.client import Client
from myapp.workflows import OrderWorkflow

async def _start(order_id: int) -> str:
    client = await Client.connect("localhost:7233")
    handle = await client.start_workflow(
        OrderWorkflow.run,
        args=[order_id],
        id=f"order-{order_id}-{int(time.time())}",
        task_queue="orders",
    )
    return handle.id

def trigger_order_workflow(order_id: int) -> str:
    return asyncio.run(_start(order_id))

Scheduling a Workflow

Need to run a workflow on a schedule? No cron needed.

from temporalio.client import (
    Client, Schedule, ScheduleActionStartWorkflow,
    ScheduleSpec, ScheduleIntervalSpec
)

async def schedule_daily(order_id: int):
    client = await Client.connect("localhost:7233")

    await client.create_schedule(
        f"daily-order-{order_id}",
        Schedule(
            action=ScheduleActionStartWorkflow(
                OrderWorkflow.run,
                args=[order_id],
                task_queue="orders",
            ),
            spec=ScheduleSpec(
                intervals=[ScheduleIntervalSpec(every=timedelta(days=1))]
            ),
        ),
    )

Temporal runs it every day and handles missed runs automatically.

Checking Progress with workflow.query

Want to know what a running workflow is doing — without waiting for it to finish?

Add a @workflow.query method. Call it from anywhere.

# Inside the workflow class
@workflow.query
def get_progress(self) -> str:
    return self.progress

# From Django
async def check_status(workflow_id: str) -> str:
    client = await Client.connect("localhost:7233")
    handle = client.get_workflow_handle(workflow_id)
    return await handle.query(OrderWorkflow.get_progress)

We use this to show a live progress bar in our UI while a long job is running.

workflow.sleep() — Not the Same as asyncio.sleep()

This trips up almost every engineer when they start with Temporal.

Wrong — if the worker restarts, this timer is lost

await asyncio.sleep(600)

Correct — the timer lives on the Temporal server, survives restarts

await workflow.sleep(timedelta(minutes=10))

We use this to poll a batch job that takes hours:

for attempt in range(50):
    is_done = await workflow.execute_activity(check_batch_status, ...)
    if is_done:
        break
    await workflow.sleep(timedelta(minutes=10))

The worker can crash and restart between any two of those polls. Temporal picks up at the right attempt every time.

Dev Tip: Skip Sleeps Locally

Testing a workflow with 10-minute sleeps is painful. We use a small helper:

import os
from temporalio import workflow

async def dev_sleep(duration):
    if os.getenv("DEV_MODE") == "true":
        return  # no wait in local dev
    await workflow.sleep(duration)

Set DEV_MODE=true locally, and the polling loop runs instantly.

The Temporal UI

Temporal comes with a built-in web UI at http://localhost:8233.

You can see every workflow that has ever run — its status, inputs, outputs, each activity step, and the full event history. No extra tooling needed.

![Temporal UI showing workflow list and event history]

This alone saves hours of debugging. Instead of digging through logs, you open the UI and see exactly where a workflow failed and why.

Quick Summary

Concept

What to remember

Activity — One async function. Import Django models inside the body.

Workflow — Calls activities. Use imports_passed_through() for imports.

Worker — Registers workflows + activities. Must list every activity.

workflow.sleep() — Durable. Use instead of asyncio.sleep().

@workflow.query — Check status of a running workflow.

Schedule — Built-in cron. No extra tooling.

UI — localhost:8233. See every workflow, step, and error.

Questions? Drop them in the comments.