class RetryPolicy
The RetryPolicy class defines how retries should be handled for failed atoms within a workflow. It provides customizable options for retrying operations, such as setting the number of attempts, delay between retries, and handling specific exceptions.

​
Flexible Application:

  • Can be applied at the workflow level (default for all atoms).
  • Can be overridden by individual atoms with a custom retry policy.

​
Parameters

​
Optional Parameters (Defaults Provided):

  • max_attempts (int): Maximum number of retry attempts.
    Default: 3
  • delay (float): Initial delay in seconds between retry attempts.
    Default: 1.0
  • backoff_factor (float): Multiplier to increase delay between each failed attempt.
    Default: 2.0
  • retry_exceptions (tuple of exceptions): Specific exceptions on which retries are attempted. By default, retries are attempted on all exceptions (Exception).
    Default: (Exception,)

​
Functions

def __init__(
    self,
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff_factor: float = 2.0,
    retry_exceptions: tuple = (Exception,),
):
If no custom retry policy is provided, the defaults above are used. 
def should_retry(self, attempt: int, error: Exception) -> bool:
Determines if another retry attempt should be made. 
def get_delay(self, attempt: int) -> float:
Calculate the delay before the next retry attempt.

​
Usage Examples

​
1. Applying a RetryPolicy to a Workflow

Define a retry policy for the entire workflow. This policy will be applied to all atoms unless overridden at the atom level. 
from preswald import Workflow, RetryPolicy

# Define a custom retry policy
custom_policy = RetryPolicy(
    max_attempts=3,
    delay=1.0,
    backoff_factor=2.0,
    retry_exceptions=(IOError, TimeoutError)
)

# Apply the policy to the workflow
workflow = Workflow(default_retry_policy=custom_policy)

​
2. Overriding a RetryPolicy for a Specific Atom

Atoms can have their own retry policy, which overrides the workflow’s default policy. 
@workflow.atom(retry_policy=RetryPolicy(max_attempts=5, delay=0.5))
def fetch_data():
    # Simulate a network operation
    raise IOError("Simulated failure")

​
3. Combining Workflow and Atom Policies

If both workflow-level and atom-level policies are specified, the atom’s policy takes precedence.

​
Key Features

  1. Customizable Retry Logic:
    • Adjust the number of attempts, delay, and backoff.
    • Target specific exceptions for retries.
  2. Exponential Backoff:
    • Avoid excessive retry attempts by increasing delay incrementally with the backoff_factor.
  3. Granular Control:
    • Use workflow-level defaults for simplicity.
    • Override defaults at the atom level for specific tasks.

​
Example Workflow with RetryPolicy

from preswald import Workflow, RetryPolicy

# Define a workflow-wide retry policy
workflow = Workflow(
    default_retry_policy=RetryPolicy(
        max_attempts=3,
        delay=1.0,
        backoff_factor=2.0,
        retry_exceptions=(IOError, TimeoutError)
    )
)

@workflow.atom()
def load_data():
    raise IOError("Simulated failure")

# Atom-specific retry policy
@workflow.atom(retry_policy=RetryPolicy(max_attempts=5, delay=0.5))
def fetch_data():
    raise TimeoutError("Another simulated failure")

# Execute the workflow
try:
    workflow.execute()
except Exception as e:
    print(f"Workflow execution failed: {e}")

​
Why Use RetryPolicy?

  • Reliability: Automatically handle transient failures with retries.
  • Efficiency: Prevent infinite retry loops with customizable limits and backoff strategies.
  • Control: Fine-tune retry behavior for specific workflows or tasks.
Enhance your workflow resilience with RetryPolicy! 🔄