Skip to main content

Foundations - Python SDK feature guide

The Foundations section of the Temporal Developer's guide covers the minimum set of concepts and implementation details needed to build and run a Temporal Application—that is, all the relevant steps to start a Workflow Execution that executes an Activity.

In this section you can find the following:

How to install Temporal CLI and run a development server​

This section describes how to install the Temporal CLI and run a development Cluster. The local development Cluster comes packaged with the Temporal Web UI.

For information on deploying and running a self-hosted production Cluster, see the Self-hosted guide, or sign up for Temporal Cloud and let us run your production Cluster for you.

Temporal CLI is a tool for interacting with a Temporal Cluster from the command line and it includes a distribution of the Temporal Server and Web UI. This local development Cluster runs as a single process with zero runtime dependencies and it supports persistence to disk and in-memory mode through SQLite.

Install the Temporal CLI

Choose one of the following install methods to install the Temporal CLI.

  • Install the Temporal CLI with Homebrew.

    brew install temporal

  • Install the Temporal CLI with cURL.

    curl -sSf https://temporal.download/cli.sh | sh

  • Install the Temporal CLI from CDN.

    1. Select the platform and architecture needed.
    2. Extract the downloaded archive.
    3. Add the temporal binary to your PATH.

Start the Temporal Development Server

Start the Temporal Development Server by using the server start-dev command.

temporal server start-dev

This command automatically starts the Web UI, creates the default Namespace, and uses an in-memory database.

The Temporal Server should be available on localhost:7233 and the Temporal Web UI should be accessible at http://localhost:8233.

The server's startup configuration can be customized using command line options. For a full list of options, run:

temporal server start-dev --help

How to install a Temporal SDK​

A Temporal SDK provides a framework for Temporal Application development.

An SDK provides you with the following:

Python 3.7+ PyPI

To install the latest version of the Temporal Python package, run the following command.

pip install temporalio

How to find the Python SDK API reference​

The Temporal Python SDK API reference is published on python.temporal.io.

Where are SDK-specific code examples?​

You can find a complete list of executable code samples in Temporal's GitHub repository.

Additionally, several of the Tutorials are backed by a fully executable template application.

How to connect a Temporal Client to a Temporal Cluster​

A Temporal Client enables you to communicate with the Temporal Cluster. Communication with a Temporal Cluster includes, but isn't limited to, the following:

  • Starting Workflow Executions.
  • Sending Signals to Workflow Executions.
  • Sending Queries to Workflow Executions.
  • Getting the results of a Workflow Execution.
  • Providing an Activity Task Token.
caution

A Temporal Client cannot be initialized and used inside a Workflow. However, it is acceptable and common to use a Temporal Client inside an Activity to communicate with a Temporal Cluster.

When you are running a Cluster locally (such as the Temporal CLI), the number of connection options you must provide is minimal. Many SDKs default to the local host or IP address and port that Temporalite and Docker Compose serve (127.0.0.1:7233).

Use the connect() method on the Client class to create and connect to a Temporal Client to the Temporal Cluster.

View the source code

in the context of the rest of the application code.


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

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your name",
        id="your-workflow-id",
        task_queue="your-task-queue",
    )

    print(f"Result: {result}")


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

How to connect to Temporal Cloud​

When you connect to Temporal Cloud, you need to provide additional connection and client options that include the following:

For more information about managing and generating client certificates for Temporal Cloud, see How to manage certificates in Temporal Cloud.

For more information about configuring TLS to secure inter- and intra-network communication for a Temporal Cluster, see Temporal Customization Samples.

Use the connect() method on the Client class to create and connect to a Temporal Client to the Temporal Cluster. Then specify the TLSConfig arguments to connect to a Temporal Cluster with TLS enabled. The client_cert must be combined with client_private_key to authenticate the Client.

View the source code

in the context of the rest of the application code.


from temporalio.client import Client, TLSConfig
# ...
# ...
async def main():
    with open("client-cert.pem", "rb") as f:
        client_cert = f.read()
    with open("client-private-key.pem", "rb") as f:
        client_private_key = f.read()
    client = await Client.connect(
        "your-custom-namespace.tmprl.cloud:7233",
        namespace="<your-custom-namespace>.<id>",
        tls=TLSConfig(
            client_cert=client_cert,
            client_private_key=client_private_key,
            # domain=domain, # TLS domain
            # server_root_ca_cert=server_root_ca_cert, # ROOT CA to validate the server cert
        ),
    )

