Hyrex provides automatic retry mechanisms for tasks that fail due to transient errors. Configure retries using max_retries and retry_backoff parameters.
Basic Retries
Set max_retries to automatically retry failed tasks:
from hyrex import HyrexRegistry
hy = HyrexRegistry()
@hy.task(max_retries=3)
def flaky_api_call():
response = requests.get("https://api.example.com")
if response.status_code >= 500:
# Raising an exception triggers a retry
raise Exception("Server error")
return response.json()
Any unhandled exception will trigger a retry. To prevent retries for specific errors, catch the exception and return an error response instead.
Backoff Strategies
Fixed Backoff
@hy.task(
max_retries=3,
retry_backoff=10 # 10 seconds between retries
)
def fixed_backoff_task():
pass
Exponential Backoff
@hy.task(
max_retries=5,
retry_backoff=lambda attempt: 2 ** attempt
)
def exponential_task():
# Delays: 2s, 4s, 8s, 16s, 32s
pass
Custom Backoff
def custom_backoff(attempt: int) -> int:
base = 5 * attempt
jitter = random.randint(0, 5)
return min(base + jitter, 300) # Max 5 minutes
@hy.task(
max_retries=5,
retry_backoff=custom_backoff
)
def custom_task():
pass
Error Handlers
Use on_error callbacks to monitor failures without affecting retry behavior:
from hyrex import get_hyrex_context
def handle_error(error: Exception):
context = get_hyrex_context()
if context:
# Log error with context
print(f"Task {context.task_name} failed on attempt {context.attempt_number}: {error}")
# Alert on final retry
if context.attempt_number == context.max_retries:
send_alert(f"Task {context.task_name} failed after {context.max_retries} attempts")
@hy.task(
max_retries=3,
on_error=handle_error
)
def monitored_task(data: dict):
# Task logic that might fail
process_data(data)
The on_error handler must accept either no arguments or exactly one Exception argument. The handler runs after each failure but doesn’t prevent retries.
Retry Context
Access retry information within tasks:
@hy.task(max_retries=3)
def adaptive_task():
context = get_hyrex_context()
if context:
# Different behavior based on attempt
if context.attempt_number == 1:
timeout = 5
elif context.attempt_number == 2:
timeout = 10
else:
timeout = 30
print(f"Attempt {context.attempt_number} of {context.max_retries + 1}")
else:
timeout = 5
return fetch_with_timeout(timeout)
Selective Retry Behavior
Control which errors trigger retries:
class PermanentError(Exception):
"""Errors that should not be retried"""
pass
@hy.task(max_retries=3)
def selective_retry_task(url: str):
try:
response = requests.get(url)
if response.status_code == 404:
# Don't retry - return error result
return {"error": "Not found", "status": 404}
if response.status_code == 400:
# Don't retry - catch and return
raise PermanentError("Bad request")
if response.status_code >= 500:
# Do retry - let exception propagate
raise Exception(f"Server error: {response.status_code}")
return response.json()
except PermanentError as e:
# Catch permanent errors to prevent retry
return {"error": str(e), "permanent": True}
# Let other exceptions propagate for retry
Next Steps
