client = await Client.connect("localhost:7233")

These samples show basic workflow and activity features.

To run, first see [README.md](../README.md) for prerequisites. Then, run the following from this directory to run the

`hello_activity.py` sample:

poetry run python hello_activity.py

The result will be:

Result: Hello, World!

Replace `hello_activity.py` in the command with any other example filename to run it instead.

* [hello_activity](hello_activity.py) - Execute an activity from a workflow.

* [hello_activity_choice](hello_activity_choice.py) - Execute certain activities inside a workflow based on dynamic

input.

* [hello_activity_multiprocess](hello_activity_multiprocess.py) - Execute a synchronous activity on a process pool.

* [hello_activity_retry](hello_activity_retry.py) - Demonstrate activity retry by failing until a certain number of

attempts.

* [hello_activity_threaded](hello_activity_threaded.py) - Execute a synchronous activity on a thread pool.

* [hello_async_activity_completion](hello_async_activity_completion.py) - Complete an activity outside of the function

that was called.

* [hello_cancellation](hello_cancellation.py) - Manually react to cancellation inside workflows and activities.

* [hello_child_workflow](hello_child_workflow.py) - Execute a child workflow from a workflow.

* [hello_continue_as_new](hello_continue_as_new.py) - Use continue as new to restart a workflow.

* [hello_cron](hello_cron.py) - Execute a workflow once a minute.

* [hello_exception](hello_exception.py) - Execute an activity that raises an error out of the workflow and out of the

program.

* [hello_local_activity](hello_local_activity.py) - Execute a local activity from a workflow.

* [hello_mtls](hello_mtls.py) - Accept URL, namespace, and certificate info as CLI args and use mTLS for connecting to

server.

* [hello_parallel_activity](hello_parallel_activity.py) - Execute multiple activities at once.

* [hello_query](hello_query.py) - Invoke queries on a workflow.

* [hello_search_attributes](hello_search_attributes.py) - Start workflow with search attributes then change while

running.

* [hello_signal](hello_signal.py) - Send signals to a workflow.

from dataclasses import dataclass

class ComposeGreetingInput:

greeting: str

name: str

async def compose_greeting(input: ComposeGreetingInput) -> str:

activity.logger.info("Running activity with parameter %s" % input)

return f"{input.greeting}, {input.name}!"

class GreetingWorkflow:

compose_greeting,

ComposeGreetingInput("Hello", name),

start_to_close_timeout=timedelta(seconds=10),

client = await Client.connect("localhost:7233")

task_queue="hello-activity-task-queue",

workflows=[GreetingWorkflow],

activities=[compose_greeting],

GreetingWorkflow.run,

"World",

id="hello-activity-workflow-id",

task_queue="hello-activity-task-queue",

import asyncio

from dataclasses import dataclass

from datetime import timedelta

from enum import IntEnum

from typing import List

from temporalio import activity, workflow

from temporalio.client import Client

from temporalio.worker import Worker

async def order_apples(amount: int) -> str:

return f"Ordered {amount} Apples..."

async def order_bananas(amount: int) -> str:

return f"Ordered {amount} Bananas..."

async def order_cherries(amount: int) -> str:

return f"Ordered {amount} Cherries..."

async def order_oranges(amount: int) -> str:

return f"Ordered {amount} Oranges..."

class Fruit(IntEnum):

APPLE = 1

BANANA = 2

CHERRY = 3

ORANGE = 4

class ShoppingListItem:

fruit: Fruit

amount: int

class ShoppingList:

items: List[ShoppingListItem]

class PurchaseFruitsWorkflow:

@workflow.run

async def run(self, list: ShoppingList) -> str:

# Order each thing on the list

ordered: List[str] = []

for item in list.items:

if item.fruit is Fruit.APPLE:

ordered.append(

await workflow.execute_activity(

order_apples,

item.amount,

start_to_close_timeout=timedelta(seconds=5),

)

)

elif item.fruit is Fruit.BANANA:

