tessl/pypi-stamina

Production-grade retries made easy.

Overview

Eval results

Files

Retry Callers

Name: tessl/pypi-stamina
Author: tessl

Reusable retry caller classes that enable pre-configuring retry parameters and applying them to multiple functions. These classes are ideal when you need consistent retry behavior across different operations or want to separate retry configuration from business logic.

Capabilities

Synchronous Retry Caller

The RetryingCaller class provides a reusable interface for calling functions with consistent retry parameters.

class RetryingCaller:
    """
    Reusable caller for retrying functions with pre-configured parameters.
    
    Instances can be reused as they create new retry contexts on each call.
    """
    
    def __init__(
        self,
        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,
    ):
        """
        Initialize retry caller with default parameters.
        
        Parameters: Same as retry() decorator
        """
    
    def __call__(
        self,
        on: ExcOrPredicate,
        callable_: Callable[P, T],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> T:
        """
        Call callable with retries if specified exceptions are raised.
        
        Parameters:
        - on: Exception(s) to retry on
        - callable_: Function to call
        - args: Positional arguments for the function
        - kwargs: Keyword arguments for the function
        
        Returns:
        Return value of the callable
        """
    
    def on(self, on: ExcOrPredicate, /) -> BoundRetryingCaller:
        """
        Create bound caller pre-configured for specific exception types.
        
        Returns:
        BoundRetryingCaller instance bound to the exception type
        """

Usage Examples:

import stamina
import httpx

# Create reusable caller with custom parameters
http_caller = stamina.RetryingCaller(
    attempts=5,
    timeout=30.0,
    wait_initial=0.5,
    wait_max=10.0
)

# Call different functions with same retry behavior
def fetch_user(user_id):
    response = httpx.get(f"/users/{user_id}")
    response.raise_for_status()
    return response.json()

def fetch_posts(user_id):
    response = httpx.get(f"/users/{user_id}/posts")
    response.raise_for_status()
    return response.json()

# Use the same caller for multiple operations
user = http_caller(httpx.HTTPError, fetch_user, 123)
posts = http_caller(httpx.HTTPError, fetch_posts, 123)

# Direct function calls
result = http_caller(
    httpx.HTTPError,
    lambda: httpx.get("https://api.example.com/data").json(),
)

# Call with additional arguments
result = http_caller(
    ValueError,
    process_data,
    input_data,
    format="json",
    validate=True
)

Bound Retry Caller

The BoundRetryingCaller class is created by calling RetryingCaller.on() and provides a simpler interface when you always retry on the same exception types.

class BoundRetryingCaller:
    """
    RetryingCaller pre-bound to specific exception types.
    
    Created by RetryingCaller.on() - do not instantiate directly.
    """
    
    def __call__(
        self, 
        callable_: Callable[P, T], 
        /, 
        *args: P.args, 
        **kwargs: P.kwargs
    ) -> T:
        """
        Call callable with bound exception handling.
        
        Parameters:
        - callable_: Function to call
        - args: Positional arguments for the function  
        - kwargs: Keyword arguments for the function
        
        Returns:
        Return value of the callable
        """

Usage Examples:

# Create caller bound to specific exception
http_caller = stamina.RetryingCaller(attempts=3, timeout=15.0)
bound_caller = http_caller.on(httpx.HTTPError)

# Cleaner syntax for repeated operations
user = bound_caller(fetch_user, 123)
posts = bound_caller(fetch_posts, 123)
profile = bound_caller(fetch_profile, 123)

# Method chaining
result = (stamina.RetryingCaller(attempts=5)
          .on(ConnectionError)
          (connect_to_service, host="localhost", port=8080))

Asynchronous Retry Caller

The AsyncRetryingCaller provides the same interface as RetryingCaller but for asynchronous functions.

class AsyncRetryingCaller:
    """
    Async version of RetryingCaller for async functions.
    """
    
    def __init__(
        self,
        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,
    ):
        """Initialize async retry caller with default parameters."""
    
    async def __call__(
        self,
        on: ExcOrPredicate,
        callable_: Callable[P, Awaitable[T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> T:
        """
        Async call callable with retries.
        
        Parameters: Same as RetryingCaller.__call__ but callable_ must be async
        
        Returns:
        Awaited return value of the callable
        """
    
    def on(self, on: ExcOrPredicate, /) -> BoundAsyncRetryingCaller:
        """Create bound async caller for specific exception types."""

Usage Examples:

import asyncio
import stamina
import httpx

# Create async retry caller
async_caller = stamina.AsyncRetryingCaller(
    attempts=3,
    timeout=20.0,
    wait_max=5.0
)

async def fetch_async(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url) 
        response.raise_for_status()
        return response.json()

async def process_async(data):
    # Simulate async processing
    await asyncio.sleep(0.1)
    if not data:
        raise ValueError("Invalid data")
    return {"processed": data}

# Use async caller
async def main():
    # Call different async functions
    data = await async_caller(httpx.HTTPError, fetch_async, "https://api.example.com/data")
    result = await async_caller(ValueError, process_async, data)
    
    # With bound caller
    bound_caller = async_caller.on(httpx.HTTPError)
    user = await bound_caller(fetch_async, "https://api.example.com/user/123")
    
    return result

# Run async code
result = asyncio.run(main())

Bound Async Retry Caller

The BoundAsyncRetryingCaller is the async equivalent of BoundRetryingCaller.

class BoundAsyncRetryingCaller:
    """
    AsyncRetryingCaller pre-bound to specific exception types.
    
    Created by AsyncRetryingCaller.on() - do not instantiate directly.
    """
    
    async def __call__(
        self,
        callable_: Callable[P, Awaitable[T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> T:
        """Async call callable with bound exception handling."""

Usage Examples:

# Bound async caller usage  
async_caller = stamina.AsyncRetryingCaller(attempts=5)
http_bound = async_caller.on(httpx.HTTPError)

async def fetch_multiple_endpoints():
    # Use same bound caller for multiple async operations
    users = await http_bound(fetch_async, "https://api.example.com/users")
    posts = await http_bound(fetch_async, "https://api.example.com/posts") 
    comments = await http_bound(fetch_async, "https://api.example.com/comments")
    
    return {"users": users, "posts": posts, "comments": comments}

Advanced Usage Patterns

Configuration Factories

Create caller factories for different retry strategies:

def create_http_caller(timeout=30.0):
    """Factory for HTTP-specific retry callers."""
    return stamina.RetryingCaller(
        attempts=3,
        timeout=timeout,
        wait_initial=0.5,
        wait_max=5.0
    ).on(httpx.HTTPError)

def create_db_caller():
    """Factory for database-specific retry callers.""" 
    return stamina.RetryingCaller(
        attempts=5,
        timeout=60.0,
        wait_initial=1.0,
        wait_max=30.0
    ).on(DatabaseError)

# Use factories
http_caller = create_http_caller(timeout=15.0)
db_caller = create_db_caller()

user = http_caller(fetch_user, 123)
records = db_caller(query_database, "SELECT * FROM users")

Caller Composition

Combine callers for complex scenarios:

# Different retry strategies for different operations
fast_caller = stamina.RetryingCaller(attempts=3, wait_max=2.0)
slow_caller = stamina.RetryingCaller(attempts=10, wait_max=30.0)

def robust_data_processing(data):
    # Quick retries for validation
    validated = fast_caller(ValueError, validate_data, data)
    
    # Longer retries for expensive processing  
    processed = slow_caller(ProcessingError, process_data, validated)
    
    # Quick retries for saving
    result = fast_caller(IOError, save_result, processed)
    
    return result

Context-Aware Callers

Use caller state for contextual retry decisions:

class ContextualRetryingCaller:
    def __init__(self, base_caller):
        self.base_caller = base_caller
        self.attempt_counts = {}
    
    def __call__(self, operation_id, on, callable_, *args, **kwargs):
        # Track attempts per operation
        self.attempt_counts[operation_id] = self.attempt_counts.get(operation_id, 0) + 1
        
        try:
            return self.base_caller(on, callable_, *args, **kwargs)
        except Exception:
            print(f"Operation {operation_id} failed after {self.attempt_counts[operation_id]} attempts")
            raise

# Usage with context
contextual_caller = ContextualRetryingCaller(
    stamina.RetryingCaller(attempts=3)
)

result = contextual_caller(
    "fetch_user_123", 
    httpx.HTTPError, 
    fetch_user, 
    123
)

Install with Tessl CLI