"""Base converter and implementations for data conversion."""

from __future__ import annotations

import collections

import collections.abc

import dataclasses

import inspect

import json

import sys

import traceback

import uuid

import warnings

from abc import ABC, abstractmethod

from dataclasses import dataclass

from datetime import datetime

from enum import IntEnum

from typing import (

Any,

Awaitable,

Callable,

ClassVar,

Dict,

List,

Mapping,

NewType,

Optional,

Sequence,

Tuple,

Type,

TypeVar,

Union,

get_type_hints,

overload,

)

import google.protobuf.json_format

import google.protobuf.message

import google.protobuf.symbol_database

from typing_extensions import Literal

import temporalio.api.common.v1

import temporalio.api.enums.v1

import temporalio.api.failure.v1

import temporalio.common

import temporalio.exceptions

import temporalio.types

if sys.version_info < (3, 11):

# Python's datetime.fromisoformat doesn't support certain formats pre-3.11

from dateutil import parser # type: ignore

# StrEnum is available in 3.11+

if sys.version_info >= (3, 11):

from enum import StrEnum

if sys.version_info >= (3, 10):

from types import UnionType

class PayloadConverter(ABC):

"""Base payload converter to/from multiple payloads/values."""

default: ClassVar[PayloadConverter]

"""Default payload converter."""

@abstractmethod

def to_payloads(

self, values: Sequence[Any]

) -> List[temporalio.api.common.v1.Payload]:

"""Encode values into payloads.

Implementers are expected to just return the payload for

:py:class:`temporalio.common.RawValue`.

Args:

values: Values to be converted.

Returns:

Converted payloads. Note, this does not have to be the same number

as values given, but must be at least one and cannot be more than

was given.

Raises:

Exception: Any issue during conversion.

"""

raise NotImplementedError

@abstractmethod

def from_payloads(

self,

payloads: Sequence[temporalio.api.common.v1.Payload],

type_hints: Optional[List[Type]] = None,

) -> List[Any]:

"""Decode payloads into values.

Implementers are expected to treat a type hint of

:py:class:`temporalio.common.RawValue` as just the raw value.

Args:

payloads: Payloads to convert to Python values.

type_hints: Types that are expected if any. This may not have any

types if there are no annotations on the target. If this is

present, it must have the exact same length as payloads even if

the values are just "object".

Returns:

Collection of Python values. Note, this does not have to be the same

number as values given, but at least one must be present.

Raises:

Exception: Any issue during conversion.

"""

raise NotImplementedError

def to_payloads_wrapper(

self, values: Sequence[Any]

) -> temporalio.api.common.v1.Payloads:

""":py:meth:`to_payloads` for the

:py:class:`temporalio.api.common.v1.Payloads` wrapper.

"""

return temporalio.api.common.v1.Payloads(payloads=self.to_payloads(values))

def from_payloads_wrapper(

self, payloads: Optional[temporalio.api.common.v1.Payloads]

) -> List[Any]:

""":py:meth:`from_payloads` for the

:py:class:`temporalio.api.common.v1.Payloads` wrapper.

"""

if not payloads or not payloads.payloads:

return []

return self.from_payloads(payloads.payloads)

def to_payload(self, value: Any) -> temporalio.api.common.v1.Payload:

"""Convert a single value to a payload.

This is a shortcut for :py:meth:`to_payloads` with a single-item list

and result.

Args:

value: Value to convert to a single payload.

Returns:

Single converted payload.

"""

return self.to_payloads([value])[0]

@overload

def from_payload(self, payload: temporalio.api.common.v1.Payload) -> Any:

...

@overload

def from_payload(

self,

payload: temporalio.api.common.v1.Payload,

type_hint: Type[temporalio.types.AnyType],

) -> temporalio.types.AnyType:

...

def from_payload(

self,

payload: temporalio.api.common.v1.Payload,

type_hint: Optional[Type] = None,

) -> Any:

"""Convert a single payload to a value.

This is a shortcut for :py:meth:`from_payloads` with a single-item list

and result.

Args:

payload: Payload to convert to value.

type_hint: Optional type hint to say which type to convert to.

Returns:

Single converted value.

"""

return self.from_payloads([payload], [type_hint] if type_hint else None)[0]

class EncodingPayloadConverter(ABC):

