This sample shows how to make a custom payload converter for a type not natively supported by Temporal.

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

worker:

poetry run python worker.py

This will start the worker. Then, in another terminal, run the following to execute the workflow:

poetry run python starter.py

The workflow should complete with the hello result. If the custom converter was not set for the custom input and output

classes, we would get an error on the client side and on the worker side.

import asyncio

import dataclasses

import temporalio.converter

from temporalio.client import Client

from custom_converter.worker import (

GreetingInput,

GreetingOutput,

GreetingPayloadConverter,

GreetingWorkflow,

)

async def main():

# Connect client

client = await Client.connect(

"localhost:7233",

# Use the default data converter, but change the payload converter.

# Without this we get:

# TypeError: Object of type GreetingInput is not JSON serializable

data_converter=dataclasses.replace(

temporalio.converter.default(),

payload_converter_class=GreetingPayloadConverter,

),

)

# Run workflow

result = await client.execute_workflow(

GreetingWorkflow.run,

GreetingInput("Temporal"),

id=f"custom_converter-workflow-id",

task_queue="custom_converter-task-queue",

)

assert isinstance(result, GreetingOutput)

print(f"Workflow result: {result.result}")

if __name__ == "__main__":

asyncio.run(main())

import asyncio

import dataclasses

from typing import Any, Optional, Type

import temporalio.converter

from temporalio import workflow

from temporalio.api.common.v1 import Payload

from temporalio.client import Client

from temporalio.converter import (

CompositePayloadConverter,

DefaultPayloadConverter,

EncodingPayloadConverter,

)

from temporalio.worker import Worker

class GreetingInput:

def __init__(self, name: str) -> None:

self.name = name

class GreetingOutput:

def __init__(self, result: str) -> None:

self.result = result

class GreetingWorkflow:

@workflow.run

async def run(self, input: GreetingInput) -> GreetingOutput:

return GreetingOutput(f"Hello, {input.name}")

class GreetingEncodingPayloadConverter(EncodingPayloadConverter):

@property

def encoding(self) -> str:

return "text/my-greeting-encoding"

def to_payload(self, value: Any) -> Optional[Payload]:

if isinstance(value, GreetingInput):

return Payload(

metadata={"encoding": self.encoding.encode(), "is_input": b"true"},

data=value.name.encode(),

)

elif isinstance(value, GreetingOutput):

return Payload(

metadata={"encoding": self.encoding.encode()},

data=value.result.encode(),

)

else:

return None

def from_payload(self, payload: Payload, type_hint: Optional[Type] = None) -> Any:

if payload.metadata.get("is_input") == b"true":

# Confirm proper type hint if present

assert not type_hint or type_hint is GreetingInput

return GreetingInput(payload.data.decode())

else:

assert not type_hint or type_hint is GreetingOutput

return GreetingOutput(payload.data.decode())

class GreetingPayloadConverter(CompositePayloadConverter):

def __init__(self) -> None:

# Just add ours as first before the defaults

super().__init__(

GreetingEncodingPayloadConverter(),

# TODO(cretz): Make this list available without instantiation - https://github.com/temporalio/sdk-python/issues/139

*DefaultPayloadConverter().converters.values(),

)

interrupt_event = asyncio.Event()

async def main():

# Connect client

client = await Client.connect(

"localhost:7233",

# Use the default data converter, but change the payload converter.

# Without this, when trying to run a workflow, we get:

# KeyError: 'Unknown payload encoding my-greeting-encoding

data_converter=dataclasses.replace(

temporalio.converter.default(),

payload_converter_class=GreetingPayloadConverter,

),

)

# Run a worker for the workflow

async with Worker(

client,

task_queue="custom_converter-task-queue",

workflows=[GreetingWorkflow],

):

# Wait until interrupted

print("Worker started, ctrl+c to exit")

await interrupt_event.wait()

print("Shutting down")

if __name__ == "__main__":

loop = asyncio.new_event_loop()

try:

loop.run_until_complete(main())

except KeyboardInterrupt:

interrupt_event.set()

loop.run_until_complete(loop.shutdown_asyncgens())

This sample shows a custom decorator can help with Temporal code reuse. Specifically, this makes a `@auto_heartbeater`

decorator that automatically configures an activity to heartbeat twice as frequently as the heartbeat timeout is set to.

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

worker:

poetry run python worker.py