How to develop a basic Workflow​

Workflows are the fundamental unit of a Temporal Application, and it all starts with the development of a Workflow Definition.

In the Temporal Python SDK programming model, Workflows are defined as classes.

Specify the @workflow.defn decorator on the Workflow 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.

View the source code

in the context of the rest of the application code.


from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

How to define Workflow parameters​

Temporal Workflows may have any number of custom parameters. However, we strongly recommend that objects are used as parameters, so that the object's individual fields may be altered without breaking the signature of the Workflow. All Workflow Definition parameters must be serializable.

Workflow parameters are the method parameters of the singular method decorated with @workflow.run. These can be any data type Temporal can convert, including dataclasses when properly type-annotated. Technically this can be multiple parameters, but Temporal strongly encourages a single dataclass parameter containing all input fields.

View the source code

in the context of the rest of the application code.


from dataclasses import dataclass
# ...
# ...
@dataclass
class YourParams:
    greeting: str
    name: str

How to define Workflow return parameters​

Workflow return values must also be serializable. Returning results, returning errors, or throwing exceptions is fairly idiomatic in each language that is supported. However, Temporal APIs that must be used to get the result of a Workflow Execution will only ever receive one of either the result or the error.

To return a value of the Workflow, use return to return an object.

To return the results of a Workflow Execution, use either start_workflow() or execute_workflow() asynchronous methods.

View the source code

in the context of the rest of the application code.


from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

How to customize your Workflow Type​

Workflows have a Type that are referred to as the Workflow name.

The following examples demonstrate how to set a custom name for your Workflow Type.

You can customize the Workflow name with a custom name in the decorator argument. For example, @workflow.defn(name="your-workflow-name"). If the name parameter is not specified, the Workflow name defaults to the function name.

View the source code

in the context of the rest of the application code.


from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

How develop Workflow logic​

Workflow logic is constrained by deterministic execution requirements. Therefore, each language is limited to the use of certain idiomatic techniques. However, each Temporal SDK provides a set of APIs that can be used inside your Workflow to interact with external (to the Workflow) application code.

Workflow code must be deterministic. This means:

  • no threading
  • no randomness
  • no external calls to processes
  • no network I/O
  • no global state mutation
  • no system date or time

All API safe for Workflows used in the temporalio.workflow must run in the implicit asyncio event loop and be deterministic.

How to develop a basic Activity​

One of the primary things that Workflows do is orchestrate the execution of Activities. An Activity is a normal function or method execution that's intended to execute a single, well-defined action (either short or long-running), such as querying a database, calling a third-party API, or transcoding a media file. An Activity can interact with world outside the Temporal Platform or use a Temporal Client to interact with a Cluster. For the Workflow to be able to execute the Activity, we must define the Activity Definition.

You can develop an Activity Definition by using the @activity.defn decorator. Register the function as an Activity with a custom name through a decorator argument, for example @activity.defn(name="your_activity").

note

The Temporal Python SDK supports multiple ways of implementing an Activity:

Blocking the async event loop in Python would turn your asynchronous program into a synchronous program that executes serially, defeating the entire purpose of using asyncio. This can also lead to potential deadlock, and unpredictable behavior that causes tasks to be unable to execute. Debugging these issues can be difficult and time consuming, as locating the source of the blocking call might not always be immediately obvious.

Due to this, consider not make blocking calls from within an asynchronous Activity, or use an async safe library to perform these actions. If you must use a blocking library, consider using a synchronous Activity instead.

View the source code

in the context of the rest of the application code.


from temporalio import activity
# ...
# ...
@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"

How to develop Activity Parameters​

There is no explicit limit to the total number of parameters that an Activity Definition may support. However, there is a limit to the total size of the data that ends up encoded into a gRPC message Payload.

A single argument is limited to a maximum size of 2 MB. And the total size of a gRPC message, which includes all the arguments, is limited to a maximum of 4 MB.

Also, keep in mind that all Payload data is recorded in the Workflow Execution Event History and large Event Histories can affect Worker performance. This is because the entire Event History could be transferred to a Worker Process with a Workflow Task.

Some SDKs require that you pass context objects, others do not. When it comes to your application data—that is, data that is serialized and encoded into a Payload—we recommend that you use a single object as an argument that wraps the application data passed to Activities. This is so that you can change what data is passed to the Activity without breaking a function or method signature.

