# Later, with different circumstances...
await workflow.execute_activity(
    process_order,
    order,
    start_to_close_timeout=timedelta(seconds=10),
    retry_policy=slow_retry,
)
```

### Don't use Workflow Retry Policies

Unlike Activities, Workflows don't retry by default, and you usually shouldn't add a Retry Policy.
Workflows are deterministic and not designed for failure-prone operations.
A Workflow failure typically indicates a code bug or bad input data—retrying the entire Workflow repeats the same logic without fixing the underlying issue.

If you need retry logic for specific Workflow operations, implement it in your Workflow code rather than using a Workflow Retry Policy.

## Mark specific errors as non-retryable {#mark-errors-as-non-retryable}

**How to mark specific errors as non-retryable using the Temporal Python SDK**

Some failures are permanent and won't resolve through retries.
Mark these as non-retryable to fail fast instead of waiting for timeouts.

Set the `non_retryable` flag when raising an `ApplicationError`:

```python
from temporalio import activity
from temporalio.exceptions import ApplicationError

@activity.defn
async def process_payment(card_number: str, amount: float):
    if not is_valid_card_format(card_number):
        # Invalid format will never become valid through retries
        raise ApplicationError(
            f"Invalid credit card format: {card_number}",
            type="InvalidCardFormat",
            non_retryable=True,
        )

    if amount <= 0:
        # Invalid amount won't be fixed by retrying
        raise ApplicationError(
            f"Amount must be positive: {amount}",
            type="InvalidAmount",
            non_retryable=True,
        )

    # Process payment...
```

An `ApplicationError` with `non_retryable=True` will never retry, regardless of the Retry Policy.

Use non-retryable errors for:
- Invalid input data that prevents the Activity from proceeding
- Business rule violations
- Authorization failures

**Use this sparingly.**
In most cases, it's better to let the Retry Policy handle when to stop retrying based on time or attempts.

## Specify non-retryable error types {#specify-non-retryable-error-types}

**How to specify non-retryable error types in Retry Policies using the Temporal Python SDK**

Sometimes you want the Workflow (caller) to decide which error types shouldn't retry, rather than the Activity (implementer).

List error types that shouldn't retry in your Retry Policy:

```python
from temporalio import workflow
from temporalio.common import RetryPolicy
from datetime import timedelta

@workflow.defn
class CheckoutWorkflow:
    @workflow.run
    async def run(self, payment_details):
        retry_policy = RetryPolicy(
            non_retryable_error_types=[
                "InvalidCardFormat",
                "InsufficientFunds",
                "AccountClosed",
            ]
        )

        try:
            result = await workflow.execute_activity(
                process_payment,
                payment_details,
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=retry_policy,
            )
            return result
        except ActivityError as e:
            workflow.logger.error(f"Payment failed: {e.cause}")
            # Handle the non-retryable error...
```

When an Activity raises an `ApplicationError`, Temporal checks if its `type` is in `non_retryable_error_types`.
If it matches, the Activity fails immediately without retries.

### When to use each approach

**`non_retryable=True` in the Activity**: Use when the Activity implementer knows the error is permanently unrecoverable.
This enforces the constraint for all callers.

**`non_retryable_error_types` in the Retry Policy**: Use when the caller wants to decide which errors are unrecoverable based on their business logic.
This lets different Workflows make different decisions about the same Activity.

## Implement rollback logic with the Saga pattern {#implement-saga-pattern}

**How to implement the Saga pattern using the Temporal Python SDK**

The Saga pattern coordinates a sequence of operations where each operation has a compensating action to undo its effects.
If any operation fails, execute compensating actions in reverse order to roll back previous operations.

Use this for multi-step processes like:
- E-commerce checkout (payment, inventory, shipping)
- Distributed transactions across services
- Multi-stage data updates

```python
from temporalio import workflow
from temporalio.exceptions import ActivityError
from datetime import timedelta