"""Base converter to/from single payload/value with a known encoding for use in CompositePayloadConverter."""

@property

@abstractmethod

def encoding(self) -> str:

"""Encoding for the payload this converter works with."""

raise NotImplementedError

@abstractmethod

def to_payload(self, value: Any) -> Optional[temporalio.api.common.v1.Payload]:

"""Encode a single value to a payload or None.

Args:

value: Value to be converted.

Returns:

Payload of the value or None if unable to convert.

Raises:

TypeError: Value is not the expected type.

ValueError: Value is of the expected type but otherwise incorrect.

RuntimeError: General error during encoding.

"""

raise NotImplementedError

@abstractmethod

def from_payload(

self,

payload: temporalio.api.common.v1.Payload,

type_hint: Optional[Type] = None,

) -> Any:

"""Decode a single payload to a Python value or raise exception.

Args:

payload: Payload to convert to Python value.

type_hint: Type that is expected if any. This may not have a type if

there are no annotations on the target.

Return:

The decoded value from the payload. Since the encoding is checked by

the caller, this should raise an exception if the payload cannot be

converted.

Raises:

RuntimeError: General error during decoding.

"""

raise NotImplementedError

class CompositePayloadConverter(PayloadConverter):

"""Composite payload converter that delegates to a list of encoding payload converters.

Encoding/decoding are attempted on each payload converter successively until

it succeeds.

Attributes:

converters: List of payload converters to delegate to, in order.

"""

converters: Mapping[bytes, EncodingPayloadConverter]

def __init__(self, *converters: EncodingPayloadConverter) -> None:

"""Initializes the data converter.

Args:

converters: Payload converters to delegate to, in order.

"""

# Insertion order preserved here since Python 3.7

self.converters = {c.encoding.encode(): c for c in converters}

def to_payloads(

self, values: Sequence[Any]

) -> List[temporalio.api.common.v1.Payload]:

"""Encode values trying each converter.

See base class. Always returns the same number of payloads as values.

Raises:

RuntimeError: No known converter

"""

payloads = []

for index, value in enumerate(values):

# We intentionally attempt these serially just in case a stateful

# converter may rely on the previous values

payload = None

# RawValue should just pass through

if isinstance(value, temporalio.common.RawValue):

payload = value.payload

else:

for converter in self.converters.values():

payload = converter.to_payload(value)

if payload is not None:

break

if payload is None:

raise RuntimeError(

f"Value at index {index} of type {type(value)} has no known converter"

)

payloads.append(payload)

return payloads

def from_payloads(

self,

payloads: Sequence[temporalio.api.common.v1.Payload],

type_hints: Optional[List[Type]] = None,

) -> List[Any]:

"""Decode values trying each converter.

See base class. Always returns the same number of values as payloads.

Raises:

KeyError: Unknown payload encoding

RuntimeError: Error during decode

"""

values = []

for index, payload in enumerate(payloads):

type_hint = None

if type_hints and len(type_hints) > index:

type_hint = type_hints[index]

# Raw value should just wrap

if type_hint == temporalio.common.RawValue:

values.append(temporalio.common.RawValue(payload))

continue

encoding = payload.metadata.get("encoding", b"<unknown>")

converter = self.converters.get(encoding)

if converter is None:

raise KeyError(f"Unknown payload encoding {encoding.decode()}")

try:

values.append(converter.from_payload(payload, type_hint))

except RuntimeError as err:

raise RuntimeError(

f"Payload at index {index} with encoding {encoding.decode()} could not be converted"

) from err

return values

class DefaultPayloadConverter(CompositePayloadConverter):

"""Default payload converter compatible with other Temporal SDKs.

This handles None, bytes, all protobuf message types, and any type that

:py:func:`json.dump` accepts. A singleton instance of this is available at

:py:attr:`PayloadConverter.default`.

"""

default_encoding_payload_converters: Tuple[EncodingPayloadConverter, ...]

"""Default set of encoding payload converters the default payload converter

uses.

"""

def __init__(self) -> None:

"""Create a default payload converter."""

super().__init__(*DefaultPayloadConverter.default_encoding_payload_converters)

class BinaryNullPayloadConverter(EncodingPayloadConverter):

"""Converter for 'binary/null' payloads supporting None values."""

@property

def encoding(self) -> str:

"""See base class."""