Activity parameters are the function parameters of the function decorated with @activity.defn. These can be any data type Temporal can convert, including dataclasses when properly type-annotated. Technically this can be multiple parameters, but Temporal strongly encourages a single dataclass parameter containing all input fields.

View the source code

in the context of the rest of the application code.


from temporalio import activity
from your_dataobject_dacx import YourParams

# ...
# ...
@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"

How to define Activity return values​

All data returned from an Activity must be serializable.

There is no explicit limit to the amount of data that can be returned by an Activity, but keep in mind that all return values are recorded in a Workflow Execution Event History.

An Activity Execution can return inputs and other Activity values.

The following example defines an Activity that takes a string as input and returns a string.

View the source code

in the context of the rest of the application code.


# ...
@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"

How to customize your Activity Type​

Activities have a Type that are referred to as the Activity name. The following examples demonstrate how to set a custom name for your Activity Type.

You can customize the Activity name with a custom name in the decorator argument. For example, @activity.defn(name="your-activity"). If the name parameter is not specified, the Activity name defaults to the function name.

View the source code

in the context of the rest of the application code.


# ...
@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"

How to start an Activity Execution​

Calls to spawn Activity Executions are written within a Workflow Definition. The call to spawn an Activity Execution generates the ScheduleActivityTask Command. This results in the set of three Activity Task related Events (ActivityTaskScheduled, ActivityTaskStarted, and ActivityTask[Closed])in your Workflow Execution Event History.

A single instance of the Activities implementation is shared across multiple simultaneous Activity invocations. Activity implementation code should be idempotent.

The values passed to Activities through invocation parameters or returned through a result value are recorded in the Execution history. The entire Execution history is transferred from the Temporal service to Workflow Workers when a Workflow state needs to recover. A large Execution history can thus adversely impact the performance of your Workflow.

Therefore, be mindful of the amount of data you transfer through Activity invocation parameters or Return Values. Otherwise, no additional limitations exist on Activity implementations.

To spawn an Activity Execution, use the execute_activity() operation from within your Workflow Definition.

execute_activity() is a shortcut for start_activity() that waits on its result.

To get just the handle to wait and cancel separately, use start_activity(). In most cases, use execute_activity() unless advanced task capabilities are needed.

A single argument to the Activity is positional. Multiple arguments are not supported in the type-safe form of start_activity() or execute_activity() and must be supplied by the args keyword argument.

View the source code

in the context of the rest of the application code.


from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

How to set the required Activity Timeouts​

Activity Execution semantics rely on several parameters. The only required value that needs to be set is either a Schedule-To-Close Timeout or a Start-To-Close Timeout. These values are set in the Activity Options.

Activity options are set as keyword arguments after the Activity arguments.

Available timeouts are:

  • schedule_to_close_timeout
  • schedule_to_start_timeout
  • start_to_close_timeout

View the source code

in the context of the rest of the application code.


# ...
        activity_timeout_result = await workflow.execute_activity(
            your_activity,
            YourParams(greeting, "Activity Timeout option"),
            # Activity Execution Timeout
            start_to_close_timeout=timedelta(seconds=10),
            # schedule_to_start_timeout=timedelta(seconds=10),
            # schedule_to_close_timeout=timedelta(seconds=10),
        )

How to get the results of an Activity Execution​

The call to spawn an Activity Execution generates the ScheduleActivityTask Command and provides the Workflow with an Awaitable. Workflow Executions can either block progress until the result is available through the Awaitable or continue progressing, making use of the result when it becomes available.

Use start_activity() to start an Activity and return its handle, ActivityHandle. Use execute_activity() to return the results.

You must provide either schedule_to_close_timeout or start_to_close_timeout.

execute_activity() is a shortcut for await start_activity(). An asynchronous execute_activity() helper is provided which takes the same arguments as start_activity() and awaits on the result. execute_activity() should be used in most cases unless advanced task capabilities are needed.

View the source code

in the context of the rest of the application code.


from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

How to run Worker Processes​

The Worker Process is where Workflow Functions and Activity Functions are executed.

  • Each Worker Entity in the Worker Process must register the exact Workflow Types and Activity Types it may execute.
  • Each Worker Entity must also associate itself with exactly one Task Queue.
  • Each Worker Entity polling the same Task Queue must be registered with the same Workflow Types and Activity Types.

A Worker Entity is the component within a Worker Process that listens to a specific Task Queue.

Although multiple Worker Entities can be in a single Worker Process, a single Worker Entity Worker Process may be perfectly sufficient. For more information, see the Worker tuning guide.

