0
# Stop Conditions
1

2
Stop conditions determine *when* to stop retrying and give up. Tenacity provides 7+ stop strategies based on attempt counts, elapsed time, external events, and logical combinations to control retry termination precisely.
3

4
## Base Classes
5

6
### stop_base
7

8
```python { .api }
9
from tenacity.stop import stop_base
10

11
class stop_base(ABC):
12
    """
13
    Abstract base class for all stop strategies.
14
    
15
    Provides logical operators for combining stop conditions
16
    and defines the interface all stop strategies must implement.
17
    """
18
    
19
    @abstractmethod
20
    def __call__(self, retry_state: RetryCallState) -> bool:
21
        """
22
        Determine whether to stop retrying based on current state.
23
        
24
        Parameters:
25
        - retry_state: Complete state of current retry session
26
        
27
        Returns:
28
        True if retrying should stop, False to continue attempting
29
        """
30
    
31
    def __and__(self, other: 'stop_base') -> 'stop_all':
32
        """Combine with another condition using AND logic."""
33
    
34
    def __or__(self, other: 'stop_base') -> 'stop_any':
35
        """Combine with another condition using OR logic."""
36
```
37

38
### StopBaseT Type
39

40
```python { .api }
41
from tenacity.stop import StopBaseT
42

43
StopBaseT = Union[stop_base, Callable[[RetryCallState], bool]]
44
```
45

46
## Basic Stop Strategies
47

48
### stop_never
49

50
```python { .api }
51
from tenacity import stop_never
52

53
# Singleton instance - never stop retrying  
54
stop_never: stop_base  # Always returns False
55
```
56

57
**Warning**: Using `stop_never` without other stop conditions can create infinite retry loops. Always combine with retry conditions that will eventually return False.
58

59
## Attempt-Based Stopping
60

61
### stop_after_attempt
62

63
```python { .api }
64
from tenacity import stop_after_attempt
65

66
class stop_after_attempt(stop_base):
67
    """
68
    Stop after reaching maximum number of attempts.
69
    
70
    Most commonly used stop strategy - limits total attempts including
71
    the initial attempt. For example, max_attempt_number=3 allows
72
    initial attempt + 2 retries.
73
    """
74
    
75
    def __init__(self, max_attempt_number: int):
76
        """
77
        Initialize with maximum attempt count.
78
        
79
        Parameters:
80
        - max_attempt_number: Maximum number of attempts (including initial)
81
                             Must be >= 1
82
        """
83
```
84

85
### Usage Examples
86

87
```python { .api }
88
# Stop after 3 total attempts (initial + 2 retries)
89
@retry(stop=stop_after_attempt(3))
90
def limited_retries():
91
    pass
92

93
# Single attempt only (no retries)
94
@retry(stop=stop_after_attempt(1))
95
def no_retries():
96
    pass
97
```
98

99
## Time-Based Stopping
100

101
### stop_after_delay
102

103
```python { .api }
104
from tenacity import stop_after_delay
105
from tenacity._utils import time_unit_type
106

107
class stop_after_delay(stop_base):
108
    """
109
    Stop after maximum elapsed time from first attempt.
110
    
111
    Measures total time from the start of the first attempt.
112
    May exceed max_delay due to the final sleep period before
113
    the last attempt that causes the timeout.
114
    """
115
    
116
    def __init__(self, max_delay: time_unit_type):
117
        """
118
        Initialize with maximum elapsed time.
119
        
120
        Parameters:
121
        - max_delay: Maximum time to spend retrying
122
                    Can be int/float (seconds) or timedelta
123
        """
124
```
125

126
### stop_before_delay
127

128
```python { .api }
129
from tenacity import stop_before_delay
130

131
class stop_before_delay(stop_base):
132
    """
133
    Stop before next sleep would exceed maximum delay.
134
    
135
    More precise timing control than stop_after_delay.
136
    Prevents starting attempts that would exceed the time limit
137
    when including their sleep period.
138
    """
139
    
140
    def __init__(self, max_delay: time_unit_type):
141
        """
142
        Initialize with maximum delay threshold.
143
        
144
        Parameters:
145
        - max_delay: Maximum total time including next sleep
146
                    Can be int/float (seconds) or timedelta
147
        """
148
```
149

150
### Time-Based Usage Examples
151