@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order):
        compensations = []

        try:
            # Reserve inventory
            compensations.append({
                "activity": revert_inventory,
                "input": order
            })
            await workflow.execute_activity(
                reserve_inventory,
                order,
                start_to_close_timeout=timedelta(seconds=10),
            )

            # Charge payment
            compensations.append({
                "activity": refund_payment,
                "input": order
            })
            payment_id = await workflow.execute_activity(
                charge_payment,
                order,
                start_to_close_timeout=timedelta(seconds=10),
            )

            # Create shipment
            compensations.append({
                "activity": cancel_shipment,
                "input": payment_id
            })
            shipment_id = await workflow.execute_activity(
                create_shipment,
                order,
                start_to_close_timeout=timedelta(seconds=10),
            )

            return {"payment_id": payment_id, "shipment_id": shipment_id}

        except ActivityError as e:
            workflow.logger.error(f"Order failed: {e.cause}, rolling back...")

            # Execute compensations in reverse order
            for compensation in reversed(compensations):
                try:
                    await workflow.execute_activity(
                        compensation["activity"],
                        compensation["input"],
                        start_to_close_timeout=timedelta(seconds=10),
                    )
                except ActivityError as comp_err:
                    # Log compensation failure but continue with others
                    workflow.logger.error(f"Compensation failed: {comp_err.cause}")

            # Re-raise the original error
            raise ApplicationError(
                f"Order failed: {e.cause}",
                type="OrderFailed"
            )
```

Key points:
- Add compensating actions to a list **before** executing each Activity
- Use `reversed(compensations)` to undo operations in the correct order
- Handle compensation failures gracefully (they might fail too)
- Temporal manages all state and retry logic, making Saga implementation straightforward

## Understand Temporal's failure types {#understand-failure-types}

Temporal uses specialized exception types to represent different failure scenarios.
All exceptions inherit from [`TemporalError`](https://python.temporal.io/temporalio.exceptions.TemporalError.html).

**Do not extend `TemporalError` or its children.**
Use the provided exception types to ensure:
- Consistent behavior across process and language boundaries
- Compatibility with the Temporal Service
- Proper serialization via Protocol Buffers

### Common failure types

**`ApplicationError`**: Raised by your code to indicate application-specific failures.
This is the only Temporal exception you should raise manually.
When you raise an `ApplicationError`, you can optionally provide a `type` string and mark it as `non_retryable`.

**`ActivityError`**: Wraps exceptions raised from Activities.
The `cause` field contains the original error (`ApplicationError`, `TimeoutError`, `CancelledError`, etc.).
Catch this in Workflows to handle Activity failures.

**`TimeoutError`**: Occurs when an Activity or Workflow exceeds its configured timeout.

**`CancelledError`**: Results from cancellation of a Workflow, Activity, or Timer.
You can catch and ignore this to continue execution despite cancellation.

**`TerminatedError`**: Occurs when a Workflow Execution is forcefully terminated.

**`ChildWorkflowError`**: Raised when a Child Workflow Execution fails.

**`WorkflowAlreadyStartedError`**: Raised when attempting to start a Workflow with an ID that's already running.

**`ServerError`**: Used for exceptions from the Temporal Service itself (like database failures).

### Workflow Task vs Workflow Execution failures

**Workflow Task failures** occur when Workflow code raises a non-Temporal exception (like `ValueError`, `TypeError`, or non-determinism errors).
These retry automatically, letting you fix bugs and redeploy without losing Workflow state.

**Workflow Execution failures** occur when Workflow code raises a Temporal exception like `ApplicationError`.
These put the Workflow in "Failed" state with no automatic retries.

Example of a permanent failure that should fail the Workflow:

```python
if distance.kilometers > MAX_DELIVERY_DISTANCE:
    # Retrying won't change the distance - this is permanent
    raise ApplicationError(
        "Customer lives outside service area",
        type="OutsideServiceArea"
    )
```

### Protecting sensitive information

The default Failure Converter copies exception messages and stack traces as plain text visible in the Web UI.
If your exceptions might contain sensitive information, configure a custom Failure Converter to encrypt this data.
See the [Securing Application Data course](https://learn.temporal.io/courses/appdatasec/) for details.

---

## Failure detection - Python SDK

This page shows how to do the following:

- [Raise and Handle Exceptions](#exception-handling)
- [Deliberately Fail Workflows](#workflow-failure)
- [Set Workflow Timeouts](#workflow-timeouts)
- [Set Workflow Retries](#workflow-retries)
- [Set Activity Timeouts](#activity-timeouts)
- [Set an Activity Retry Policy](#activity-retries)
- [Heartbeat an Activity](#activity-heartbeats)

## Raise and Handle Exceptions {#exception-handling}

In each Temporal SDK, error handling is implemented idiomatically, following the conventions of the language.
Temporal uses several different error classes internally — for example, [`CancelledError`](https://python.temporal.io/temporalio.exceptions.CancelledError.html) in the Python SDK, to handle a Workflow cancellation. 
You should not raise or otherwise implement these manually, as they are tied to Temporal platform logic.

The one Temporal error class that you will typically raise deliberately is [`ApplicationError`](https://python.temporal.io/temporalio.exceptions.ApplicationError.html).
In fact, *any* other exceptions that are raised from your Python code in a Temporal Activity will be converted to an `ApplicationError` internally.
This way, an error's type, severity, and any additional details can be sent to the Temporal Service, indexed by the Web UI, and even serialized across language boundaries.

In other words, these two code samples do the same thing:

```python 
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})"