return "binary/null"

def to_payload(self, value: Any) -> Optional[temporalio.api.common.v1.Payload]:

"""See base class."""

if value is None:

return temporalio.api.common.v1.Payload(

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

)

return None

def from_payload(

self,

payload: temporalio.api.common.v1.Payload,

type_hint: Optional[Type] = None,

) -> Any:

"""See base class."""

if len(payload.data) > 0:

raise RuntimeError("Expected empty data set for binary/null")

return None

class BinaryPlainPayloadConverter(EncodingPayloadConverter):

"""Converter for 'binary/plain' payloads supporting bytes values."""

@property

def encoding(self) -> str:

"""See base class."""

return "binary/plain"

def to_payload(self, value: Any) -> Optional[temporalio.api.common.v1.Payload]:

"""See base class."""

if isinstance(value, bytes):

return temporalio.api.common.v1.Payload(

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

)

return None

def from_payload(

self,

payload: temporalio.api.common.v1.Payload,

type_hint: Optional[Type] = None,

) -> Any:

"""See base class."""

return payload.data

_sym_db = google.protobuf.symbol_database.Default()

class JSONProtoPayloadConverter(EncodingPayloadConverter):

"""Converter for 'json/protobuf' payloads supporting protobuf Message values."""

def __init__(self, ignore_unknown_fields: bool = False):

"""Initialize a JSON proto converter.

Args:

ignore_unknown_fields: Determines whether converter should error if

unknown fields are detected

"""

super().__init__()

self._ignore_unknown_fields = ignore_unknown_fields

@property

def encoding(self) -> str:

"""See base class."""

return "json/protobuf"

def to_payload(self, value: Any) -> Optional[temporalio.api.common.v1.Payload]:

"""See base class."""

if (

isinstance(value, google.protobuf.message.Message)

and value.DESCRIPTOR is not None

):

# We have to convert to dict then to JSON because MessageToJson does

# not have a compact option removing spaces and newlines

json_str = json.dumps(

google.protobuf.json_format.MessageToDict(value),

separators=(",", ":"),

sort_keys=True,

)

return temporalio.api.common.v1.Payload(

metadata={

"encoding": self.encoding.encode(),

"messageType": value.DESCRIPTOR.full_name.encode(),

},

data=json_str.encode(),

)

return None

def from_payload(

self,

payload: temporalio.api.common.v1.Payload,

type_hint: Optional[Type] = None,

) -> Any:

"""See base class."""

message_type = payload.metadata.get("messageType", b"<unknown>").decode()

try:

value = _sym_db.GetSymbol(message_type)()

return google.protobuf.json_format.Parse(

payload.data,

value,

ignore_unknown_fields=self._ignore_unknown_fields,

)

except KeyError as err:

raise RuntimeError(f"Unknown Protobuf type {message_type}") from err

except google.protobuf.json_format.ParseError as err:

raise RuntimeError("Failed parsing") from err

class BinaryProtoPayloadConverter(EncodingPayloadConverter):

"""Converter for 'binary/protobuf' payloads supporting protobuf Message values."""

@property

def encoding(self) -> str:

"""See base class."""

return "binary/protobuf"

def to_payload(self, value: Any) -> Optional[temporalio.api.common.v1.Payload]:

"""See base class."""

if (

isinstance(value, google.protobuf.message.Message)

and value.DESCRIPTOR is not None

):

return temporalio.api.common.v1.Payload(

metadata={

"encoding": self.encoding.encode(),

"messageType": value.DESCRIPTOR.full_name.encode(),

},

data=value.SerializeToString(),

)

return None

def from_payload(

self,

payload: temporalio.api.common.v1.Payload,

type_hint: Optional[Type] = None,

) -> Any:

"""See base class."""

message_type = payload.metadata.get("messageType", b"<unknown>").decode()

try:

value = _sym_db.GetSymbol(message_type)()

value.ParseFromString(payload.data)

return value

except KeyError as err:

raise RuntimeError(f"Unknown Protobuf type {message_type}") from err

except google.protobuf.message.DecodeError as err:

raise RuntimeError("Failed parsing") from err

class AdvancedJSONEncoder(json.JSONEncoder):

"""Advanced JSON encoder.

This encoder supports dataclasses, classes with dict() functions, and

all iterables as lists.

"""