152
```python { .api }
153
from datetime import timedelta
154

155
# Stop after 30 seconds total
156
@retry(stop=stop_after_delay(30))
157
def time_limited_operation():
158
    pass
159

160
# Stop after 5 minutes using timedelta
161
@retry(stop=stop_after_delay(timedelta(minutes=5)))
162
def long_running_operation():
163
    pass
164

165
# Precise timing - stop before exceeding 10 seconds
166
@retry(stop=stop_before_delay(10))
167
def precisely_timed_operation():
168
    pass
169
```
170

171
## Event-Based Stopping
172

173
### stop_when_event_set
174

175
```python { .api }
176
from tenacity import stop_when_event_set
177
import threading
178

179
class stop_when_event_set(stop_base):
180
    """
181
    Stop when a threading.Event is set.
182
    
183
    Enables external control over retry stopping through
184
    threading events. Useful for graceful shutdown scenarios
185
    or coordinated stopping across multiple threads.
186
    """
187
    
188
    def __init__(self, event: threading.Event):
189
        """
190
        Initialize with threading event.
191
        
192
        Parameters:
193
        - event: threading.Event that controls stopping
194
                When event.is_set() returns True, retrying stops
195
        """
196
```
197

198
### Event-Based Usage
199

200
```python { .api }
201
import threading
202

203
# Create shutdown event
204
shutdown_event = threading.Event()
205

206
@retry(stop=stop_when_event_set(shutdown_event))
207
def interruptible_operation():
208
    pass
209

210
# In another thread or signal handler
211
def shutdown_handler():
212
    shutdown_event.set()  # This will stop all retrying operations
213
```
214

215
## Logical Combinations
216

217
### stop_any
218

219
```python { .api }
220
from tenacity import stop_any
221

222
class stop_any(stop_base):
223
    """
224
    Stop if ANY of the provided conditions are true (OR logic).
225
    
226
    Combines multiple stop strategies with logical OR.
227
    Stops as soon as any condition is met.
228
    """
229
    
230
    def __init__(self, *stops: stop_base):
231
        """
232
        Initialize with stop strategies to combine.
233
        
234
        Parameters:
235
        - *stops: Variable number of stop strategies
236
        
237
        Returns True if any strategy returns True.
238
        """
239
```
240

241
### stop_all
242

243
```python { .api }
244
from tenacity import stop_all
245

246
class stop_all(stop_base):
247
    """
248
    Stop if ALL of the provided conditions are true (AND logic).
249
    
250
    Combines multiple stop strategies with logical AND.
251
    Only stops when all conditions are simultaneously met.
252
    """
253
    
254
    def __init__(self, *stops: stop_base):
255
        """
256
        Initialize with stop strategies to combine.
257
        
258
        Parameters:
259
        - *stops: Variable number of stop strategies
260
        
261
        Returns True only if all strategies return True.
262
        """
263
```
264

265
## Comprehensive Usage Examples
266

267
### Basic Patterns
268

269
```python { .api }
270
# Stop after either 5 attempts OR 30 seconds
271
@retry(stop=stop_any(
272
    stop_after_attempt(5),
273
    stop_after_delay(30)
274
))
275
def flexible_stopping():
276
    pass
277

278
# Stop only when BOTH 10 attempts AND 60 seconds are reached
279
@retry(stop=stop_all(
280
    stop_after_attempt(10),
281
    stop_after_delay(60)
282
))
283
def conservative_stopping():
284
    pass
285
```
286

287
### Advanced Combinations
288

289
```python { .api }
290
import threading
291
from datetime import timedelta
292

293
# Complex stopping logic
294
shutdown_event = threading.Event()
295

296
comprehensive_stop = stop_any(
297
    stop_after_attempt(10),           # Max 10 attempts
298
    stop_after_delay(timedelta(minutes=5)),  # Or 5 minutes  
299
    stop_when_event_set(shutdown_event)      # Or external shutdown
300
)
301

302
@retry(stop=comprehensive_stop)
303
def robust_operation():
304
    pass
305
```
306

307
### Operator Overloading
308

309
```python { .api }
310
# Use & and | operators for logical combinations
311
attempt_limit = stop_after_attempt(5)
312
time_limit = stop_after_delay(30)
313

314
# OR logic using | operator
315
flexible_stop = attempt_limit | time_limit
316

317
# AND logic using & operator  
318
strict_stop = attempt_limit & time_limit
319

320
@retry(stop=flexible_stop)
321
def operation_with_flexible_stopping():
322
    pass
323
```
324

325
### Practical Scenarios
326

327
#### API Rate Limiting
328

