- Performance and Security
- Data, AI & Analytics
Getting started with Temporal in Python: Workflow that never fails.
Getting started with Temporal in Python: Workflow that never fails.
You have a background job with 5 steps. Your server crashes at step 3. The job restarts from step 1. Temporal fixes that. It saves progress after every step, so if anything crashes it picks up exactly where it stopped. This is a short, practical guide for backend engineers who want reliable, long-running workflows in Python.
When a Crash Means Starting Over
You have a background job with five steps. Your server crashes at step 3. The job restarts from step 1, repeating work that already succeeded and, worse, sometimes repeating side effects that should only happen once.
Temporal fixes this. It saves progress after every step. If anything crashes, the worker restarts and the workflow picks up exactly where it stopped. No re-running completed steps, no lost state, no fragile bookkeeping you have to write yourself.
What Is Temporal?
Temporal is a workflow engine. You write workflows in plain Python, and Temporal runs them reliably with retries, timeouts, and durable state, without you having to manage any of that infrastructure yourself.
It has four main pieces. Once these click, everything else follows.
Writing Activities
An activity is just an async function decorated with @activity.defn. Each one does a single job.
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 so Temporal can replay the workflow deterministically.
workflow.unsafe.imports_passed_through(). Temporal wraps workflow code in a sandbox, and 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 resultRunning a Worker
The worker connects to Temporal and listens on a task queue.
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 you can use later to query or wait on it.
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. Temporal has scheduling built in.
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))]
),
),
)Checking Progress with Queries
Want to know what a running workflow is doing without waiting for it to finish? Add a @workflow.query method and 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 the UI while a long job is running.
workflow.sleep() Is Not asyncio.sleep()
This trips up almost every engineer who starts with Temporal. The two look identical but behave completely differently when a worker restarts.
# 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 and the Temporal UI
Skip Sleeps Locally
Testing a workflow with 10-minute sleeps is painful. We use a small helper that no-ops the wait in local development.
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.
Quick Summary
Everything in this guide, in one table to keep open while you build.
| Concept | What to remember |
|---|---|
| Activity | One async function. Import Django models inside the body, never at the top. |
| Workflow | Calls activities, no direct I/O. Use imports_passed_through() for imports. |
| Worker | Registers workflows and activities. Must list every activity it uses. |
| workflow.sleep() | Durable. Use it instead of asyncio.sleep() for all delays in a workflow. |
| @workflow.query | Check the status of a running workflow without waiting for it to finish. |
| Schedule | Built-in cron replacement. Handles missed runs. No extra tooling. |
| Temporal UI | localhost:8233. See every workflow, every step, and every error. |
Temporal in Python — FAQ
temporalio SDK to write workflows as plain async code, and Temporal runs them reliably with automatic retries, timeouts, and durable state. If your process crashes, Temporal resumes the workflow from exactly where it stopped instead of restarting from the beginning.asyncio.sleep() runs in memory on the worker process, so if the worker restarts the timer is lost. workflow.sleep() registers a durable timer on the Temporal server, so the wait survives worker restarts and the workflow resumes correctly. Always use workflow.sleep() for delays inside a workflow.client.create_schedule with a ScheduleSpec and ScheduleIntervalSpec to run a workflow on an interval. Temporal handles missed runs automatically, so no external cron job is needed.@workflow.query to your workflow class that returns the current state. From your application, get a workflow handle and call handle.query() at any time, even while the workflow is still running. This is commonly used to show a live progress bar in a UI.workflow.unsafe.imports_passed_through() tells the sandbox to allow those modules through unmodified, which prevents subtle import-related errors in workflow definitions.Worker with the task queue name plus the list of workflows and activities it should handle, and call worker.run(). You must register every workflow and activity the worker uses, or the workflow will hang waiting for a worker that can handle it.http://localhost:8233. It shows every workflow that has run, including status, inputs, outputs, each activity step, and the full event history, which makes debugging far faster than reading logs. The Temporal server itself listens on localhost:7233.Building Reliable Backend Systems?
Durable workflows, data pipelines, and AI systems that hold up at scale are what Auriga IT builds every day. From NHAI processing 20M+ daily transactions to data migrations with zero data loss, this is our home turf.
Related content
Auriga: Leveling Up for Enterprise Growth!
Auriga’s journey began in 2010 crafting products for India’s [...]