A Worker Entity contains a Workflow Worker and/or an Activity Worker, which makes progress on Workflow Executions and Activity Executions, respectively.

To develop a Worker, use the Worker() constructor and add your Client, Task Queue, Workflows, and Activities as arguments. The following code example creates a Worker that polls for tasks from the Task Queue and executes the Workflow. When a Worker is created, it accepts a list of Workflows in the workflows parameter, a list of Activities in the activities parameter, or both.

View the source code

in the context of the rest of the application code.


from temporalio.client import Client
from temporalio.worker import Worker
# ...
# ...
async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="your-task-queue",
        workflows=[YourWorkflow],
        activities=[your_activity],
    )
    await worker.run()


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

How to register types​

All Workers listening to the same Task Queue name must be registered to handle the exact same Workflows Types and Activity Types.

If a Worker polls a Task for a Workflow Type or Activity Type it does not know about, it fails that Task. However, the failure of the Task does not cause the associated Workflow Execution to fail.

When a Worker is created, it accepts a list of Workflows in the workflows parameter, a list of Activities in the activities parameter, or both.

View the source code

in the context of the rest of the application code.


# ...
async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="your-task-queue",
        workflows=[YourWorkflow],
        activities=[your_activity],
    )
    await worker.run()


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

How to start a Workflow Execution​

Workflow Execution semantics rely on several parameters—that is, to start a Workflow Execution you must supply a Task Queue that will be used for the Tasks (one that a Worker is polling), the Workflow Type, language-specific contextual data, and Workflow Function parameters.

In the examples below, all Workflow Executions are started using a Temporal Client. To spawn Workflow Executions from within another Workflow Execution, use either the Child Workflow or External Workflow APIs.

See the Customize Workflow Type section to see how to customize the name of the Workflow Type.

A request to spawn a Workflow Execution causes the Temporal Cluster to create the first Event (WorkflowExecutionStarted) in the Workflow Execution Event History. The Temporal Cluster then creates the first Workflow Task, resulting in the first WorkflowTaskScheduled Event.

To start a Workflow Execution in Python, use either the start_workflow() or execute_workflow() asynchronous methods in the Client.

View the source code

in the context of the rest of the application code.


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

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your name",
        id="your-workflow-id",
        task_queue="your-task-queue",
    )

    print(f"Result: {result}")


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

How to set a Workflow's Task Queue​

In most SDKs, the only Workflow Option that must be set is the name of the Task Queue.

For any code to execute, a Worker Process must be running that contains a Worker Entity that is polling the same Task Queue name.

To set a Task Queue in Python, specify the task_queue argument when executing a Workflow with either start_workflow() or execute_workflow() methods.

View the source code

in the context of the rest of the application code.


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

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your name",
        id="your-workflow-id",
        task_queue="your-task-queue",
    )

    print(f"Result: {result}")


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

How to set a Workflow Id​

You must set a Workflow Id.

When setting a Workflow Id, we recommended mapping it to a business process or business entity identifier, such as an order identifier or customer identifier.

To set a Workflow Id in Python, specify the id argument when executing a Workflow with either start_workflow() or execute_workflow() methods.

The id argument should be a unique identifier for the Workflow Execution.

View the source code

in the context of the rest of the application code.


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

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your name",
        id="your-workflow-id",
        task_queue="your-task-queue",
    )

    print(f"Result: {result}")


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

How to get the results of a Workflow Execution​

If the call to start a Workflow Execution is successful, you will gain access to the Workflow Execution's Run Id.

The Workflow Id, Run Id, and Namespace may be used to uniquely identify a Workflow Execution in the system and get its result.

It's possible to both block progress on the result (synchronous execution) or get the result at some other point in time (asynchronous execution).

In the Temporal Platform, it's also acceptable to use Queries as the preferred method for accessing the state and results of Workflow Executions.

Use start_workflow() or get_workflow_handle() to return a Workflow handle. Then use the result method to await on the result of the Workflow.

To get a handle for an existing Workflow by its Id, you can use get_workflow_handle(), or use get_workflow_handle_for() for type safety.

Then use describe() to get the current status of the Workflow. If the Workflow does not exist, this call fails.

View the source code

in the context of the rest of the application code.


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

    handle = client.get_workflow_handle(
        workflow_id="your-workflow-id",
    )
    results = await handle.result()
    print(f"Result: {results}")


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