@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
```

```python 
from temporalio.exceptions import ApplicationError

@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
```

Depending on your implementation, you may decide to use either method.
One reason to use the Temporal `ApplicationError` class is because it allows you to set an additional `non_retryable` parameter.
This way, you can decide whether an error should not be retried automatically by Temporal.
This can be useful for deliberately failing a Workflow due to bad input data, rather than waiting for a timeout to elapse:

```python
from temporalio.exceptions import ApplicationError

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

You can alternately specify a list of errors that are non-retryable in your Activity [Retry Policy](#activity-retries).

## Failing Workflows {#workflow-failure}

One of the core design principles of Temporal is that an Activity Failure will never directly cause a Workflow Failure — a Workflow should never return as Failed unless deliberately.
The default retry policy associated with Temporal Activities is to retry them until reaching a certain timeout threshold.
Activities will not actually *return* a failure to your Workflow until this condition or another non-retryable condition is met.
At this point, you can decide how to handle an error returned by your Activity the way you would in any other program.
For example, you could implement a [Saga Pattern](https://learn.temporal.io/tutorials/python/trip-booking-app/) that uses `try` and `except` blocks to "unwind" some of the steps your Workflow has performed up to the point of Activity Failure.

**You will only fail a Workflow by manually raising an `ApplicationError` from the Workflow code.**
You could do this in response to an Activity Failure, if the failure of that Activity means that your Workflow should not continue:

```python
try:
  credit_card_confirmation = await workflow.execute_activity_method()
except ActivityError as e:
  workflow.logger.error(f"Unable to process credit card {e.message}")
    raise ApplicationError(
      "Unable to process credit card", "CreditCardProcessingError"
    )
```

This works differently in a Workflow than raising exceptions from Activities.
In an Activity, any Python exceptions or custom exceptions are converted to a Temporal `ApplicationError`.
In a Workflow, any exceptions that are raised other than an explicit Temporal `ApplicationError` will only fail that particular [Workflow Task](https://docs.temporal.io/tasks#workflow-task-execution) and be retried.
This includes any typical Python runtime errors like a `NameError` or a `TypeError` that are raised automatically.
These errors are treated as bugs that can be corrected with a fixed deployment, rather than a reason for a Temporal Workflow Execution to return unexpectedly.

## Workflow timeouts {#workflow-timeouts}

**How to set Workflow timeouts using the Temporal Python SDK**

Each Workflow timeout controls the maximum duration of a different aspect of a Workflow Execution.

Before we continue, we want to note that we generally do not recommend setting Workflow Timeouts, because Workflows are designed to be long-running and resilient.
Instead, setting a Timeout can limit its ability to handle unexpected delays or long-running processes.
If you need to perform an action inside your Workflow after a specific period of time, we recommend using a Timer.

Workflow timeouts are set when [starting the Workflow Execution](#workflow-timeouts).

- **[Workflow Execution Timeout](/encyclopedia/detecting-workflow-failures#workflow-execution-timeout)** - restricts the maximum amount of time that a single Workflow Execution can be executed.
- **[Workflow Run Timeout](/encyclopedia/detecting-workflow-failures#workflow-run-timeout):** restricts the maximum amount of time that a single Workflow Run can last.
- **[Workflow Task Timeout](/encyclopedia/detecting-workflow-failures#workflow-task-timeout):** restricts the maximum amount of time that a Worker can execute a Workflow Task.

Set the timeout to either the [`start_workflow()`](https://python.temporal.io/temporalio.client.Client.html#start_workflow) or [`execute_workflow()`](https://python.temporal.io/temporalio.client.Client.html#execute_workflow) asynchronous methods.

Available timeouts are:

- `execution_timeout`
- `run_timeout`
- `task_timeout`

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

```python