# ...
@workflow.defn
class ComposeGreetingWorkflow:
    @workflow.run
    async def run(self, input: ComposeGreetingInput) -> str:
        return f"{input.greeting}, {input.name}!"

@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_child_workflow(
            ComposeGreetingWorkflow.run,
            ComposeGreetingInput("Hello", name),
            id="hello-child-workflow-workflow-child-id",
            parent_close_policy=ParentClosePolicy.ABANDON,
        )
```

---

## Continue-As-New - Python SDK

This page answers the following questions for Python developers:

- [What is Continue-As-New?](#what)
- [How to Continue-As-New?](#how)
- [When is it right to Continue-as-New?](#when)
- [How to test Continue-as-New?](#how-to-test)

## What is Continue-As-New? {#what}

[Continue-As-New](/workflow-execution/continue-as-new) lets a Workflow Execution close successfully and creates a new Workflow Execution.
You can think of it as a checkpoint when your Workflow gets too long or approaches certain scaling limits.

The new Workflow Execution is in the same [chain](/workflow-execution#workflow-execution-chain); it keeps the same Workflow Id but gets a new Run Id and a fresh Event History.
It also receives your Workflow's usual parameters.

## How to Continue-As-New using the Python SDK {#how}

First, design your Workflow parameters so that you can pass in the "current state" when you Continue-As-New into the next Workflow run.
This state is typically set to `None` for the original caller of the Workflow.

  
    View the source code
  {' '}
  in the context of the rest of the application code.

```python
@dataclass
class ClusterManagerInput:
    state: Optional[ClusterManagerState] = None
    test_continue_as_new: bool = False

@workflow.run
async def run(self, input: ClusterManagerInput) -> ClusterManagerResult:

````
The test hook in the above snippet is covered [below](#how-to-test).

Inside your Workflow, call the [`continue_as_new()`](https://python.temporal.io/temporalio.workflow.html#continue_as_new) function with the same type.
This stops the Workflow right away and starts a new one.

  
    View the source code
  {' '}
  in the context of the rest of the application code.

```python
workflow.continue_as_new(
  ClusterManagerInput(
      state=self.state,
      test_continue_as_new=input.test_continue_as_new,
  )
)
````

### Considerations for Workflows with Message Handlers {#with-message-handlers}

If you use Updates or Signals, don't call Continue-as-New from the handlers.
Instead, wait for your handlers to finish in your main Workflow before you run `continue_as_new`.
See the [`all_handlers_finished`](message-passing#wait-for-message-handlers) example for guidance.

## When is it right to Continue-as-New using the Python SDK? {#when}

Use Continue-as-New when your Workflow might encounter degraded performance or [Event History Limits](/workflow-execution/event#event-history).

Temporal tracks your Workflow's progress against these limits to let you know when you should Continue-as-New.
Call `workflow.info().is_continue_as_new_suggested()` to check if it's time.

## How to test Continue-as-New using the Python SDK {#how-to-test}

Testing Workflows that naturally Continue-as-New may be time-consuming and resource-intensive.
Instead, add a test hook to check your Workflow's Continue-as-New behavior faster in automated tests.

For example, when `test_continue_as_new == True`, this sample creates a test-only variable called `self.max_history_length` and sets it to a small value.
A helper method in the Workflow checks it each time it considers using Continue-as-New:

  
    View the source code
  {' '}
  in the context of the rest of the application code.

```python
def should_continue_as_new(self) -> bool:
    if workflow.info().is_continue_as_new_suggested():
        return True
    # For testing
    if (
        self.max_history_length
        and workflow.info().get_current_history_length() > self.max_history_length
    ):
        return True
    return False
```

---

## Converters and encryption - Python SDK

Temporal's security model is designed around client-side encryption of Payloads.
A client may encrypt Payloads before sending them to the server, and decrypt them after receiving them from the server.
This provides a high degree of confidentiality because the Temporal Server itself has absolutely no knowledge of the actual data.
It also gives implementers more power and more freedom regarding which client is able to read which data -- they can control access with keys, algorithms, or other security measures.

A Temporal developer adds client-side encryption of Payloads by providing a Custom Payload Codec to its Client.
Depending on business needs, a complete implementation of Payload Encryption may involve selecting appropriate encryption algorithms, managing encryption keys, restricting a subset of their users from viewing payload output, or a combination of these.

The server itself never adds encryption over Payloads.
Therefore, unless client-side encryption is implemented, Payload data will be persisted in non-encrypted form to the data store, and any Client that can make requests to a Temporal namespace (including the Temporal UI and CLI) will be able to read Payloads contained in Workflows.
When working with sensitive data, you should always implement Payload encryption.

## Custom Payload Codec {#custom-payload-codec}

**How to use a custom Payload Codec using Python with the Temporal Python SDK.**

Custom Data Converters can change the default Temporal Data Conversion behavior by adding hooks, sending payloads to external storage, or performing different encoding steps.
If you only need to change the encoding performed on your payloads -- by adding compression or encryption -- you can override the default Data Converter by creating a new `PayloadCodec`.

The `PayloadCodec` needs to implement `encode()` and `decode()` functions at a minimum.
These should loop through all of a Workflow's payloads, perform all of your necessary marshaling, compression, or encryption steps in order, and set an `"encoding"` metadata field.

In this example, the `encode` method marshals and then compresses a payload using Python's [cramjam](https://github.com/milesgranger/cramjam) library to provide `snappy` compression.
The `decode()` function implements the `encode()` logic in reverse:

```python

from temporalio.api.common.v1 import Payload
from temporalio.converter import PayloadCodec

class EncryptionCodec(PayloadCodec):
    async def encode(self, payloads: Iterable[Payload]) -> List[Payload]:
        return [
            Payload(
                metadata={
                    "encoding": b"binary/snappy",
                },
                data=(bytes(cramjam.snappy.compress(p.SerializeToString()))),
            )
            for p in payloads
        ]

    async def decode(self, payloads: Iterable[Payload]) -> List[Payload]:
        ret: List[Payload] = []
        for p in payloads:
            if p.metadata.get("encoding", b"").decode() != "binary/snappy":
                ret.append(p)
                continue
            ret.append(Payload.FromString(bytes(cramjam.snappy.decompress(p.data))))
        return ret
```

This example verifies that an encoded payload matches the `binary/snappy` filetype -- i.e., that it was encoded using the same custom `encode()` function -- and if so, performs decompression followed by unmarshaling.

**Set Data Converter to use custom Payload Codec**

Add a `data_converter` parameter to your `Client.connect()` options that overrides the default Converter with your Payload Codec:

```python
from codec import EncryptionCodec

client = await Client.connect(
	"localhost:7233",
	data_converter=dataclasses.replace(
		temporalio.converter.default(), payload_codec=EncryptionCodec()
	),
)
```

- Data **encoding** is performed by the client using the converters and codecs provided by Temporal or your custom implementation when passing input to the Temporal Cluster. For example, plain text input is usually serialized into a JSON object, and can then be compressed or encrypted.
- Data **decoding** may be performed by your application logic during your Workflows or Activities as necessary, but decoded Workflow results are never persisted back to the Temporal Cluster. Instead, they are stored encoded on the Cluster, and you need to provide an additional parameter when using the [temporal workflow show](/cli/workflow#show) command or when browsing the Web UI to view output.

For reference, see the [Encryption](https://github.com/temporalio/samples-python/tree/main/encryption) sample.

### Using a Codec Server

A Codec Server is an HTTP server that uses your custom Codec logic to decode your data remotely.
The Codec Server is independent of the Temporal Cluster and decodes your encrypted payloads through predefined endpoints.
You create, operate, and manage access to your Codec Server in your own environment.
The Temporal CLI and the Web UI in turn provide built-in hooks to call the Codec Server to decode encrypted payloads on demand.
Refer to the [Codec Server](/production-deployment/data-encryption) documentation for information on how to design and deploy a Codec Server.

## Payload conversion

Temporal SDKs provide a default [Payload Converter](/payload-converter) that can be customized to convert a custom data type to [Payload](/dataconversion#payload) and back.

### Conversion sequence {#conversion-sequence}

The order in which your encoding Payload Converters are applied depends on the order given to the Data Converter.
You can set multiple encoding Payload Converters to run your conversions.
When the Data Converter receives a value for conversion, it passes through each Payload Converter in sequence until the converter that handles the data type does the conversion.

Payload Converters can be customized independently of a Payload Codec.
Temporal's Converter architecture looks like this:

<CaptionedImage
    src="/img/info/converter-architecture.png"
    title="Temporal converter architecture"
/>

### Custom Payload Converter {#custom-payload-converter}

**How to use a custom Payload Converter using the Temporal Python SDK.**

Data Converters convert raw Temporal payloads to/from actual Python types.
A custom Data Converter of type `temporalio.converter.DataConverter` can be set through the `data_converter` parameter of the `Client` constructor.
Data Converters are a combination of Payload Converters, Payload Codecs, and Failure Converters.
Payload Converters convert Python values to/from serialized bytes.
Payload Codecs convert bytes to bytes (for example, for compression or encryption).
Failure Converters convert exceptions to/from serialized failures.

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](https://docs.python.org/3/library/dataclasses.html)
  - Iterables including ones JSON dump may not support by default, e.g. `set`
  - [IntEnum, StrEnum](https://docs.python.org/3/library/enum.html) based enumerates
  - [UUID](https://docs.python.org/3/library/uuid.html)

To use Pydantic model instances, see [Pydantic Support](#pydantic-support).

`datetime.date`, `datetime.time`, and `datetime.datetime` can only be used with the Pydantic Data Converter.

Although Workflows, Updates, Signals, and Queries can all be defined with multiple input parameters, users are strongly
encouraged to use a single `dataclass` or Pydantic model parameter so that fields with defaults can be easily added
without breaking compatibility.
Similar advice applies to return values.

Classes with generics may not have the generics properly resolved.
The current implementation does not have generic type resolution.
Users should use concrete types.

### Pydantic Support

To use Pydantic model instances, install Pydantic and set the Pydantic Data Converter when creating Client instances:

```python
from temporalio.contrib.pydantic import pydantic_data_converter

client = Client(data_converter=pydantic_data_converter, ...)
```

This Data Converter supports conversion of all [types supported by Pydantic](https://docs.pydantic.dev/latest/api/standard_library_types/) to and from JSON.
In addition to Pydantic models, supported types include:

- Everything that [`json.dumps()`](https://docs.python.org/3/library/json.html#py-to-json-table) supports by default.
- Several standard library types that `json.dumps()` does not support, including dataclasses, types from the datetime module, sets, UUID, etc.
- Custom types composed of any of these, with any degree of nesting.
  For example, a list of Pydantic models with `datetime` fields.

See the [Pydantic documentation](https://docs.pydantic.dev/latest/api/standard_library_types/) for full details.

:::note

Pydantic v1 isn't supported by this Data Converter.
If you aren't yet able to upgrade from Pydantic v1, see https://github.com/temporalio/samples-python/tree/main/pydantic_converter/v1 for limited v1 support.

:::

### Custom Type Data Conversion

When converting from JSON, Workflow and Activity type hints are taken into account to convert to the proper types.
All common Python typings including `Optional`, `Union`, all forms of iterables and mappings, and `NewType` are supported in addition the regular JSON values mentioned before.

In Python, Data Converters contain a reference to a Payload Converter class that is used to convert input and output payloads.
By default, the Payload Converter is a `CompositePayloadConverter` which contains multiple `EncodingPayloadConverter`s to try to serialize/deserialize payloads.
Upon serialization, each `EncodingPayloadConverter` is used in order until one succeeds.

To implement a custom encoding for a custom type, a new `EncodingPayloadConverter` can be created.
For example, to support `IPv4Address` types:

```python
class IPv4AddressEncodingPayloadConverter(EncodingPayloadConverter):
    @property
    def encoding(self) -> str:
        return "text/ipv4-address"

    def to_payload(self, value: Any) -> Optional[Payload]:
        if isinstance(value, ipaddress.IPv4Address):
            return Payload(
                metadata={"encoding": self.encoding.encode()},
                data=str(value).encode(),
            )
        else:
            return None

    def from_payload(self, payload: Payload, type_hint: Optional[Type] = None) -> Any:
        assert not type_hint or type_hint is ipaddress.IPv4Address
        return ipaddress.IPv4Address(payload.data.decode())

class IPv4AddressPayloadConverter(CompositePayloadConverter):
    def __init__(self) -> None:
        # Just add ours as first before the defaults
        super().__init__(
            IPv4AddressEncodingPayloadConverter(),
            *DefaultPayloadConverter.default_encoding_payload_converters,
        )

my_data_converter = dataclasses.replace(
    DataConverter.default,
    payload_converter_className=IPv4AddressPayloadConverter,
)
```

This is good for many custom types. However, you might need to override the behavior of the just the existing JSON encoding payload converter to support a new type.
It is already the last encoding data converter in the list, so it's the fall-through behavior for any otherwise unknown type.
Customizing the existing JSON converter has the benefit of making the type work in lists, unions, etc.
The conversion can be customized for serialization with a custom `json.JSONEncoder` and deserialization with a custom `JSONTypeConverter`.
For example, to support `IPv4Address` types in existing JSON conversion:

```python
class IPv4AddressJSONEncoder(AdvancedJSONEncoder):
    def default(self, o: Any) -> Any:
        if isinstance(o, ipaddress.IPv4Address):
            return str(o)
        return super().default(o)
class IPv4AddressJSONTypeConverter(JSONTypeConverter):
    def to_typed_value(
        self, hint: Type, value: Any
    ) -> Union[Optional[Any], _JSONTypeConverterUnhandled]:
        if issubclass(hint, ipaddress.IPv4Address):
            return ipaddress.IPv4Address(value)
        return JSONTypeConverter.Unhandled

class IPv4AddressPayloadConverter(CompositePayloadConverter):
    def __init__(self) -> None:
        # Replace default JSON plain with our own that has our encoder and type
        # converter
        json_converter = JSONPlainPayloadConverter(
            encoder=IPv4AddressJSONEncoder,
            custom_type_converters=[IPv4AddressJSONTypeConverter()],
        )
        super().__init__(
            *[
                c if not isinstance(c, JSONPlainPayloadConverter) else json_converter
                for c in DefaultPayloadConverter.default_encoding_payload_converters
            ]
        )

my_data_converter = dataclasses.replace(
    DataConverter.default,
    payload_converter_className=IPv4AddressPayloadConverter,
)
```

Now `IPv4Address` can be used in type hints including collections, optionals, etc.

---

## Core application - Python SDK

This page shows how to do the following:

- [Develop a basic Workflow](#develop-workflows)
- [Define Workflow parameters](#workflow-parameters)
- [Define Workflow return parameters](#workflow-return-values)
- [Customize your Workflow Type](#workflow-type)
- [Develop Workflow logic](#workflow-logic-requirements)
- [Develop a basic Activity](#develop-activities)
- [Develop Activity Parameters](#activity-parameters)
- [Define Activity return values](#activity-return-values)
- [Customize your Activity Type](#activity-type)
- [Start an Activity Execution](#activity-execution)
- [Set the required Activity Timeouts](#required-timeout)
- [Get the results of an Activity Execution](#get-activity-results)
- [Run a Worker Process](#run-a-dev-worker)
- [Register types](#register-types)

## Develop a basic Workflow {#develop-workflows}

**How to develop a basic Workflow using the Temporal Python SDK.**

Workflows are the fundamental unit of a Temporal Application, and it all starts with the development of a
[Workflow Definition](/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.

```python
from temporalio import workflow