Failure detection - Python SDK | Temporal Platform Documentation

On this page

This page shows how to do the following:

• Raise and Handle Exceptions
• [Deliberately Fail Workflows](#workflow...

Sign up or sign in to view the full extracted text.

Looking through this documentation, I found several clear issues that need to be fixed:

Location: In the "Raise and Handle Exceptions" section, first code example

Context:

class MyCustomError(Exception):
    def __init__(self, message, error_code):
        super().__init__(message)
        self.error_code = error_code

    def __str__(self):
        return f"{>>>self.message<<<} (Error Code: {self.error_code})"

Problem: The code references self.message but this attribute is never defined. The message parameter is passed to the parent class but not stored as an instance variable.

Fix: Either store the message as an instance variable:

def __init__(self, message, error_code):
    super().__init__(message)
    self.message = message
    self.error_code = error_code

Or use the inherited message handling:

def __str__(self):
    return f"{super().__str__()} (Error Code: {self.error_code})"

Location: In the "Raise and Handle Exceptions" section, first code example

Context:

@activity.defn
async def my_activity(input: MyActivityInput):
    try:
        # Your activity logic goes here
    except Exception as e:
        raise MyCustomError(
            f"Error encountered on attempt {>>>attempt<<<}",
        ) from e

Problem: The variable attempt is used but never defined. Additionally, MyCustomError expects two parameters (message and error_code) but only one is provided.

Fix: Either define the attempt variable or remove it from the message, and provide both required parameters:

raise MyCustomError(
    "Error encountered",
    "ACTIVITY_ERROR"
) from e

Location: In the "Raise and Handle Exceptions" section, second and third code examples

Context:

@activity.defn
async def my_activity(input: MyActivityInput):
    try:
        # Your activity logic goes here
    except Exception as e:
        raise ApplicationError(
            type="MyCustomError",
            message=f"Error encountered on attempt {>>>attempt<<<}",
        ) from e

Problem: The variable attempt is used but never defined in both examples.

Fix: Either define the attempt variable or remove it from the message:

message="Error encountered",

Location: In the "Failing Workflows" section, code example

Context:

try:
  credit_card_confirmation = await >>>workflow.execute_activity_method()<<<
except ActivityError as e:
  workflow.logger.error(f"Unable to process credit card {e.message}")

Problem: The method workflow.execute_activity_method() is not a valid Temporal Python SDK method. It should be workflow.execute_activity().

Fix: Change to:

credit_card_confirmation = await workflow.execute_activity(
    your_activity_function,
    args,
    start_to_close_timeout=timedelta(seconds=10)
)

Location: In the "Set Activity timeouts" section, link reference

Context: "[View the source code](https://github.com/temporalio/documentation/blob/main/sample-apps/python/>>>activity_timeouts_retires<<

No product messaging analysis available for this page.

No competitive analysis available for this page.