def default(self, o: Any) -> Any:

"""Override JSON encoding default.

See :py:meth:`json.JSONEncoder.default`.

"""

# Dataclass support

if dataclasses.is_dataclass(o):

return dataclasses.asdict(o)

# Support for models with "dict" function like Pydantic

dict_fn = getattr(o, "dict", None)

if callable(dict_fn):

return dict_fn()

# Support for non-list iterables like set

if not isinstance(o, list) and isinstance(o, collections.abc.Iterable):

return list(o)

# Support for UUID

if isinstance(o, uuid.UUID):

return str(o)

return super().default(o)

class JSONPlainPayloadConverter(EncodingPayloadConverter):

"""Converter for 'json/plain' payloads supporting common Python values.

For encoding, this supports all values that :py:func:`json.dump` supports

and by default adds extra encoding support for dataclasses, classes with

``dict()`` methods, and all iterables.

For decoding, this uses type hints to attempt to rebuild the type from the

type hint.

"""

_encoder: Optional[Type[json.JSONEncoder]]

_decoder: Optional[Type[json.JSONDecoder]]

_encoding: str

def __init__(

self,

*,

encoder: Optional[Type[json.JSONEncoder]] = AdvancedJSONEncoder,

decoder: Optional[Type[json.JSONDecoder]] = None,

encoding: str = "json/plain",

custom_type_converters: Sequence[JSONTypeConverter] = [],

) -> None:

"""Initialize a JSON data converter.

Args:

encoder: Custom encoder class object to use.

decoder: Custom decoder class object to use.

encoding: Encoding name to use.

custom_type_converters: Set of custom type converters that are used

when converting from a payload to type-hinted values.

"""

super().__init__()

self._encoder = encoder

self._decoder = decoder

self._encoding = encoding

self._custom_type_converters = custom_type_converters

@property

def encoding(self) -> str:

"""See base class."""

return self._encoding

def to_payload(self, value: Any) -> Optional[temporalio.api.common.v1.Payload]:

"""See base class."""

# Check for pydantic then send warning

if hasattr(value, "parse_obj"):

warnings.warn(

"If you're using pydantic model, refer to https://github.com/temporalio/samples-python/tree/main/pydantic_converter for better support"

)

# We let JSON conversion errors be thrown to caller

return temporalio.api.common.v1.Payload(

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

data=json.dumps(

value, cls=self._encoder, separators=(",", ":"), sort_keys=True

).encode(),

)

def from_payload(

self,

payload: temporalio.api.common.v1.Payload,

type_hint: Optional[Type] = None,

) -> Any:

"""See base class."""

try:

obj = json.loads(payload.data, cls=self._decoder)

if type_hint:

obj = value_to_type(type_hint, obj, self._custom_type_converters)

return obj

except json.JSONDecodeError as err:

raise RuntimeError("Failed parsing") from err

_JSONTypeConverterUnhandled = NewType("_JSONTypeConverterUnhandled", object)

class JSONTypeConverter(ABC):

"""Converter for converting an object from Python :py:func:`json.loads`

result (e.g. scalar, list, or dict) to a known type.

"""

Unhandled = _JSONTypeConverterUnhandled(object())

"""Sentinel value that must be used as the result of

:py:meth:`to_typed_value` to say the given type is not handled by this

converter."""

@abstractmethod

def to_typed_value(

self, hint: Type, value: Any

) -> Union[Optional[Any], _JSONTypeConverterUnhandled]:

"""Convert the given value to a type based on the given hint.

Args:

hint: Type hint to use to help in converting the value.

value: Value as returned by :py:func:`json.loads`. Usually a scalar,

list, or dict.

Returns:

The converted value or :py:attr:`Unhandled` if this converter does

not handle this situation.

"""

raise NotImplementedError

class PayloadCodec(ABC):

"""Codec for encoding/decoding to/from bytes.

Commonly used for compression or encryption.

"""

@abstractmethod

async def encode(

self, payloads: Sequence[temporalio.api.common.v1.Payload]

) -> List[temporalio.api.common.v1.Payload]:

"""Encode the given payloads.

Args:

payloads: Payloads to encode. This value should not be mutated.

Returns:

Encoded payloads. Note, this does not have to be the same number as

payloads given, but must be at least one and cannot be more than was

given.

"""

