-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathretry.py
More file actions
118 lines (88 loc) · 3.58 KB
/
Copy pathretry.py
File metadata and controls
118 lines (88 loc) · 3.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
"""
Generic retry utilities with exponential backoff.
Provides consistent retry behavior across all operations.
"""
import functools
import time
from collections.abc import Callable
from typing import Any
from utils.logger import get_logger
logger = get_logger(__name__)
def retry_on_exception(
exceptions: tuple[type[Exception], ...] = (Exception,), max_retries: int = 3, base_delay: float = 0.1, exponential_backoff: bool = True, log_retries: bool = True
):
"""
Decorator for retrying operations with exponential backoff.
Args:
exceptions: Tuple of exception types to catch and retry
max_retries: Maximum number of retry attempts
base_delay: Base delay in seconds between retries
exponential_backoff: Use exponential backoff (delay = base_delay * 2^attempt)
log_retries: Log retry attempts
Returns:
Decorated function that retries on specified exceptions
Example:
@retry_on_exception(
exceptions=(sqlite3.OperationalError,),
max_retries=5,
base_delay=0.05
)
def insert_data(conn, data):
conn.execute("INSERT INTO table VALUES (?)", data)
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt == max_retries - 1:
raise
if exponential_backoff:
delay = base_delay * (2**attempt)
else:
delay = base_delay
if log_retries:
logger.warning(f"Retry {attempt + 1}/{max_retries} for {func.__name__} after {delay:.3f}s due to: {type(e).__name__}: {e}")
time.sleep(delay)
if last_exception:
raise last_exception
return wrapper
return decorator
def retry_on_db_locked(max_retries: int = 3, base_delay: float = 0.1):
"""
Specialized retry decorator for database locked errors.
Wrapper around retry_on_exception with sqlite3.OperationalError filter.
Args:
max_retries: Maximum number of retry attempts
base_delay: Base delay in seconds between retries
Returns:
Decorated function that retries on database locked errors
"""
import sqlite3
def is_db_locked(e: Exception) -> bool:
"""Check if exception is a database locked error."""
return isinstance(e, sqlite3.OperationalError) and "database is locked" in str(e).lower()
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except sqlite3.OperationalError as e:
if not is_db_locked(e):
raise
last_exception = e
if attempt == max_retries - 1:
raise
delay = base_delay * (2**attempt)
logger.warning(f"Database locked, retry {attempt + 1}/{max_retries} for {func.__name__} after {delay:.3f}s")
time.sleep(delay)
if last_exception:
raise last_exception
return wrapper
return decorator