This will start the worker. Then, in another terminal, run the following to execute the workflow:

poetry run python starter.py

The workflow will be started, and then after 5 seconds will be sent a signal to cancel its forever-running activity.

The activity has a heartbeat timeout set to 2s, so since it has the `@auto_heartbeater` decorator set, it will heartbeat

every second. If this was not set, the workflow would fail with an activity heartbeat timeout failure.

import asyncio

from datetime import datetime

from functools import wraps

from typing import Any, Awaitable, Callable, TypeVar, cast

from temporalio import activity

F = TypeVar("F", bound=Callable[..., Awaitable[Any]])

def auto_heartbeater(fn: F) -> F:

# We want to ensure that the type hints from the original callable are

# available via our wrapper, so we use the functools wraps decorator

@wraps(fn)

async def wrapper(*args, **kwargs):

done = asyncio.Event()

# Heartbeat twice as often as the timeout

heartbeat_timeout = activity.info().heartbeat_timeout

if heartbeat_timeout:

asyncio.create_task(

heartbeat_every(heartbeat_timeout.total_seconds() / 2, done)

)

try:

return await fn(*args, **kwargs)

finally:

done.set()

return cast(F, wrapper)

async def heartbeat_every(

delay: float, done_event: asyncio.Event, *details: Any

) -> None:

# Heartbeat every so often while not cancelled

while not done_event.is_set():

try:

await asyncio.wait_for(done_event.wait(), delay)

except asyncio.TimeoutError:

print(f"Heartbeating at {datetime.now()}")

activity.heartbeat(*details)

import asyncio

from temporalio.client import Client

from custom_decorator.worker import WaitForCancelWorkflow

async def main():

# Connect client

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

# Start the workflow

handle = await client.start_workflow(

WaitForCancelWorkflow.run,

id=f"custom_decorator-workflow-id",

task_queue="custom_decorator-task-queue",

)

print("Started workflow, waiting 5 seconds before cancelling")

await asyncio.sleep(5)

# Send a signal asking workflow to cancel the activity

await handle.signal(WaitForCancelWorkflow.cancel_activity)

# Wait and expect to be told about the activity being cancelled. If we did

# not have the automatic heartbeater decorator, the signal would have failed

# because the workflow would already be completed as failed with activity

# heartbeat timeout.

result = await handle.result()

print(f"Result: {result}")

if __name__ == "__main__":

asyncio.run(main())

import asyncio

from datetime import timedelta

from temporalio import activity, workflow

from temporalio.client import Client

from temporalio.common import RetryPolicy

from temporalio.worker import Worker

from custom_decorator.activity_utils import auto_heartbeater

async def wait_for_cancel_activity() -> str:

# Wait forever, catch the cancel, and return some value

try:

await asyncio.Future()

raise RuntimeError("unreachable")

except asyncio.CancelledError:

return "activity cancelled!"

class WaitForCancelWorkflow:

@workflow.run

async def run(self) -> str:

# Start activity and wait on it (it will get cancelled from signal)

self.activity = workflow.start_activity(

wait_for_cancel_activity,

start_to_close_timeout=timedelta(hours=20),

# We set a heartbeat timeout so Temporal knows if the activity

# failed/crashed. If we don't heartbeat within this time, Temporal

# will consider the activity failed.

heartbeat_timeout=timedelta(seconds=2),

# Tell the activity not to retry for demonstration purposes only

retry_policy=RetryPolicy(maximum_attempts=1),

# Tell the workflow to wait for the post-cancel result

cancellation_type=workflow.ActivityCancellationType.WAIT_CANCELLATION_COMPLETED,

)

return await self.activity

@workflow.signal

def cancel_activity(self) -> None:

self.activity.cancel()

interrupt_event = asyncio.Event()

async def main():

# Connect client

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

# Run a worker for the workflow

async with Worker(

client,

task_queue="custom_decorator-task-queue",

workflows=[WaitForCancelWorkflow],

activities=[wait_for_cancel_activity],

):

# Wait until interrupted

print("Worker started, ctrl+c to exit")

await interrupt_event.wait()

print("Shutting down")

if __name__ == "__main__":

loop = asyncio.new_event_loop()

try:

loop.run_until_complete(main())

except KeyboardInterrupt:

interrupt_event.set()

loop.run_until_complete(loop.shutdown_asyncgens())