raise NotImplementedError

@abstractmethod

async def decode(

self, payloads: Sequence[temporalio.api.common.v1.Payload]

) -> List[temporalio.api.common.v1.Payload]:

"""Decode the given payloads.

Args:

payloads: Payloads to decode. This value should not be mutated.

Returns:

Decoded payloads. Note, this does not have to be the same number as

payloads given, but must be at least one and cannot be more than was

given.

"""

raise NotImplementedError

async def encode_wrapper(self, payloads: temporalio.api.common.v1.Payloads) -> None:

""":py:meth:`encode` for the

:py:class:`temporalio.api.common.v1.Payloads` wrapper.

This replaces the payloads within the wrapper.

"""

new_payloads = await self.encode(payloads.payloads)

del payloads.payloads[:]

# TODO(cretz): Copy too expensive?

payloads.payloads.extend(new_payloads)

async def decode_wrapper(self, payloads: temporalio.api.common.v1.Payloads) -> None:

""":py:meth:`decode` for the

:py:class:`temporalio.api.common.v1.Payloads` wrapper.

This replaces the payloads within.

"""

new_payloads = await self.decode(payloads.payloads)

del payloads.payloads[:]

# TODO(cretz): Copy too expensive?

payloads.payloads.extend(new_payloads)

async def encode_failure(self, failure: temporalio.api.failure.v1.Failure) -> None:

"""Encode payloads of a failure."""

await self._apply_to_failure_payloads(failure, self.encode_wrapper)

async def decode_failure(self, failure: temporalio.api.failure.v1.Failure) -> None:

"""Decode payloads of a failure."""

await self._apply_to_failure_payloads(failure, self.decode_wrapper)

async def _apply_to_failure_payloads(

self,

failure: temporalio.api.failure.v1.Failure,

cb: Callable[[temporalio.api.common.v1.Payloads], Awaitable[None]],

) -> None:

if failure.HasField("encoded_attributes"):

# Wrap in payloads and merge back

payloads = temporalio.api.common.v1.Payloads(

payloads=[failure.encoded_attributes]

)

await cb(payloads)

failure.encoded_attributes.CopyFrom(payloads.payloads[0])

if failure.HasField(

"application_failure_info"

) and failure.application_failure_info.HasField("details"):

await cb(failure.application_failure_info.details)

elif failure.HasField(

"timeout_failure_info"

) and failure.timeout_failure_info.HasField("last_heartbeat_details"):

await cb(failure.timeout_failure_info.last_heartbeat_details)

elif failure.HasField(

"canceled_failure_info"

) and failure.canceled_failure_info.HasField("details"):

await cb(failure.canceled_failure_info.details)

elif failure.HasField(

"reset_workflow_failure_info"

) and failure.reset_workflow_failure_info.HasField("last_heartbeat_details"):

await cb(failure.reset_workflow_failure_info.last_heartbeat_details)

if failure.HasField("cause"):

await self._apply_to_failure_payloads(failure.cause, cb)

class FailureConverter(ABC):

"""Base failure converter to/from errors.

Note, for workflow exceptions, :py:attr:`to_failure` is only invoked if the

exception is an instance of :py:class:`temporalio.exceptions.FailureError`.

Users should extend :py:class:`temporalio.exceptions.ApplicationError` if

they want a custom workflow exception to work with this class.

"""

default: ClassVar[FailureConverter]

"""Default failure converter."""

@abstractmethod

def to_failure(

self,

exception: BaseException,

payload_converter: PayloadConverter,

failure: temporalio.api.failure.v1.Failure,

) -> None:

"""Convert the given exception to a Temporal failure.

Users should make sure not to alter the ``exception`` input.

Args:

exception: The exception to convert.

payload_converter: The payload converter to use if needed.

failure: The failure to update with error information.

"""

raise NotImplementedError

@abstractmethod

def from_failure(

self,

failure: temporalio.api.failure.v1.Failure,

payload_converter: PayloadConverter,

) -> BaseException:

"""Convert the given Temporal failure to an exception.

Users should make sure not to alter the ``failure`` input.

Args:

failure: The failure to convert.

payload_converter: The payload converter to use if needed.

Returns:

Converted error.

"""

raise NotImplementedError

class DefaultFailureConverter(FailureConverter):

