or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

configuration.mdindex.mdinstrumentation.mdretry-callers.mdretry-core.md
tile.json

tessl/pypi-stamina

Production-grade retries made easy.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/stamina@25.1.x

To install, run

npx @tessl/cli install tessl/pypi-stamina@25.1.0

index.mddocs/

Stamina

Production-grade retries made easy. Stamina is an ergonomic wrapper around the Tenacity package that implements retry best practices by default, including exponential backoff with jitter, configurable retry limits, time-based bounds, and comprehensive instrumentation.

Package Information

  • Package Name: stamina
  • Language: Python
  • Installation: pip install stamina
  • Python Support: 3.8+
  • Dependencies: tenacity, typing-extensions (Python < 3.10)

Core Imports

import stamina

Common imports for specific functionality:

from stamina import retry, retry_context
from stamina import RetryingCaller, AsyncRetryingCaller
from stamina import set_active, set_testing
from stamina.instrumentation import set_on_retry_hooks

Basic Usage

import stamina
import httpx

# Decorator approach - retry on specific exceptions
@stamina.retry(on=httpx.HTTPError, attempts=3, timeout=30.0)
def fetch_data(url):
    response = httpx.get(url)
    response.raise_for_status()
    return response.json()

# Context manager approach - manual retry control  
def process_data():
    for attempt in stamina.retry_context(on=ValueError, attempts=5):
        with attempt:
            # Code that might raise ValueError
            result = risky_operation()
            return result

# Caller approach - reusable retry configuration
caller = stamina.RetryingCaller(attempts=5, timeout=60.0)
result = caller(httpx.HTTPError, httpx.get, "https://api.example.com/data")

# Async 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()

Architecture

Stamina's architecture centers around three core patterns:

  • Decorator Pattern: @stamina.retry() provides the simplest interface for function-level retries
  • Context Manager Pattern: stamina.retry_context() offers manual control over retry loops with explicit attempt handling
  • Caller Pattern: RetryingCaller and AsyncRetryingCaller enable reusable retry configurations for multiple operations

The instrumentation system uses a hook-based architecture that supports logging, metrics collection, and custom observability integrations. All retry behavior can be globally controlled for testing and debugging scenarios.

Capabilities

Retry Decorators and Context Managers

Core retry functionality including the main @retry decorator and retry_context iterator for manual retry control. Supports both synchronous and asynchronous code with configurable backoff strategies.

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."""

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."""

class Attempt:
    """Context manager for individual retry attempts."""
    num: int          # Current attempt number (1-based)
    next_wait: float  # Seconds to wait before next attempt

Retry Decorators and Context Managers

Retry Callers

Reusable retry caller classes that allow pre-configuring retry parameters and calling multiple functions with the same retry behavior. Includes both synchronous and asynchronous versions with method chaining support.

class RetryingCaller:
    """Reusable caller for retrying functions with pre-configured parameters."""
    def __init__(self, attempts=10, timeout=45.0, wait_initial=0.1, wait_max=5.0, wait_jitter=1.0, wait_exp_base=2.0): ...
    def __call__(self, on, callable_, *args, **kwargs): ...
    def on(self, exception_type) -> BoundRetryingCaller: ...

class AsyncRetryingCaller:
    """Async version of RetryingCaller."""
    def __init__(self, attempts=10, timeout=45.0, wait_initial=0.1, wait_max=5.0, wait_jitter=1.0, wait_exp_base=2.0): ...
    async def __call__(self, on, callable_, *args, **kwargs): ...
    def on(self, exception_type) -> BoundAsyncRetryingCaller: ...

Retry Callers

Configuration and Testing

Global configuration functions for activating/deactivating retry behavior and enabling test mode with modified retry parameters for faster test execution.

def is_active() -> bool: 
    """Check whether retrying is globally active."""

def set_active(active: bool) -> None:
    """Globally activate or deactivate retrying."""

def is_testing() -> bool:
    """Check whether test mode is enabled."""

def set_testing(testing: bool, *, attempts: int = 1, cap: bool = False):
    """Activate/deactivate test mode with configurable behavior."""

Configuration and Testing

Instrumentation and Hooks

Comprehensive instrumentation system with built-in hooks for logging, Prometheus metrics, and structured logging, plus support for custom hooks and observability integrations.

# Hook management
def set_on_retry_hooks(hooks: Iterable[RetryHook | RetryHookFactory] | None) -> None:
    """Set hooks called when retries are scheduled."""

def get_on_retry_hooks() -> tuple[RetryHook, ...]:
    """Get currently active retry hooks."""

# Built-in hooks  
LoggingOnRetryHook: RetryHookFactory      # Standard library logging
StructlogOnRetryHook: RetryHookFactory    # Structlog integration
PrometheusOnRetryHook: RetryHookFactory   # Prometheus metrics

# Custom hooks
class RetryHook(Protocol):
    """Protocol for retry hook callables."""
    def __call__(self, details: RetryDetails) -> None | AbstractContextManager[None]: ...

Instrumentation and Hooks

Types

# Core type definitions
from typing import Type, Tuple, Union, Callable, Iterator, AsyncIterator, TypeVar, ParamSpec
from dataclasses import dataclass
import datetime

P = ParamSpec("P")
T = TypeVar("T")

ExcOrPredicate = Union[
    Type[Exception], 
    Tuple[Type[Exception], ...], 
    Callable[[Exception], bool]
]

class _RetryContextIterator:
    """Iterator that yields Attempt context managers for retry loops."""
    def __iter__(self) -> Iterator[Attempt]: ...
    def __aiter__(self) -> AsyncIterator[Attempt]: ...

@dataclass(frozen=True)
class RetryDetails:
    """Details about retry attempt passed to hooks."""
    name: str                    # Name of callable being retried
    args: tuple[object, ...]     # Positional arguments 
    kwargs: dict[str, object]    # Keyword arguments
    retry_num: int              # Retry attempt number (starts at 1)
    wait_for: float             # Seconds to wait before next attempt
    waited_so_far: float        # Total seconds waited so far
    caused_by: Exception        # Exception that triggered retry

@dataclass(frozen=True)  
class RetryHookFactory:
    """Wraps callable that returns RetryHook for delayed initialization."""
    hook_factory: Callable[[], RetryHook]