Python SDK Data Handling

Overview

The Python SDK uses data converters to serialize/deserialize workflow inputs, outputs, and activity parameters.

Default Data Converter

The default converter handles:

None
bytes (as binary)
Protobuf messages
JSON-serializable types (dict, list, str, int, float, bool)

Pydantic Integration

Use Pydantic models for validated, typed data.

In your workflow definition, just use input and result types that subclass pydantic.BaseModel:

from pydantic import BaseModel

class OrderInput(BaseModel):
    order_id: str
    items: list[str]
    total: float
    customer_email: str

class OrderResult(BaseModel):
    order_id: str
    status: str
    tracking_number: str | None = None

@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, input: OrderInput) -> OrderResult:
        # Pydantic validation happens automatically
        return OrderResult(
            order_id=input.order_id,
            status="completed",
            tracking_number="TRK123",
        )

And when you configure the client, pass the pydantic_data_converter:

from temporalio.contrib.pydantic import pydantic_data_converter
# Configure client with Pydantic support
client = await Client.connect(
    "localhost:7233",
    namespace="default",
    data_converter=pydantic_data_converter,
)

Custom Data Conversion

Usually the easiest way to do this is via implementing an EncodingPayloadConverter and CompositePayloadConverter. See:

for an extended example.

Payload Encryption

Encrypt sensitive workflow data.

from temporalio.converter import PayloadCodec
from temporalio.api.common.v1 import Payload
from cryptography.fernet import Fernet
from typing import Sequence

class EncryptionCodec(PayloadCodec):
    def __init__(self, key: bytes):
        self._fernet = Fernet(key)

    async def encode(self, payloads: Sequence[Payload]) -> list[Payload]:
        return [
            Payload(
                metadata={"encoding": b"binary/encrypted"},
                # Since encryption uses C extensions that give up the GIL, we can avoid blocking the async event loop here.
                data=await asyncio.to_thread(self._fernet.encrypt, p.SerializeToString()),
            )
            for p in payloads
        ]

    async def decode(self, payloads: Sequence[Payload]) -> list[Payload]:
        result = []
        for p in payloads:
            if p.metadata.get("encoding") == b"binary/encrypted":
                decrypted = await asyncio.to_thread(self._fernet.decrypt, p.data)
                decoded = Payload()
                decoded.ParseFromString(decrypted)
                result.append(decoded)
            else:
                result.append(p)
        return result

# Apply encryption codec
client = await Client.connect(
    "localhost:7233",
    namespace="default",
    data_converter=DataConverter(
        payload_codec=EncryptionCodec(encryption_key),
    ),
)

Search Attributes

Custom searchable fields for workflow visibility. These can be created at workflow start:

from temporalio.common import (
    SearchAttributeKey,
    SearchAttributePair,
    TypedSearchAttributes,
)
from datetime import datetime
from datetime import timezone

ORDER_ID = SearchAttributeKey.for_keyword("OrderId")
ORDER_STATUS = SearchAttributeKey.for_keyword("OrderStatus")
ORDER_TOTAL = SearchAttributeKey.for_float("OrderTotal")
CREATED_AT = SearchAttributeKey.for_datetime("CreatedAt")

# At workflow start
handle = await client.start_workflow(
    OrderWorkflow.run,
    order,
    id=f"order-{order.id}",
    task_queue="orders",
    search_attributes=TypedSearchAttributes([
        SearchAttributePair(ORDER_ID, order.id),
        SearchAttributePair(ORDER_STATUS, "pending"),
        SearchAttributePair(ORDER_TOTAL, order.total),
        SearchAttributePair(CREATED_AT, datetime.now(timezone.utc)),
    ]),
)

Or upserted during workflow execution:

from temporalio import workflow
from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes

ORDER_STATUS = SearchAttributeKey.for_keyword("OrderStatus")

@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order: Order) -> str:
        # ... process order ...

        # Update search attribute
        workflow.upsert_search_attributes(TypedSearchAttributes([
            SearchAttributePair(ORDER_STATUS, "completed"),
        ]))
        return "done"

Querying Workflows by Search Attributes

# List workflows using search attributes
async for workflow in client.list_workflows(
    'OrderStatus = "processing" OR OrderStatus = "pending"'
):
    print(f"Workflow {workflow.id} is still processing")

Workflow Memo

Store arbitrary metadata with workflows (not searchable).

# Set memo at workflow start
await client.execute_workflow(
    OrderWorkflow.run,
    order,
    id=f"order-{order.id}",
    task_queue="orders",
    memo={
        "customer_name": order.customer_name,
        "notes": "Priority customer",
    },
)

# Read memo from workflow
@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order: Order) -> str:
        notes: str = workflow.memo_value("notes", type_hint=str)
        ...

Deterministic APIs for Values

Use these APIs within workflows for deterministic random values and UUIDs:

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        # Deterministic UUID (same on replay)
        unique_id = workflow.uuid4()

        # Deterministic random (same on replay)
        rng = workflow.random()
        value = rng.randint(1, 100)

        return str(unique_id)

Best Practices

Use Pydantic for input/output validation
Keep payloads small—see references/core/gotchas.md for limits
Encrypt sensitive data with PayloadCodec
Use dataclasses for simple data structures
Use workflow.uuid4() and workflow.random() for deterministic values

For External Storage drivers that need the owning Workflow or Standalone-Activity identity, see references/python/serialization-context.md.

6.3 KiB Raw Permalink Blame History