"""Default failure converter.

A singleton instance of this is available at

:py:attr:`FailureConverter.default`.

"""

def __init__(self, *, encode_common_attributes: bool = False) -> None:

"""Create the default failure converter.

Args:

encode_common_attributes: If ``True``, the message and stack trace

of the failure will be moved into the encoded attribute section

of the failure which can be encoded with a codec.

"""

super().__init__()

self._encode_common_attributes = encode_common_attributes

def to_failure(

self,

exception: BaseException,

payload_converter: PayloadConverter,

failure: temporalio.api.failure.v1.Failure,

) -> None:

"""See base class."""

# If already a failure error, use that

if isinstance(exception, temporalio.exceptions.FailureError):

self._error_to_failure(exception, payload_converter, failure)

else:

# Convert to failure error

failure_error = temporalio.exceptions.ApplicationError(

str(exception), type=exception.__class__.__name__

)

failure_error.__traceback__ = exception.__traceback__

failure_error.__cause__ = exception.__cause__

self._error_to_failure(failure_error, payload_converter, failure)

# Encode common attributes if requested

if self._encode_common_attributes:

# Move message and stack trace to encoded attribute payload

failure.encoded_attributes.CopyFrom(

payload_converter.to_payloads(

[{"message": failure.message, "stack_trace": failure.stack_trace}]

)[0]

)

failure.message = "Encoded failure"

failure.stack_trace = ""

def _error_to_failure(

self,

error: temporalio.exceptions.FailureError,

payload_converter: PayloadConverter,

failure: temporalio.api.failure.v1.Failure,

) -> None:

# If there is an underlying proto already, just use that

if error.failure:

failure.CopyFrom(error.failure)

return

# Set message, stack, and cause. Obtaining cause follows rules from

# https://docs.python.org/3/library/exceptions.html#exception-context

failure.message = error.message

if error.__traceback__:

failure.stack_trace = "\n".join(traceback.format_tb(error.__traceback__))

if error.__cause__:

self.to_failure(error.__cause__, payload_converter, failure.cause)

elif not error.__suppress_context__ and error.__context__:

self.to_failure(error.__context__, payload_converter, failure.cause)

# Set specific subclass values

if isinstance(error, temporalio.exceptions.ApplicationError):

failure.application_failure_info.SetInParent()

if error.type:

failure.application_failure_info.type = error.type

failure.application_failure_info.non_retryable = error.non_retryable

if error.details:

failure.application_failure_info.details.CopyFrom(

payload_converter.to_payloads_wrapper(error.details)

)

elif isinstance(error, temporalio.exceptions.TimeoutError):

failure.timeout_failure_info.SetInParent()

failure.timeout_failure_info.timeout_type = (

temporalio.api.enums.v1.TimeoutType.ValueType(error.type or 0)

)

if error.last_heartbeat_details:

failure.timeout_failure_info.last_heartbeat_details.CopyFrom(

payload_converter.to_payloads_wrapper(error.last_heartbeat_details)

)

elif isinstance(error, temporalio.exceptions.CancelledError):

failure.canceled_failure_info.SetInParent()

if error.details:

failure.canceled_failure_info.details.CopyFrom(

payload_converter.to_payloads_wrapper(error.details)

)

elif isinstance(error, temporalio.exceptions.TerminatedError):

failure.terminated_failure_info.SetInParent()

elif isinstance(error, temporalio.exceptions.ServerError):

failure.server_failure_info.SetInParent()

failure.server_failure_info.non_retryable = error.non_retryable

elif isinstance(error, temporalio.exceptions.ActivityError):

failure.activity_failure_info.SetInParent()

failure.activity_failure_info.scheduled_event_id = error.scheduled_event_id

failure.activity_failure_info.started_event_id = error.started_event_id

failure.activity_failure_info.identity = error.identity

failure.activity_failure_info.activity_type.name = error.activity_type

failure.activity_failure_info.activity_id = error.activity_id

failure.activity_failure_info.retry_state = (

temporalio.api.enums.v1.RetryState.ValueType(error.retry_state or 0)

)

elif isinstance(error, temporalio.exceptions.ChildWorkflowError):

failure.child_workflow_execution_failure_info.SetInParent()

failure.child_workflow_execution_failure_info.namespace = error.namespace

failure.child_workflow_execution_failure_info.workflow_execution.workflow_id = (

error.workflow_id

)