ordered.append(

await workflow.execute_activity(

order_bananas,

item.amount,

start_to_close_timeout=timedelta(seconds=5),

)

)

elif item.fruit is Fruit.CHERRY:

ordered.append(

await workflow.execute_activity(

order_cherries,

item.amount,

start_to_close_timeout=timedelta(seconds=5),

)

)

elif item.fruit is Fruit.ORANGE:

ordered.append(

await workflow.execute_activity(

order_oranges,

item.amount,

start_to_close_timeout=timedelta(seconds=5),

)

)

else:

raise ValueError(f"Unrecognized fruit: {item.fruit}")

return "".join(ordered)

async def main():

# Start client

client = await Client.connect("localhost:7233")

# Run a worker for the workflow

async with Worker(

client,

task_queue="hello-activity-choice-task-queue",

workflows=[PurchaseFruitsWorkflow],

activities=[order_apples, order_bananas, order_cherries, order_oranges],

):

# While the worker is running, use the client to run the workflow and

# print out its result. Note, in many production setups, the client

# would be in a completely separate process from the worker.

result = await client.execute_workflow(

PurchaseFruitsWorkflow.run,

ShoppingList(

[

ShoppingListItem(Fruit.APPLE, 8),

ShoppingListItem(Fruit.BANANA, 5),

ShoppingListItem(Fruit.CHERRY, 1),

ShoppingListItem(Fruit.ORANGE, 4),

]

),

id="hello-activity-choice-workflow-id",

task_queue="hello-activity-choice-task-queue",

)

print(f"Order result: {result}")

if __name__ == "__main__":

asyncio.run(main())

import asyncio

import multiprocessing

import os

import time

from concurrent.futures import ProcessPoolExecutor

from dataclasses import dataclass

from datetime import timedelta

from temporalio import activity, workflow

from temporalio.client import Client

from temporalio.worker import SharedStateManager, Worker

class ComposeGreetingInput:

greeting: str

name: str

def compose_greeting(input: ComposeGreetingInput) -> str:

# We'll wait for 3 seconds, heartbeating in between (like all long-running

# activities should do), then return the greeting

for _ in range(0, 3):

print(f"Heartbeating activity on PID {os.getpid()}")

activity.heartbeat()

time.sleep(1)

return f"{input.greeting}, {input.name}!"

class GreetingWorkflow:

@workflow.run

async def run(self, name: str) -> str:

return await workflow.execute_activity(

compose_greeting,

ComposeGreetingInput("Hello", name),

start_to_close_timeout=timedelta(seconds=10),

# Always set a heartbeat timeout for long-running activities

heartbeat_timeout=timedelta(seconds=2),

)

async def main():

# Start client

client = await Client.connect("localhost:7233")

# Run a worker for the workflow

async with Worker(

client,

task_queue="hello-activity-multiprocess-task-queue",

workflows=[GreetingWorkflow],

activities=[compose_greeting],

# Synchronous activities are not allowed unless we provide some kind of

# executor. Here we are giving a process pool executor which means the

# activity will actually run in a separate process. This same executor

# could be passed to multiple workers if desired.

activity_executor=ProcessPoolExecutor(5),

# Since we are using an executor that is not a thread pool executor,

# Temporal needs some kind of manager to share state such as

# cancellation info and heartbeat info between the host and the

# activity. Therefore, we must provide a shared_state_manager here. A

# helper is provided to create it from a multiprocessing manager.

shared_state_manager=SharedStateManager.create_from_multiprocessing(

multiprocessing.Manager()

),

):

# While the worker is running, use the client to run the workflow and

# print out its result. Note, in many production setups, the client

# would be in a completely separate process from the worker.

result = await client.execute_workflow(

GreetingWorkflow.run,

"World",

id="hello-activity-multiprocess-workflow-id",

task_queue="hello-activity-multiprocess-task-queue",

)

print(f"Result on PID {os.getpid()}: {result}")

if __name__ == "__main__":

asyncio.run(main())