0
# Retry Decorators and Context Managers
1

2
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.
3

4
## Capabilities
5

6
### Retry Decorator
7

8
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.
9

10
```python { .api }
11
def retry(
12
    *,
13
    on: ExcOrPredicate,
14
    attempts: int | None = 10,
15
    timeout: float | datetime.timedelta | None = 45.0,
16
    wait_initial: float | datetime.timedelta = 0.1,
17
    wait_max: float | datetime.timedelta = 5.0,
18
    wait_jitter: float | datetime.timedelta = 1.0,
19
    wait_exp_base: float = 2.0,
20
) -> Callable[[Callable[P, T]], Callable[P, T]]:
21
    """
22
    Decorator that retries decorated function if specified exceptions are raised.
23
    
24
    The backoff delays grow exponentially with jitter:
25
    min(wait_max, wait_initial * wait_exp_base^(attempt - 1) + random(0, wait_jitter))
26
    
27
    Parameters:
28
    - on: Exception(s) or predicate function to retry on (required)
29
    - attempts: Maximum total attempts (None for unlimited)
30
    - timeout: Maximum total time for all retries  
31
    - wait_initial: Minimum backoff before first retry
32
    - wait_max: Maximum backoff time at any point
33
    - wait_jitter: Maximum random jitter added to backoff
34
    - wait_exp_base: Exponential base for backoff calculation
35
    
36
    All time parameters accept float (seconds) or datetime.timedelta objects.
37
    """
38
```
39

40
**Usage Examples:**
41

42
```python
43
import stamina
44
import httpx
45
import datetime as dt
46

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

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

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

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

71
# Timedelta parameters
72
@stamina.retry(
73
    on=ValueError,
74
    timeout=dt.timedelta(minutes=2),
75
    wait_initial=dt.timedelta(milliseconds=100),
76
    wait_max=dt.timedelta(seconds=10)
77
)
78
def process_with_timedeltas():
79
    return risky_operation()
80

81
# Async function support
82
@stamina.retry(on=httpx.HTTPError, attempts=3)
83
async def fetch_async(url):
84
    async with httpx.AsyncClient() as client:
85
        response = await client.get(url)
86
        response.raise_for_status()
87
        return response.json()
88
```
89

90
### Retry Context Manager
91

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

94
```python { .api }
95
def retry_context(
96
    on: ExcOrPredicate,
97
    attempts: int | None = 10,
98
    timeout: float | datetime.timedelta | None = 45.0,
99
    wait_initial: float | datetime.timedelta = 0.1,
100
    wait_max: float | datetime.timedelta = 5.0,
101
    wait_jitter: float | datetime.timedelta = 1.0,
102
    wait_exp_base: float = 2.0,
103
) -> _RetryContextIterator:
104
    """
105
    Iterator yielding context managers for retry blocks.
106
    
107
    Parameters: Same as retry() decorator
108
    
109
    Returns:
110
    Iterator yielding Attempt context managers
111
    """
112
```
113

114
**Usage Examples:**
115

116
```python
117
import stamina
118

119
# Basic context manager usage
120
def fetch_with_context(url):
121
    for attempt in stamina.retry_context(on=httpx.HTTPError, attempts=3):
122
        with attempt:
123
            response = httpx.get(url)
124
            response.raise_for_status()
125
            return response.json()
126

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

136
# Async context manager  
137
async def fetch_async_context(url):
138
    async for attempt in stamina.retry_context(on=httpx.HTTPError, attempts=3):
139
        with attempt:
140
            async with httpx.AsyncClient() as client:
141
                response = await client.get(url)
142
                response.raise_for_status()
143
                return response.json()
144

145
# Complex retry logic
146
def complex_operation():
147
    for attempt in stamina.retry_context(
148
        on=lambda e: isinstance(e, ValueError) and "temporary" in str(e),
149
        attempts=10,
150
        timeout=60.0
151
    ):
152
        with attempt:
153
            if attempt.num > 3:
154
                # Change strategy after several attempts
155
                result = fallback_operation()
156
            else:
157
                result = primary_operation()
158
            
159
            if not is_valid(result):
160
                raise ValueError("temporary validation failure")
161
            
162
            return result
163
```
164

165
### Attempt Context Manager
166

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

169
```python { .api }
170
class Attempt:
171
    """
172
    Context manager for individual retry attempts.
173
    
174
    Yielded by retry_context() iterator.
175
    """
176
    
177
    @property
178
    def num(self) -> int:
179
        """Current attempt number (1-based)."""
180
    
181
    @property 
182
    def next_wait(self) -> float:
183
        """
184
        Seconds to wait before next attempt if this attempt fails.
185
        
186
        Warning: This is a jitter-less lower bound, actual wait may be higher.
187
        """
188
```
189

190
**Usage Examples:**
191

192
```python
193
# Monitor retry progress
194
def monitored_operation():
195
    for attempt in stamina.retry_context(on=Exception, attempts=5):
196
        with attempt:
197
            print(f"Starting attempt {attempt.num}")
198
            if attempt.num > 1:
199
                print(f"Will wait {attempt.next_wait}s if this fails")
200
            
201
            result = risky_operation()
202
            print(f"Attempt {attempt.num} succeeded")
203
            return result
204

205
# Conditional logic based on attempt number
206
def adaptive_operation():
207
    for attempt in stamina.retry_context(on=ValueError, attempts=10):
208
        with attempt:
209
            if attempt.num <= 3:
210
                timeout = 5.0  # Short timeout for early attempts
211
            else:
212
                timeout = 30.0  # Longer timeout for later attempts
213
                
214
            return operation_with_timeout(timeout)
215
```
216

217
## Exception Handling
218

219
### Exception Type Specifications
220

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

223
```python
224
# Single exception type
225
@stamina.retry(on=httpx.HTTPError)
226

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

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

235
@stamina.retry(on=http_5xx_errors)
236
```
237

238
### Predicate Functions
239

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

242
```python
243
def should_retry_db_error(exc):
244
    """Retry on temporary database errors but not schema errors."""
245
    if isinstance(exc, DatabaseError):
246
        # Retry on connection issues and deadlocks
247
        return any(msg in str(exc).lower() for msg in 
248
                  ['connection', 'timeout', 'deadlock', 'lock wait'])
249
    return False
250

251
@stamina.retry(on=should_retry_db_error, attempts=5)
252
def database_operation():
253
    return execute_query()
254
```
255

256
## Backoff Configuration
257

258
### Exponential Backoff Formula
259

260
The backoff delay for attempt number N is calculated as:
261

262
```
263
min(wait_max, wait_initial * wait_exp_base^(N-1) + random(0, wait_jitter))
264
```
265

266
### Parameter Examples
267

268
```python
269
# Fast retries for quick operations
270
@stamina.retry(
271
    on=ConnectionError,
272
    attempts=10,
273
    wait_initial=0.01,    # 10ms initial
274
    wait_max=1.0,         # Cap at 1 second
275
    wait_jitter=0.1       # Small jitter
276
)
277

278
# Conservative retries for expensive operations  
279
@stamina.retry(
280
    on=Exception,
281
    attempts=5,
282
    wait_initial=1.0,     # 1 second initial
283
    wait_max=60.0,        # Cap at 1 minute
284
    wait_jitter=10.0,     # High jitter
285
    wait_exp_base=3.0     # Faster exponential growth
286
)
287

288
# Linear backoff (set exp_base=1.0)
289
@stamina.retry(
290
    on=ValueError,
291
    wait_initial=2.0,
292
    wait_exp_base=1.0,    # Linear progression
293
    wait_jitter=0.5
294
)
295
```