Run your first Temporal application with the Python SDK

~15 minutes totalTemporal beginnerHands-on tutorial

Understand the application
Run the application
Simulate failures

In this tutorial, you'll run your first Temporal Application using the Python SDK. You'll use the Web UI for state visibility, then explore how Temporal helps you recover from common failures.

What you'll do

Explore Temporal's core terminology and concepts.
Run a Temporal Workflow Application using a Temporal Cluster and the Python SDK.
Practice reviewing the state of the Workflow.
Understand the inherent reliability of Workflow methods.

Prerequisites

Before starting this tutorial:

Set up a local development environment for developing Temporal Applications with Python
Ensure you have Git installed to clone the project.

Application overview

This project simulates a money transfer application: withdrawals, deposits, and refunds. Money comes out of one account and goes into another. If the withdrawal succeeds but the deposit fails, the money needs to go back to the original account.

Temporal automatically maintains application state when something fails - recovering processes where they left off or rolling them back. You focus on business logic instead of writing recovery code.

The following diagram illustrates what happens when you start the Workflow:

High level project design

None of your application code runs on the Temporal Server. Your Worker, Workflow, and Activity run on your infrastructure, along with the rest of your applications.

Download the example application

The application is available in a GitHub repository. Clone it:

git clone https://github.com/temporalio/money-transfer-project-template-python

cd money-transfer-project-template-python

tip

The repository is a GitHub Template, so you can clone it to your own account and use it as the foundation for your own Temporal application.

Workflow Definition

A Workflow Definition in Python uses the @workflow.defn decorator on the Workflow class. Here's what the Workflow Definition looks like:

workflows.py
from datetime import timedelta

from temporalio import workflow
from temporalio.common import RetryPolicy
from temporalio.exceptions import ActivityError

with workflow.unsafe.imports_passed_through():
    from activities import BankingActivities
    from shared import PaymentDetails


@workflow.defn
class MoneyTransfer:
    @workflow.run
    async def run(self, payment_details: PaymentDetails) -> str:
        retry_policy = RetryPolicy(
            maximum_attempts=3,
            maximum_interval=timedelta(seconds=2),
            non_retryable_error_types=["InvalidAccountError", "InsufficientFundsError"],
        )

        # Withdraw money
        withdraw_output = await workflow.execute_activity_method(
            BankingActivities.withdraw,
            payment_details,
            start_to_close_timeout=timedelta(seconds=5),
            retry_policy=retry_policy,
        )

        # Deposit money
        try:
            deposit_output = await workflow.execute_activity_method(
                BankingActivities.deposit,
                payment_details,
                start_to_close_timeout=timedelta(seconds=5),
                retry_policy=retry_policy,
            )

            result = f"Transfer complete (transaction IDs: {withdraw_output}, {deposit_output})"
            return result
        except ActivityError as deposit_err:
            workflow.logger.error(f"Deposit failed: {deposit_err}")
            try:
                refund_output = await workflow.execute_activity_method(
                    BankingActivities.refund,
                    payment_details,
                    start_to_close_timeout=timedelta(seconds=5),
                    retry_policy=retry_policy,
                )
                workflow.logger.info(f"Refund successful. Confirmation ID: {refund_output}")
                raise deposit_err
            except ActivityError as refund_error:
                workflow.logger.error(f"Refund failed: {refund_error}")
                raise refund_error

The MoneyTransfer class takes transaction details, executes Activities to withdraw and deposit, and returns results. The run method takes a PaymentDetails input:

shared.py
from dataclasses import dataclass

@dataclass
class PaymentDetails:
    source_account: str
    target_account: str
    amount: int
    reference_id: str

tip

It's a good practice to send a single data class object into a Workflow as its input, rather than multiple separate arguments.

Activity Definition

In the Python SDK, you define an Activity by decorating a method with @activity.defn. The withdraw() Activity takes transfer details and calls a service:

activities.py
@activity.defn
async def withdraw(self, data: PaymentDetails) -> str:
    reference_id = f"{data.reference_id}-withdrawal"
    try:
        confirmation = await asyncio.to_thread(
            self.bank.withdraw, data.source_account, data.amount, reference_id
        )
        return confirmation
    except InvalidAccountError:
        raise
    except Exception:
        activity.logger.exception("Withdrawal failed")
        raise

The deposit() method looks almost identical:

activities.py
@activity.defn
async def deposit(self, data: PaymentDetails) -> str:
    reference_id = f"{data.reference_id}-deposit"
    try:
        confirmation = await asyncio.to_thread(
            self.bank.deposit, data.target_account, data.amount, reference_id
        )
        """
        confirmation = await asyncio.to_thread(
            self.bank.deposit_that_fails,
            data.target_account,
            data.amount,
            reference_id,
        )
        """
        return confirmation
    except InvalidAccountError:
        raise
    except Exception:
        activity.logger.exception("Deposit failed")
        raise

The commented block in deposit() is what you'll uncomment later to simulate a failure.

Why you use Activities

Temporal Workflows have deterministic constraints and must produce the same output each time, given the same input. Non-deterministic work like file or network access must be done by Activities. Use Activities for business logic and Workflows to coordinate.

Set the Retry Policy

If an Activity fails, Temporal Workflows automatically retry. At the top of the MoneyTransfer Workflow, you'll see a Retry Policy:

workflows.py
retry_policy = RetryPolicy(
    maximum_attempts=3,
    maximum_interval=timedelta(seconds=2),
    non_retryable_error_types=["InvalidAccountError", "InsufficientFundsError"],
)

By default, Temporal retries failed Activities forever, but you can specify non-retryable error types and maximum attempts. This example retries up to 3 times.

This is a simplified example

Transferring money is tricky and this tutorial doesn't cover all edge cases. In production you'd add more advanced logic - including a "human in the loop" step where someone is notified of refund issues and can intervene.

PreviousSet up your dev environment Next: chapter 2 Run the application