Skip to main content

Set up a Temporal Application project - Python SDK dev guide

Last updated on
Tags:

This section covers how to use a terminal, a code editor, and a development Cluster to create a Namespace, write a single Activity Workflow, run a Worker that talks to your development Cluster, run a Workflow using the temporal CLI, add a testing framework, and view Workflows in the Web UI.

Construct a new Temporal Application project

This section of the Temporal Python SDK Background Check tutorial covers the minimum set of concepts and implementation details needed to build and run a Temporal Application using Python.

By the end of this section you will know how to construct a new Temporal Application project.

Choose your development environment

There are three ways to follow this guide:

Read more in the Choose a development Cluster section on this page.

Install the Temporal CLI​

The Temporal CLI is available on MacOS, Windows, and Linux. Reference the documentation for detailed install information.

Install via download​

  1. Download the version for your OS and architecture:
  2. Extract the downloaded archive.
  3. Add the temporal binary to your PATH (temporal.exe for Windows).

Install via Homebrew​

brew install temporal

Build​

  1. Install Go
  2. Clone repository
  3. Switch to cloned directory, and run go build ./cmd/temporal
note
  • The executable will be at temporal (temporal.exe for Windows). Reference the documentation for detailed usage information.

Choose a development Cluster​

Which development Cluster should you choose?

We recommend choosing a development environment based on your requirements.

The source code for the Temporal Server (the orchestrating component of the Temporal Cluster) is licensed under the MIT open source license. So, in theory, anyone can take the Temporal Server code and run their Temporal Platform in any number of creative ways.

However, for most developers we recommend starting by choosing one of the following:

Temporal does not directly run your code

Keep in mind that in every scenario, the “Temporal Platform” does not host and run your Workers (application code). It is up to you, the developer, to host your application code. The Temporal Platform ensures that properly written code durably executes in the face of platform-level failures.

Local dev server​

When to use a local development server?

We recommend using the local development server if you are new to Temporal, or want to start something from scratch and don’t have a self-hosted environment ready or want to pay for a Temporal Cloud account.

The Temporal CLI comes bundled with a development server and provides a fast way to start running Temporal Applications.

However, the local development server does not emit any metrics. If you are eager to to set up Cluster-level metrics for performance tuning, we recommend using a self-hosted Cluster or Temporal Cloud.

Start the dev server​

How to start a local development server

If you have successfully installed the Temporal CLI, open a new terminal and run the following command:

temporal server start-dev --db-filename temporal.db

This command automatically starts the Temporal Web UI, creates a default Namespace, and creates a persistence database.

The Temporal Web UI serves to http://localhost:8233.

For more command details and options, see the CLI reference

Create a custom Namespace​

How to create a Namespace on the development server

The development server does automatically create a default Namespace (named "default") when it starts up. However, you will create a custom one for our application. Since this is something recommended at a production level, it's recommend practicing it with the development server.

Use the temporal operator namespace create command using the Temporal CLI to create a Namespace on the development server.

temporal operator namespace create backgroundcheck_namespace

For command details and options, see the CLI reference.

Temporal Cloud​

When to use Temporal Cloud

If you do not have a Temporal Cloud Account, you can request one using the link on the Get started with Temporal Cloud guide.

We recommend starting off with Temporal Cloud if you already have a production use case, or need to move a scalable proof of concept into production.

In other words, Temporal Cloud is perfect if you are ready to run at scale and don’t want the overhead of managing your own self-hosted Cluster.

To create a Namespace in Temporal Cloud, follow the instructions in How to create a Namespace.

Safely store your certificate and private key

Store certificates and private keys generated for your Namespace as files or environment variables in your project. You need access to your certificate and key to run your Workers and start Workflows.

For more information on certificate requirements, see How to manage certificates in Temporal Cloud.

Self-hosted Temporal Cluster​

We recommend using a self-hosted environment if you are starting something new and need to scale with production-level features, but don’t yet need or want to pay for Temporal Cloud.

For example, running a self-hosted Cluster lets you try different databases, view Cluster metrics, use custom Search Attributes, and even play with the Archival feature.

