0
# Stamina
1
2
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.
3
4
## Package Information
5
6
- **Package Name**: stamina
7
- **Language**: Python
8
- **Installation**: `pip install stamina`
9
- **Python Support**: 3.8+
10
- **Dependencies**: tenacity, typing-extensions (Python < 3.10)
11
12
## Core Imports
13
14
```python
15
import stamina
16
```
17
18
Common imports for specific functionality:
19
20
```python
21
from stamina import retry, retry_context
22
from stamina import RetryingCaller, AsyncRetryingCaller
23
from stamina import set_active, set_testing
24
from stamina.instrumentation import set_on_retry_hooks
25
```
26
27
## Basic Usage
28
29
```python
30
import stamina
31
import httpx
32
33
# Decorator approach - retry on specific exceptions
34
@stamina.retry(on=httpx.HTTPError, attempts=3, timeout=30.0)
35
def fetch_data(url):
36
response = httpx.get(url)
37
response.raise_for_status()
38
return response.json()
39
40
# Context manager approach - manual retry control
41
def process_data():
42
for attempt in stamina.retry_context(on=ValueError, attempts=5):
43
with attempt:
44
# Code that might raise ValueError
45
result = risky_operation()
46
return result
47
48
# Caller approach - reusable retry configuration
49
caller = stamina.RetryingCaller(attempts=5, timeout=60.0)
50
result = caller(httpx.HTTPError, httpx.get, "https://api.example.com/data")
51
52
# Async support
53
@stamina.retry(on=httpx.HTTPError, attempts=3)
54
async def fetch_async(url):
55
async with httpx.AsyncClient() as client:
56
response = await client.get(url)
57
response.raise_for_status()
58
return response.json()
59
```
60
61
## Architecture
62
63
Stamina's architecture centers around three core patterns:
64
65
- **Decorator Pattern**: `@stamina.retry()` provides the simplest interface for function-level retries
66
- **Context Manager Pattern**: `stamina.retry_context()` offers manual control over retry loops with explicit attempt handling
67
- **Caller Pattern**: `RetryingCaller` and `AsyncRetryingCaller` enable reusable retry configurations for multiple operations
68
69
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.
70
71
## Capabilities
72
73
### Retry Decorators and Context Managers
74
75
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.
76
77
```python { .api }
78
def retry(
79
*,
80
on: ExcOrPredicate,
81
attempts: int | None = 10,
82
timeout: float | datetime.timedelta | None = 45.0,
83
wait_initial: float | datetime.timedelta = 0.1,
84
wait_max: float | datetime.timedelta = 5.0,
85
wait_jitter: float | datetime.timedelta = 1.0,
86
wait_exp_base: float = 2.0,
87
) -> Callable[[Callable[P, T]], Callable[P, T]]:
88
"""Decorator that retries decorated function if specified exceptions are raised."""
89
90
def retry_context(
91
on: ExcOrPredicate,
92
attempts: int | None = 10,
93
timeout: float | datetime.timedelta | None = 45.0,
94
wait_initial: float | datetime.timedelta = 0.1,
95
wait_max: float | datetime.timedelta = 5.0,
96
wait_jitter: float | datetime.timedelta = 1.0,
97
wait_exp_base: float = 2.0,
98
) -> _RetryContextIterator:
99
"""Iterator yielding context managers for retry blocks."""
100
101
class Attempt:
102
"""Context manager for individual retry attempts."""
103
num: int # Current attempt number (1-based)
104
next_wait: float # Seconds to wait before next attempt
105
```
106
107
[Retry Decorators and Context Managers](./retry-core.md)
108
109
### Retry Callers
110
111
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.
112
113
```python { .api }
114
class RetryingCaller:
115
"""Reusable caller for retrying functions with pre-configured parameters."""
116
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): ...
117
def __call__(self, on, callable_, *args, **kwargs): ...
118
def on(self, exception_type) -> BoundRetryingCaller: ...
119
120
class AsyncRetryingCaller:
121
"""Async version of RetryingCaller."""
122
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): ...
123
async def __call__(self, on, callable_, *args, **kwargs): ...
124
def on(self, exception_type) -> BoundAsyncRetryingCaller: ...
125
```
126
127
[Retry Callers](./retry-callers.md)
128
129
### Configuration and Testing
130
131
Global configuration functions for activating/deactivating retry behavior and enabling test mode with modified retry parameters for faster test execution.
132
133
```python { .api }
134
def is_active() -> bool:
135
"""Check whether retrying is globally active."""
136
137
def set_active(active: bool) -> None:
138
"""Globally activate or deactivate retrying."""
139
140
def is_testing() -> bool:
141
"""Check whether test mode is enabled."""
142
143
def set_testing(testing: bool, *, attempts: int = 1, cap: bool = False):
144
"""Activate/deactivate test mode with configurable behavior."""
145
```
146
147
[Configuration and Testing](./configuration.md)
148
149
### Instrumentation and Hooks
150
151
Comprehensive instrumentation system with built-in hooks for logging, Prometheus metrics, and structured logging, plus support for custom hooks and observability integrations.
152
153
```python { .api }
154
# Hook management
155
def set_on_retry_hooks(hooks: Iterable[RetryHook | RetryHookFactory] | None) -> None:
156
"""Set hooks called when retries are scheduled."""
157
158
def get_on_retry_hooks() -> tuple[RetryHook, ...]:
159
"""Get currently active retry hooks."""
160
161
# Built-in hooks
162
LoggingOnRetryHook: RetryHookFactory # Standard library logging
163
StructlogOnRetryHook: RetryHookFactory # Structlog integration
164
PrometheusOnRetryHook: RetryHookFactory # Prometheus metrics
165
166
# Custom hooks
167
class RetryHook(Protocol):
168
"""Protocol for retry hook callables."""
169
def __call__(self, details: RetryDetails) -> None | AbstractContextManager[None]: ...
170
```
171
172
[Instrumentation and Hooks](./instrumentation.md)
173
174
## Types
175
176
```python { .api }
177
# Core type definitions
178
from typing import Type, Tuple, Union, Callable, Iterator, AsyncIterator, TypeVar, ParamSpec
179
from dataclasses import dataclass
180
import datetime
181
182
P = ParamSpec("P")
183
T = TypeVar("T")
184
185
ExcOrPredicate = Union[
186
Type[Exception],
187
Tuple[Type[Exception], ...],
188
Callable[[Exception], bool]
189
]
190
191
class _RetryContextIterator:
192
"""Iterator that yields Attempt context managers for retry loops."""
193
def __iter__(self) -> Iterator[Attempt]: ...
194
def __aiter__(self) -> AsyncIterator[Attempt]: ...
195
196
@dataclass(frozen=True)
197
class RetryDetails:
198
"""Details about retry attempt passed to hooks."""
199
name: str # Name of callable being retried
200
args: tuple[object, ...] # Positional arguments
201
kwargs: dict[str, object] # Keyword arguments
202
retry_num: int # Retry attempt number (starts at 1)
203
wait_for: float # Seconds to wait before next attempt
204
waited_so_far: float # Total seconds waited so far
205
caused_by: Exception # Exception that triggered retry
206
207
@dataclass(frozen=True)
208
class RetryHookFactory:
209
"""Wraps callable that returns RetryHook for delayed initialization."""
210
hook_factory: Callable[[], RetryHook]
211
```