329
```python { .api }
330
# Stop before hitting rate limit reset time
331
@retry(
332
    stop=stop_before_delay(3600),  # API rate limit resets every hour
333
    wait=wait_exponential(multiplier=1, min=4, max=10)
334
)
335
def rate_limited_api_call():
336
    pass
337
```
338

339
#### Database Connection Retry
340

341
```python { .api }
342
# Conservative database connection with multiple stop conditions
343
@retry(stop=stop_any(
344
    stop_after_attempt(3),        # Don't overwhelm DB
345
    stop_after_delay(30),         # Don't block application startup
346
    stop_when_event_set(shutdown_event)  # Respect shutdown requests
347
))
348
def connect_to_database():
349
    pass
350
```
351

352
#### Long-Running Background Task
353

354
```python { .api }
355
# Background task that should retry until explicitly stopped
356
@retry(
357
    stop=stop_when_event_set(task_stop_event),
358
    wait=wait_exponential(multiplier=1, max=300)  # Max 5 minute wait
359
)
360
def background_sync_task():
361
    pass
362
```
363

364
#### Circuit Breaker Pattern
365

366
```python { .api }
367
# Implement circuit breaker with time-based recovery
368
failure_count = 0
369
circuit_open_time = None
370

371
def circuit_breaker_stop(retry_state):
372
    global failure_count, circuit_open_time
373
    
374
    # Open circuit after 5 failures
375
    if failure_count >= 5:
376
        if circuit_open_time is None:
377
            circuit_open_time = time.time()
378
        
379
        # Keep circuit open for 60 seconds
380
        if time.time() - circuit_open_time < 60:
381
            return True
382
        else:
383
            # Reset circuit after timeout
384
            failure_count = 0
385
            circuit_open_time = None
386
    
387
    return False
388

389
@retry(stop=circuit_breaker_stop)
390
def circuit_protected_operation():
391
    pass
392
```
393

394
### Time Unit Flexibility
395

396
```python { .api }
397
from datetime import timedelta
398

399
# All equivalent - 30 second time limit
400
@retry(stop=stop_after_delay(30))          # int seconds
401
@retry(stop=stop_after_delay(30.0))        # float seconds  
402
@retry(stop=stop_after_delay(timedelta(seconds=30)))  # timedelta
403

404
# Complex time specifications
405
@retry(stop=stop_after_delay(timedelta(minutes=2, seconds=30)))
406
def precisely_timed_operation():
407
    pass
408
```
409

410
### Graceful Shutdown Integration
411

412
```python { .api }
413
import signal
414
import threading
415

416
# Global shutdown event for coordinated stopping
417
global_shutdown = threading.Event()
418

419
def signal_handler(signum, frame):
420
    global_shutdown.set()
421

422
signal.signal(signal.SIGTERM, signal_handler)
423
signal.signal(signal.SIGINT, signal_handler)
424

425
# All retry operations respect global shutdown
426
@retry(stop=stop_any(
427
    stop_after_attempt(5),
428
    stop_when_event_set(global_shutdown)
429
))
430
def service_operation():
431
    pass
432
```
433

434
### Testing Stop Conditions
435

436
```python { .api }
437
# Testing with immediate stop
438
@retry(stop=stop_after_attempt(1))  
439
def test_no_retry():
440
    pass
441

442
# Testing with never stop (use with mock that eventually succeeds)
443
@retry(stop=stop_never, retry=retry_if_result(lambda x: x != "success"))
444
def test_until_success():
445
    pass
446
```
447

448
## Performance Considerations
449

450
### Efficient Stop Checking
451

452
```python { .api }
453
# Cheap checks first in combinations
454
efficient_stop = stop_any(
455
    stop_after_attempt(3),          # Fast: simple counter check
456
    stop_when_event_set(event),     # Fast: single flag check  
457
    stop_after_delay(60)            # Slower: time calculation
458
)
459
```
460

461
### Memory Efficient Long-Running Retries
462

463
```python { .api }
464
# For very long-running retry scenarios, prefer stop_before_delay
465
# to avoid accumulating large idle_for values in retry state
466
@retry(
467
    stop=stop_before_delay(timedelta(hours=24)),  # 24 hour limit
468
    wait=wait_exponential(multiplier=1, max=3600)  # Max 1 hour wait
469
)
470
def daily_sync_operation():
471
    pass
472
```
473

474
This comprehensive coverage of stop conditions provides precise control over retry termination, enabling robust retry behavior that respects time limits, attempt limits, and external coordination requirements.