For the purposes of this guide, we show how to use a self-hosted environment that runs completely out of Docker. We acknowledge that it takes a fair amount of experience to elevate from a self-hosted environment in Docker to something that can run at an enterprise production scale. The self-hosted information in this guide should help you make more informed decisions.

To follow along with self-hosted parts of this guide, install the following:

Then, clone the temporalio/docker-compose repository.

Change directory into the root of the project.

Run the docker compose up command.

git clone https://github.com/temporalio/docker-compose.git
cd  docker-compose
docker compose up

Create a command alias for the Temporal CLI:

alias temporal_docker="docker exec temporal-admin-tools temporal"

Create a Namespace.

temporal_docker operator namespace create backgroundcheck_namespace

Boilerplate Temporal Application project code​

What is the minimum code I need to create a boilerplate Temporal Application?

Let’s start with a single Activity Workflow and register those functions with a Worker.

After we get the Worker running and have started a Workflow Execution, we will add a testing framework.

Project structure​

You can organize Temporal Application code to suit various needs in a way that aligns with the idiomatic style of the language you are working in. This includes structuring your files according to your organization's best practices.

However, there are some general ways to think about organizing code.

The best practice is to group Workflows together, Activities together, and separate your Worker process into a standalone file. Often this happens respectively per use case, business process, or domain.

For monorepo-style organizational techniques, consider a designated Workflow directory for each use case and place each Workflow in its own file, but also maintain a dedicated place for shared Activities.

For example, your project structure could look like this:

/monorepo
    /shared_activities
        | payment.py
        | send_email.py
    /background_check
        /workflows
            | background_check_workflow.py
        /activities
            | ssn_trace_activity.py
        /worker
            | main.py
    /loan_application
        /workflows
            | loan_application_workflow.py
        /activities
            | credit_check_activity.py
        /worker
            | main.py
    /tests
       | pytest.ini
       | workflow_tests.py
       | activity_tests.py

If you are following along with this guide, your project will look like this:

/backgroundcheck
    /workflows
        | background_check_workflow.py
    /activities
        | ssn_trace_activity.py
    /worker
        | main.py
    /tests
       | pytest.ini
       | workflow_tests.py
       | activity_tests.py

Initialize Python project dependency framework​

In Python, you’d typically use pip and virtualenv or venv for dependency management and environment isolation.

For more information, see Creating Virtual Environments.

Set up a virtual environment for the project and initialize it using pip.

mkdir background_check
cd background_check
python -m venv venv
source venv/bin/activate  # On Windows, use `venv\Scripts\activate`
pip install temporalio

After activation, any Python command you run will use the virtual environment's isolated Python interpreter and libraries. Remember to always activate the virtual environment when working on this project.

Boilerplate Workflow code​

In the Temporal Python SDK programming model, a Workflow Definition is defined as a class. The BackgroundCheck class below is an example of a basic Workflow Definition.

docs/tutorials/python/background-check/code/backgroundcheck_boilerplate/workflows/backgroundcheck.py

from datetime import timedelta

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities.ssntraceactivity_dacx import ssn_trace_activity

@workflow.defn
class BackgroundCheck:
    @workflow.run
    async def run(self, ssn: str) -> str:
        return await workflow.execute_activity(
            ssn_trace_activity,
            ssn,
            schedule_to_close_timeout=timedelta(seconds=5),
        )

Use the @workflow.defn decorator on the BackgroundCheck class to identify a Workflow.

Use the @workflow.run to mark the entry point method to be invoked. This must be set on one asynchronous method defined on the same class as @workflow.defn.

Run methods have positional parameters.

In this example, pass in the Activity name, ssn_trace_activity and an argument, ssn. We get into the best practices around Workflow params and returns in the one of the next sections.

Boilerplate Activity code​

In the Temporal Python SDK programming model, an Activity is a function and can be used as an instance method of a class You can use asynchronous, synchronous multithreaded, and synchronous multiprocess/other functions to define an Activity.

Below is an example of an Activity defined as a function.

docs/tutorials/python/background-check/code/backgroundcheck_boilerplate/activities/ssntraceactivity.py

from temporalio import activity

@activity.defn
async def ssn_trace_activity(ssn) -> str:
    return "pass"

The ssn_trace_activity function passes a string and returns pass.

An Activity Definition can support as many other custom parameters as needed; however, all parameters must be serializable.

