tessl/pypi-stamina

Production-grade retries made easy.

Overview

Eval results

Files

Retry Decorators and Context Managers

Name: tessl/pypi-stamina
Author: tessl

Core retry functionality providing both decorator and context manager approaches for handling transient failures. Supports comprehensive configuration of backoff strategies, timeouts, and exception handling for both synchronous and asynchronous code.

Capabilities

Retry Decorator

The main @retry decorator provides the simplest interface for adding retry behavior to functions. It automatically detects sync vs async functions and applies appropriate retry logic.

def retry(
    *,
    on: ExcOrPredicate,
    attempts: int | None = 10,
    timeout: float | datetime.timedelta | None = 45.0,
    wait_initial: float | datetime.timedelta = 0.1,
    wait_max: float | datetime.timedelta = 5.0,
    wait_jitter: float | datetime.timedelta = 1.0,
    wait_exp_base: float = 2.0,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """
    Decorator that retries decorated function if specified exceptions are raised.
    
    The backoff delays grow exponentially with jitter:
    min(wait_max, wait_initial * wait_exp_base^(attempt - 1) + random(0, wait_jitter))
    
    Parameters:
    - on: Exception(s) or predicate function to retry on (required)
    - attempts: Maximum total attempts (None for unlimited)
    - timeout: Maximum total time for all retries  
    - wait_initial: Minimum backoff before first retry
    - wait_max: Maximum backoff time at any point
    - wait_jitter: Maximum random jitter added to backoff
    - wait_exp_base: Exponential base for backoff calculation
    
    All time parameters accept float (seconds) or datetime.timedelta objects.
    """

Usage Examples:

import stamina
import httpx
import datetime as dt

# Basic usage - retry on specific exception
@stamina.retry(on=httpx.HTTPError, attempts=3)
def fetch_data(url):
    response = httpx.get(url)
    response.raise_for_status()
    return response.json()

# Multiple exception types
@stamina.retry(on=(httpx.HTTPError, ConnectionError), attempts=5)
def robust_fetch(url):
    return httpx.get(url).json()

# Custom predicate function for fine-grained control
def should_retry(exc):
    if isinstance(exc, httpx.HTTPStatusError):
        return 500 <= exc.response.status_code < 600
    return isinstance(exc, (httpx.ConnectError, httpx.TimeoutException))

@stamina.retry(on=should_retry, attempts=3, timeout=30.0)
def smart_fetch(url):
    response = httpx.get(url)
    response.raise_for_status() 
    return response.json()

# Timedelta parameters
@stamina.retry(
    on=ValueError,
    timeout=dt.timedelta(minutes=2),
    wait_initial=dt.timedelta(milliseconds=100),
    wait_max=dt.timedelta(seconds=10)
)
def process_with_timedeltas():
    return risky_operation()

# Async function support
@stamina.retry(on=httpx.HTTPError, attempts=3)
async def fetch_async(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

Retry Context Manager

The retry_context function provides manual control over retry loops, yielding Attempt context managers for each retry attempt.

def retry_context(
    on: ExcOrPredicate,
    attempts: int | None = 10,
    timeout: float | datetime.timedelta | None = 45.0,
    wait_initial: float | datetime.timedelta = 0.1,
    wait_max: float | datetime.timedelta = 5.0,
    wait_jitter: float | datetime.timedelta = 1.0,
    wait_exp_base: float = 2.0,
) -> _RetryContextIterator:
    """
    Iterator yielding context managers for retry blocks.
    
    Parameters: Same as retry() decorator
    
    Returns:
    Iterator yielding Attempt context managers
    """

Usage Examples:

import stamina

# Basic context manager usage
def fetch_with_context(url):
    for attempt in stamina.retry_context(on=httpx.HTTPError, attempts=3):
        with attempt:
            response = httpx.get(url)
            response.raise_for_status()
            return response.json()

# Access attempt information
def fetch_with_logging(url):
    for attempt in stamina.retry_context(on=httpx.HTTPError, attempts=5):
        with attempt:
            print(f"Attempt {attempt.num}, next wait: {attempt.next_wait}s")
            response = httpx.get(url)
            response.raise_for_status()
            return response.json()

# Async context manager  
async def fetch_async_context(url):
    async for attempt in stamina.retry_context(on=httpx.HTTPError, attempts=3):
        with attempt:
            async with httpx.AsyncClient() as client:
                response = await client.get(url)
                response.raise_for_status()
                return response.json()

# Complex retry logic
def complex_operation():
    for attempt in stamina.retry_context(
        on=lambda e: isinstance(e, ValueError) and "temporary" in str(e),
        attempts=10,
        timeout=60.0
    ):
        with attempt:
            if attempt.num > 3:
                # Change strategy after several attempts
                result = fallback_operation()
            else:
                result = primary_operation()
            
            if not is_valid(result):
                raise ValueError("temporary validation failure")
            
            return result

Attempt Context Manager

Individual context managers yielded by retry_context that provide information about the current retry attempt.

class Attempt:
    """
    Context manager for individual retry attempts.
    
    Yielded by retry_context() iterator.
    """
    
    @property
    def num(self) -> int:
        """Current attempt number (1-based)."""
    
    @property 
    def next_wait(self) -> float:
        """
        Seconds to wait before next attempt if this attempt fails.
        
        Warning: This is a jitter-less lower bound, actual wait may be higher.
        """

Usage Examples:

# Monitor retry progress
def monitored_operation():
    for attempt in stamina.retry_context(on=Exception, attempts=5):
        with attempt:
            print(f"Starting attempt {attempt.num}")
            if attempt.num > 1:
                print(f"Will wait {attempt.next_wait}s if this fails")
            
            result = risky_operation()
            print(f"Attempt {attempt.num} succeeded")
            return result

# Conditional logic based on attempt number
def adaptive_operation():
    for attempt in stamina.retry_context(on=ValueError, attempts=10):
        with attempt:
            if attempt.num <= 3:
                timeout = 5.0  # Short timeout for early attempts
            else:
                timeout = 30.0  # Longer timeout for later attempts
                
            return operation_with_timeout(timeout)

Exception Handling

Exception Type Specifications

The on parameter accepts various formats for specifying which exceptions to retry:

# Single exception type
@stamina.retry(on=httpx.HTTPError)

# Multiple exception types  
@stamina.retry(on=(httpx.HTTPError, ConnectionError, TimeoutError))

# Predicate function for custom logic
def http_5xx_errors(exc):
    return (isinstance(exc, httpx.HTTPStatusError) and 
            500 <= exc.response.status_code < 600)

@stamina.retry(on=http_5xx_errors)

Predicate Functions

Custom predicate functions provide fine-grained control over retry conditions:

def should_retry_db_error(exc):
    """Retry on temporary database errors but not schema errors."""
    if isinstance(exc, DatabaseError):
        # Retry on connection issues and deadlocks
        return any(msg in str(exc).lower() for msg in 
                  ['connection', 'timeout', 'deadlock', 'lock wait'])
    return False

@stamina.retry(on=should_retry_db_error, attempts=5)
def database_operation():
    return execute_query()

Backoff Configuration

Exponential Backoff Formula

The backoff delay for attempt number N is calculated as:

min(wait_max, wait_initial * wait_exp_base^(N-1) + random(0, wait_jitter))

Parameter Examples

# Fast retries for quick operations
@stamina.retry(
    on=ConnectionError,
    attempts=10,
    wait_initial=0.01,    # 10ms initial
    wait_max=1.0,         # Cap at 1 second
    wait_jitter=0.1       # Small jitter
)

# Conservative retries for expensive operations  
@stamina.retry(
    on=Exception,
    attempts=5,
    wait_initial=1.0,     # 1 second initial
    wait_max=60.0,        # Cap at 1 minute
    wait_jitter=10.0,     # High jitter
    wait_exp_base=3.0     # Faster exponential growth
)

# Linear backoff (set exp_base=1.0)
@stamina.retry(
    on=ValueError,
    wait_initial=2.0,
    wait_exp_base=1.0,    # Linear progression
    wait_jitter=0.5
)

Install with Tessl CLI