failure.child_workflow_execution_failure_info.workflow_execution.run_id = (

error.run_id

)

failure.child_workflow_execution_failure_info.workflow_type.name = (

error.workflow_type

)

failure.child_workflow_execution_failure_info.initiated_event_id = (

error.initiated_event_id

)

failure.child_workflow_execution_failure_info.started_event_id = (

error.started_event_id

)

failure.child_workflow_execution_failure_info.retry_state = (

temporalio.api.enums.v1.RetryState.ValueType(error.retry_state or 0)

)

def from_failure(

self,

failure: temporalio.api.failure.v1.Failure,

payload_converter: PayloadConverter,

) -> BaseException:

"""See base class."""

# If encoded attributes are present and have the fields we expect,

# extract them

if failure.HasField("encoded_attributes"):

# Clone the failure to not mutate the incoming failure

new_failure = temporalio.api.failure.v1.Failure()

new_failure.CopyFrom(failure)

failure = new_failure

try:

encoded_attributes: Dict[str, Any] = payload_converter.from_payloads(

[failure.encoded_attributes]

)[0]

if isinstance(encoded_attributes, dict):

message = encoded_attributes.get("message")

if isinstance(message, str):

failure.message = message

stack_trace = encoded_attributes.get("stack_trace")

if isinstance(stack_trace, str):

failure.stack_trace = stack_trace

except:

pass

err: temporalio.exceptions.FailureError

if failure.HasField("application_failure_info"):

app_info = failure.application_failure_info

err = temporalio.exceptions.ApplicationError(

failure.message or "Application error",

*payload_converter.from_payloads_wrapper(app_info.details),

type=app_info.type or None,

non_retryable=app_info.non_retryable,

)

elif failure.HasField("timeout_failure_info"):

timeout_info = failure.timeout_failure_info

err = temporalio.exceptions.TimeoutError(

failure.message or "Timeout",

type=temporalio.exceptions.TimeoutType(int(timeout_info.timeout_type))

if timeout_info.timeout_type

else None,

last_heartbeat_details=payload_converter.from_payloads_wrapper(

timeout_info.last_heartbeat_details

),

)

elif failure.HasField("canceled_failure_info"):

cancel_info = failure.canceled_failure_info

err = temporalio.exceptions.CancelledError(

failure.message or "Cancelled",

*payload_converter.from_payloads_wrapper(cancel_info.details),

)

elif failure.HasField("terminated_failure_info"):

err = temporalio.exceptions.TerminatedError(failure.message or "Terminated")

elif failure.HasField("server_failure_info"):

server_info = failure.server_failure_info

err = temporalio.exceptions.ServerError(

failure.message or "Server error",

non_retryable=server_info.non_retryable,

)

elif failure.HasField("activity_failure_info"):

act_info = failure.activity_failure_info

err = temporalio.exceptions.ActivityError(

failure.message or "Activity error",

scheduled_event_id=act_info.scheduled_event_id,

started_event_id=act_info.started_event_id,

identity=act_info.identity,

activity_type=act_info.activity_type.name,

activity_id=act_info.activity_id,

retry_state=temporalio.exceptions.RetryState(int(act_info.retry_state))

if act_info.retry_state

else None,

)

elif failure.HasField("child_workflow_execution_failure_info"):

child_info = failure.child_workflow_execution_failure_info

err = temporalio.exceptions.ChildWorkflowError(

failure.message or "Child workflow error",

namespace=child_info.namespace,

workflow_id=child_info.workflow_execution.workflow_id,

run_id=child_info.workflow_execution.run_id,

workflow_type=child_info.workflow_type.name,

initiated_event_id=child_info.initiated_event_id,

started_event_id=child_info.started_event_id,

retry_state=temporalio.exceptions.RetryState(

int(child_info.retry_state)

)

if child_info.retry_state

else None,

)

else:

err = temporalio.exceptions.FailureError(failure.message or "Failure error")

err._failure = failure

if failure.HasField("cause"):

err.__cause__ = self.from_failure(failure.cause, payload_converter)

return err

class DefaultFailureConverterWithEncodedAttributes(DefaultFailureConverter):

"""Implementation of :py:class:`DefaultFailureConverter` which moves message

and stack trace to encoded attributes subject to a codec.