0
# Retry Logic and Exception Handling
1
2
Retry mechanism for flaky operations and sophisticated exception catching with inspection capabilities.
3
4
## Capabilities
5
6
### Retry Decorator
7
8
Configurable retry logic for functions and steps, with support for attempts, delays, and exception filtering.
9
10
```python { .api }
11
def ensure(*, attempts: Optional[int] = None,
12
delay: Optional[Union[float, int, Callable[[int], Union[float, int]]]] = None,
13
swallow: Optional[Union[Type[BaseException], Tuple[Type[BaseException], ...]]] = None) -> Ensure:
14
"""
15
Decorator to add retry logic to a function or coroutine.
16
17
Args:
18
attempts: The maximum number of times the function can be called.
19
To run the function and retry once if an exception is raised, set attempts=2.
20
To run the function and retry twice, set attempts=3.
21
delay: The delay between attempts, which can be a fixed value or a callable
22
returning a value.
23
swallow: The exception(s) to be caught and retried.
24
25
Returns:
26
An Ensure instance configured with the provided or default parameters.
27
"""
28
29
class Ensure:
30
"""
31
Provides functionality to ensure a function succeeds within a specified number of attempts.
32
33
This class retries a given function or coroutine function a specified number of times,
34
optionally with a delay between attempts, and can log each attempt.
35
"""
36
37
def __init__(self, *, attempts: int = 3,
38
delay: Union[float, int, Callable[[int], Union[float, int]]] = 0.0,
39
swallow: Union[Type[BaseException], Tuple[Type[BaseException], ...]] = BaseException,
40
logger: Optional[Callable] = None): ...
41
42
def __call__(self, fn: Callable) -> Callable: ...
43
```
44
45
#### Usage Example
46
47
```python
48
from vedro import scenario, given, when, then, ensure
49
50
@scenario("Flaky API operation")
51
def test_api_with_retries():
52
53
@given("API client")
54
def setup():
55
return APIClient("https://flaky-service.com")
56
57
@when("making request with retry logic")
58
def action(client):
59
# Retry up to 3 times with 1 second delay
60
@ensure(attempts=3, delay=1.0, swallow=(ConnectionError, TimeoutError))
61
def make_request():
62
return client.get("/users/123")
63
64
return make_request()
65
66
@then("request eventually succeeds")
67
def verification(response):
68
assert response.status_code == 200
69
assert "user_id" in response.json()
70
```
71
72
### Exception Handling Context Manager
73
74
Sophisticated exception catching with inspection capabilities for both sync and async code.
75
76
```python { .api }
77
class catched:
78
"""
79
Context manager for catching and inspecting exceptions.
80
81
Supports both synchronous and asynchronous contexts.
82
Can be used to verify that specific exceptions are raised
83
and inspect their properties.
84
"""
85
86
def __init__(self, expected_exc = BaseException):
87
"""
88
Initialize exception catcher.
89
90
Args:
91
expected_exc: Exception type(s) to catch. Can be single type
92
or tuple of types. Defaults to BaseException.
93
"""
94
95
@property
96
def type(self) -> Type[BaseException] | None:
97
"""The type of the caught exception, if any."""
98
99
@property
100
def value(self) -> BaseException | None:
101
"""The caught exception instance, if any."""
102
103
@property
104
def traceback(self) -> TracebackType | None:
105
"""The traceback of the caught exception, if any."""
106
107
def __enter__(self) -> "catched": ...
108
def __exit__(self, exc_type, exc_value, traceback) -> bool: ...
109
def __aenter__(self) -> "catched": ...
110
def __aexit__(self, exc_type, exc_value, traceback) -> bool: ...
111
def __repr__(self) -> str: ...
112
```
113
114
#### Usage Example - Basic Exception Catching
115
116
```python
117
from vedro import scenario, given, when, then, catched
118
119
@scenario("Invalid input handling")
120
def test_invalid_input():
121
122
@given("invalid user data")
123
def setup():
124
return {
125
"name": "", # Invalid: empty name
126
"email": "not-an-email", # Invalid: malformed email
127
"age": -5 # Invalid: negative age
128
}
129
130
@when("attempting to create user")
131
def action(invalid_data):
132
with catched(ValidationError) as caught:
133
create_user(invalid_data)
134
return caught
135
136
@then("appropriate validation error is raised")
137
def verification(caught_exception):
138
# Verify exception was caught
139
assert caught_exception.value is not None
140
assert caught_exception.type == ValidationError
141
142
# Inspect exception details
143
error = caught_exception.value
144
assert "name" in error.field_errors
145
assert "email" in error.field_errors
146
assert "age" in error.field_errors
147
assert "validation failed" in str(error)
148
```
149
150
#### Usage Example - Multiple Exception Types
151
152
```python
153
@scenario("Database operation error handling")
154
def test_database_errors():
155
156
@when("database operation fails")
157
def action():
158
# Catch multiple possible exception types
159
with catched((ConnectionError, TimeoutError, DatabaseError)) as caught:
160
unreliable_database_operation()
161
return caught
162
163
@then("appropriate error handling occurs")
164
def verification(caught_exception):
165
assert caught_exception.value is not None
166
167
# Handle different exception types appropriately
168
if caught_exception.type == ConnectionError:
169
assert "connection" in str(caught_exception.value)
170
elif caught_exception.type == TimeoutError:
171
assert "timeout" in str(caught_exception.value)
172
elif caught_exception.type == DatabaseError:
173
assert "database" in str(caught_exception.value)
174
```
175
176
#### Usage Example - Async Exception Handling
177
178
```python
179
@scenario("Async operation error handling")
180
def test_async_errors():
181
182
@when("async operation fails")
183
async def action():
184
async with catched(AsyncOperationError) as caught:
185
await failing_async_operation()
186
return caught
187
188
@then("async error is properly caught")
189
def verification(caught_exception):
190
assert caught_exception.value is not None
191
assert caught_exception.type == AsyncOperationError
192
193
# Check async-specific error details
194
error = caught_exception.value
195
assert error.operation_id is not None
196
assert error.retry_count > 0
197
```
198
199
### Combined Retry and Exception Handling
200
201
Combining retry logic with exception handling for robust testing patterns.
202
203
```python
204
@scenario("Resilient API operations")
205
def test_resilient_operations():
206
207
@when("performing operation with retries and error handling")
208
def action():
209
@ensure(attempts=3, delay=0.5, swallow=(ConnectionError, TimeoutError))
210
def reliable_operation():
211
# This might fail with connection issues but will be retried
212
return api_call_that_might_fail()
213
214
try:
215
result = reliable_operation()
216
return {"success": True, "result": result}
217
except Exception as e:
218
with catched(type(e)) as caught:
219
raise
220
return {"success": False, "caught": caught}
221
222
@then("operation handles failures gracefully")
223
def verification(outcome):
224
if outcome["success"]:
225
assert outcome["result"] is not None
226
else:
227
caught = outcome["caught"]
228
assert caught.value is not None
229
# Verify it's not a retryable error (those should have been handled)
230
assert not isinstance(caught.value, (ConnectionError, TimeoutError))
231
```
232
233
## Types
234
235
### Type Definitions
236
237
Type definitions for retry and exception handling components.
238
239
```python { .api }
240
from types import TracebackType
241
from typing import Type, Union, Tuple, Callable, Optional
242
243
# Retry mechanism types
244
AttemptType = int
245
DelayValueType = Union[float, int]
246
DelayCallableType = Callable[[AttemptType], DelayValueType]
247
DelayType = Union[DelayValueType, DelayCallableType]
248
249
ExceptionType = Type[BaseException]
250
SwallowExceptionType = Union[Tuple[ExceptionType, ...], ExceptionType]
251
252
LoggerType = Callable[[Callable, AttemptType, Union[BaseException, None]], Any]
253
254
# Exception handling types
255
ExpectedExcType = Union[Type[BaseException], Tuple[Type[BaseException], ...]]
256
```
257
258
## Advanced Patterns
259
260
### Retry with Exponential Backoff
261
262
Combine retry logic with sophisticated delay strategies:
263
264
```python
265
@scenario("Exponential backoff retry")
266
def test_exponential_backoff():
267
268
@when("using exponential backoff for unreliable service")
269
def action():
270
def exponential_delay(attempt: int) -> float:
271
return min(2 ** attempt, 30) # Cap at 30 seconds
272
273
@ensure(attempts=5, delay=exponential_delay, swallow=ServiceUnavailableError)
274
def call_unreliable_service():
275
return external_api_call()
276
277
return call_unreliable_service()
278
279
@then("service call eventually succeeds with backoff")
280
def verification(result):
281
assert result is not None
282
assert result.status == "success"
283
```
284
285
### Exception Chain Inspection
286
287
Test exception chaining and cause relationships:
288
289
```python
290
@scenario("Exception chaining")
291
def test_exception_chain():
292
293
@when("nested operation fails")
294
def action():
295
with catched(ServiceError) as caught:
296
# This should catch a ServiceError caused by a DatabaseError
297
complex_service_operation()
298
return caught
299
300
@then("exception chain is preserved")
301
def verification(caught_exception):
302
service_error = caught_exception.value
303
assert service_error is not None
304
assert service_error.__cause__ is not None
305
assert type(service_error.__cause__) == DatabaseError
306
307
# Verify the full chain
308
db_error = service_error.__cause__
309
assert db_error.connection_string is not None
310
assert "service unavailable" in str(service_error)
311
assert "connection failed" in str(db_error)
312
```