Build a Temporal Application from scratch in Python

Prerequisites

Before starting this tutorial:

• Set up a local development environment for developing Temporal Applications with Python. Ensure the Temporal Service is running locally and you can access the Web UI on port 8233 (the default).
• Follow the Run your first Temporal application with the Python SDK tutorial to understand how Temporal's components fit together.

Create a new Python project

To get started with the Temporal Python SDK, create a new Python project and initialize a new virtual environment, just like any other Python program.

In a terminal, create a directory called hello-world-temporal:

mkdir hello-world-temporal

Change into that directory:

cd hello-world-temporal

Create a Python virtual environment with venv:

python -m venv env

python3 -m venv env

Activate the environment:

env\Scripts\activate

source env/bin/activate

Then install the Temporal SDK:

python -m pip install temporalio

You'll use the pytest framework to create and run your tests. To install pytest, use the following command:

python -m pip install pytest

You'll also need the pytest_asyncio package. Install that with the following command:

python -m pip install pytest_asyncio

With your project workspace configured, you're ready to create your first Temporal Workflow and Activity. In this tutorial, you'll start with the Workflow.

Create a Workflow

Workflows are where you configure and organize the execution of Activities. You write a Workflow using one of the programming languages supported by a Temporal SDK. This code is known as a Workflow Definition.

In the Temporal Python SDK, you define Workflow Definitions by creating a class and annotate it with the @workflow.defn decorator.

You then use the @workflow.run decorator to specify the method that the Workflow should invoke. Exactly one method must have this decorator and it must be added to an async def method.

Create the file workflows.py in the root of your project and add the following code to create a SayHello class to define the Workflow:

from datetime import timedelta
from temporalio import workflow

# Import activity, passing it through the sandbox without reloading the module
with workflow.unsafe.imports_passed_through():
    from activities import say_hello

@workflow.defn
class SayHello:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            say_hello, name, start_to_close_timeout=timedelta(seconds=5)
        )

In this example, the run method is decorated with @workflow.run, so it's the method that the Workflow will invoke.

This method accepts a string value that will hold the name, and it returns a string. You can learn more in the Workflow parameters section of the Temporal documentation.

The method calls the workflow.execute_activity method which executes an Activity called say_hello, which you'll define next. workflow.execute_activity needs the Activity Type, the input parameters for the Activity, and a Start-To-Close Timeout or Schedule-To-Close Timeout.

Finally, the run method returns the result of the Activity Execution.

With your Workflow Definition created, you're ready to create the say_hello Activity.

Create an Activity

In a Temporal Application, Activities are where you execute non-deterministic code or perform operations that may fail, such as API requests or database calls. Your Workflow Definitions call Activities and process the results. Complex Temporal Applications have Workflows that invoke many Activities, using the results of one Activity to execute another.

For this tutorial, your Activity won't be complex; you'll define an Activity that takes a string as input and uses it to create a new string as output, which is then returned to the Workflow. This will let you see how Workflows and Activities work together without building something complicated.

In the Temporal Python SDK, you define an Activity by decorating a function with @activity.defn.

Create a new file called activities.py and add the following code to define a say_hello function to define the Activity:

from temporalio import activity

@activity.defn
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

The logic within the say_hello function creates the string and returns the greeting.

Your Activity Definition can accept input parameters just like Workflow Definitions. Review the Activity parameters section of the Temporal documentation for more details, as there are some limitations you'll want to be aware of when running more complex applications.

Like Workflow Definitions, if you have more than one parameter for an Activity, you should bundle the data into a data class rather than sending multiple input parameters. This will make future updates easier.

You've completed the logic for the application; you have a Workflow and an Activity defined. Before moving on, you'll write a unit test for your Workflow.