The default data converter supports converting multiple types including:

  • None
  • bytes
  • google.protobuf.message.Message - As JSON when encoding, but has ability to decode binary proto from other languages
  • Anything that can be converted to JSON including:
    • Anything that [json.dump](https://docs.python.org/3/library/json.html#json.dump) supports natively
    • dataclasses
    • Iterables including ones JSON dump may not support by default, e.g. set
    • Any class with a dict() method and a static parse_obj() method, e.g. Pydantic models
      • The default data converter is deprecated for Pydantic models and will warn if used since not all fields work. See this sample for the recommended approach.
    • IntEnum, StrEnum based enumerates
    • UUID

This notably doesn't include any date, time, or datetime objects as they may not work across SDKs.

Users are strongly encouraged to use a single dataclass for parameter and return types so fields with defaults can be easily added without breaking compatibility.

Run a dev server Worker​

To run a Worker Process with a local development server, define the following steps in code:

  • Initialize a Temporal Client.
  • Create a new Worker by passing the Client to creation call.
  • Register the application's Workflow and Activity functions.
  • Call run on the Worker.

In regards to organization, we recommend keeping Worker code separate from Workflow and Activity code.

docs/tutorials/python/background-check/code/backgroundcheck_boilerplate/dev_server_worker/main.py

import asyncio

from temporalio.client import Client
from temporalio.worker import Worker

from activities.ssntraceactivity_dacx import ssn_trace_activity
from workflows.backgroundcheck_dacx import BackgroundCheck

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

    worker = Worker(
        client,
        namespace="backgroundcheck_namespace",
        task_queue="backgroundcheck-boilerplate-task-queue",
        workflows=[BackgroundCheck],
        activities=[ssn_trace_activity],
    )

    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

Run a Temporal Cloud Worker​

A Temporal Cloud Worker requires that you specify the following in the Client connection options:

  • Temporal Cloud Namespace
  • Temporal Cloud Address
  • Certificate and private key associated with the Namespace

docs/tutorials/python/background-check/code/backgroundcheck_boilerplate/cloud_worker/main.py

import asyncio
import os

from temporalio.client import Client, TLSConfig
from temporalio.worker import Worker

from activities.ssntraceactivity_dacx import ssn_trace_activity
from workflows.backgroundcheck_dacx import BackgroundCheck

async def main():
    with open(os.getenv("TEMPORAL_MTLS_TLS_CERT"), "rb") as f:
        client_cert = f.read()

    with open(os.getenv("TEMPORAL_MTLS_TLS_KEY"), "rb") as f:
        client_key = f.read()

    client = await Client.connect(
        os.getenv("TEMPORAL_HOST_URL"),
        namespace=os.getenv("TEMPORAL_NAMESPACE"),
        tls=TLSConfig(
            client_cert=client_cert,
            client_private_key=client_key,
        ),
    )

    worker = Worker(
        client,
        task_queue="backgroundcheck-boilerplate-task-queue",
        workflows=[BackgroundCheck],
        activities=[ssn_trace_activity],
    )
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

To run a Temporal Cloud Worker, you'll change some parameters in your Client connection code, such as updating the namespace and gRPC endpoint. You'll use:

Run a Self-hosted Worker​

To deploy a self-hosted Worker to your Docker environment, you need to configure your Worker with the appropriate IP address and port.

Confirm network​

The default docker-compose.yml file in the temporalio/docker-compose repo has the Temporal Server exposed on port 7233 on the temporal-network.

services:
  # ...
  temporal:
    container_name: temporal
    # ...
    networks:
      - temporal-network
    ports:
      - 7233:7233
    # ...
  # ...

If you are using a different or customized docker compose file, you can see the available networks by using the following command:

docker network ls

Confirm IP address​

Get the IP address of the Docker network that the containers are using.

To do that, first inspect the network:

docker network inspect temporal-network

Look for the container named temporal.

Example output:

[
  {
    "Name": "temporal-network",
    // ...
    "Containers": {
      // ...
      "53cf62f0cc6cfd2a9627a2b5a4c9f48ffe5a858f0ef7b2eaa51bf7ea8fd0e86f": {
        "Name": "temporal",
        // ...
        "IPv4Address": "172.18.0.4/16"
        // ...
      }
      // ...
    }
    // ...
  }
]

Copy the IP address part.

Customize Client options​

Set IP address, port, and Namespace in the Temporal Client options.

docs/tutorials/python/background-check/code/backgroundcheck_boilerplate/self_hosted_worker/main.py

import asyncio

from temporalio.client import Client
from temporalio.worker import Worker

from activities.ssntraceactivity_dacx import ssn_trace_activity
from workflows.backgroundcheck_dacx import BackgroundCheck

async def main():
    client = await Client.connect(
        "172.18.0.4:7233"  # The IP address of the Temporal Server on your network.
    )

    worker = Worker(
        client,
        task_queue="backgroundcheck-boilerplate-task-queue",
        workflows=[BackgroundCheck],
        activities=[ssn_trace_activity],
    )
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

Build and deploy Docker image​

note

This Dockerfile is used to containerize the Background Check application so that it can run seamlessly in any environment that supports Docker.

Add a Docker file to the root of your Background Check application project.

Name the file Dockerfile, with no extensions, and add the following configuration:

FROM python:3.11

RUN mkdir /app

COPY . /app

COPY pyproject.toml /app

WORKDIR /app

RUN pip3 install poetry

RUN poetry config virtualenvs.create false

RUN poetry install

CMD [ "poetry", "run", "python", "/app/run_worker.py" ]

Then build the Docker image using the following command:

docker build . -t backgroundcheck-worker-image:latest

Now run the Worker on the same network as the Temporal Cluster containers using the following command:

docker run --network temporal-network backgroundcheck-worker-image:latest

Start Workflow using the Temporal CLI​

How to start a Workflow using the Temporal CLI

You can use the Temporal CLI to start a Workflow whether you are using a local development server, Temporal Cloud, or are in a self-hosted environment. However, you need to provide additional options to the command when operating with the Temporal Cloud or self-hosted environments.

Local dev Server​

How to start a Workflow with the Temporal CLI while using the local development server

Use the Temporal CLI temporal workflow start command to start your Workflow.

temporal workflow start \
 --task-queue backgroundcheck-boilerplate-task-queue \
 --type BackgroundCheck \
 --input '"555-55-5555"' \
 --namespace backgroundcheck_namespace

For more details, see the temporal workflow start command API reference.

After you start the Workflow, you can see it in the Temporal Platform. Use the Temporal CLI or the Temporal Web UI to monitor the Workflow's progress.

List Workflows​

Use the 'temporal workflow list` command to list all of the Workflows in the Namespace:

temporal workflow list \
 --namespace backgroundcheck_namespace

View in Web UI​

You can also use the Web UI to see the Workflows associated with the Namespace.

The local development server starts the Web UI at http://localhost:8233.

When you visit for the first time, the Web UI directs you to http://localhost:8233/namespaces/default/workflows.

Use the Namespace dropdown to select the project Namespace you created earlier.

Web UI Namespace selection

You should now be at http://localhost:8233/namespaces/backgroundcheck_namespace/workflows.

Temporal Cloud​

How to start a Workflow with Temporal CLI when using Temporal Cloud

Run the temporal workflow start command, and make sure to specify the certificate and private key arguments.

temporal workflow start \
 --task-queue backgroundcheck-boilerplate-task-queue-cloud \
 --type BackgroundCheck \
 --tls-cert-path ca.pem \
 --tls-key-path ca.key \
 --input '"555-55-5555"' \
 --namespace <namespace>.<account-id> \
 --address <namespace>.<account-id>.tmprl.cloud:<port>

Make sure that the certificate path, private key path, Namespace, and address argument values match your project.

Use environment variables

Use environment variables as a way to quickly switch between a local dev server and Temporal Cloud, for example.

You can customize the environment names to be anything you want.

# set Cloud env variables
temporal env set cloud.namespace <namespace>.<account-id>
temporal env set cloud.address <namespace>.<account-id>.tmprl.cloud:<port>
temporal env set cloud.tls-cert-path ca.pem
temporal env set cloud.tls-key-path ca.key
# set local env variables
temporal env set local.namespace <namespace>

In this way, you can just provide a single --env command option when using the Temporal CLI rather than specifying each connection option in every command.

temporal workflow start \
 # ...
 --env cloud \
 # ...

List Workflows​

Run the temporal workflow list command, and make sure to specify the certificate and private key arguments.

temporal workflow list \
 --tls-cert-path ca.pem \
 --tls-key-path ca.key \
 --namespace <namespace>.<account-id> \
 --address <namespace>.<account-id>.tmprl.cloud:<port>

View in Web UI​

Visit the Workflows page of your Cloud Namespace. The URL will look something like the following:

https://cloud.temporal.io/namespaces/<namespace>.<account-id>/workflows

View Workflows in the Cloud UI

Self-hosted​

How to start a Workflow with the Temporal CLI when using a Self-hosted Cluster

Use your Temporal CLI alias to run the temporal workflow start command and start your Workflow.

temporal_docker workflow start \
 --task-queue backgroundcheck-boilerplate-task-queue-self-hosted \
 --type BackgroundCheck \
 --input '"555-55-5555"' \
 --namespace backgroundcheck_namespace

List Workflows​

Using your Temporal CLI alias, run the temporal workflow list command. This command lists the Workflows Executions within the Namespace:

temporal_docker workflow list \
 --namespace backgroundcheck_namespace

View in the Web UI​

When you visit for the first time, the Web UI directs you to http://localhost:8233/namespaces/default/workflows.

Use the Namespace dropdown to select the project Namespace you created earlier.

You should now be at http://localhost:8080/namespaces/backgroundcheck_namespace/workflows.

Add a testing framework​

Each Temporal SDK has a testing suite that can be used in conjunction with a typical language specific testing framework. In the Temporal Python SDK, the testing package (https://python.temporal.io/temporalio.testing.html) provides a test environment in which the Workflow and Activity code may be run for test purposes.

The BackgroundCheck Workflow code checks the following conditions:

  1. It receives a social security number and a unique ID as input parameters.
  2. It starts a new Activity ssn_trace_activity with the input SSN.
  3. It waits for the Activity to complete and returns the result.
  4. If the Activity returns "pass", it logs a message indicating that the background check passed.
  5. If the Activity returns "fail", it raises an exception indicating that the background check failed.

We can also perform a Workflow Replay test, and we'll provide detailed coverage of this topic in another section.

Add Workflow function tests​

This is a unit test written in Python using the pytest library.

The test checks the execute_workflow method of the BackgroundCheck Workflow.

The test creates a new WorkflowEnvironment and a Worker with a Task Queue and the BackgroundCheck Workflow and ssn_trace_activity activity.

Then, it executes the BackgroundCheck.run method with a social security number and a unique ID, and asserts that the result is equal to "pass".

The test is marked with @pytest.mark.asyncio to indicate that it is an asynchronous test.

docs/tutorials/python/background-check/code/backgroundcheck_boilerplate/tests/workflow_test.py

import uuid

import pytest

from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

from activities.ssntraceactivity_dacx import ssn_trace_activity
from workflows.backgroundcheck_dacx import BackgroundCheck

@pytest.mark.asyncio
async def test_execute_workflow():
    task_queue_name = str(uuid.uuid4())
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client,
            task_queue=task_queue_name,
            workflows=[BackgroundCheck],
            activities=[ssn_trace_activity],
        ):
            assert "pass" == await env.client.execute_workflow(
                BackgroundCheck.run,
                "555-55-5555",
                id=str(uuid.uuid4()),
                task_queue=task_queue_name,
            )

Add Activity function tests​

This is a Python using the pytest framework and the ActivityEnvironment class from the Temporal Python SDK. It tests the ssn_trace_activity function from the activities module. The function takes a social security number as input and returns a string indicating whether the SSN is valid or not. The test checks if the function returns "pass" when given the SSN "55-55-555".

docs/tutorials/python/background-check/code/backgroundcheck_boilerplate/tests/activity_test.py

import pytest
from temporalio.testing import ActivityEnvironment
from activities.ssntraceactivity_dacx import ssn_trace_activity

@pytest.mark.asyncio
async def test_ssn_trace_activity() -> str:
    activity_environment = ActivityEnvironment()
    expected_output = "pass"
    assert expected_output == await activity_environment.run(
        ssn_trace_activity, "55-55-555"
    )