Make retry handlers static methods

Convert handle_network_failure and handle_download_failure from instance methods to static methods. Hardcode retry params (max_retries, delays) instead of using instance state. Improves testability and removes implicit dependencies.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
2026-06-04 22:29:59 +02:00
parent 2be7b692b9
commit 25dc66fec3
5 changed files with 26 additions and 14 deletions

View File

@@ -74,22 +74,28 @@ class RecoveryStrategies:
delay = self.base_delay * (self.exponential_base ** attempt)
return min(delay, self.max_delay)
@staticmethod
def handle_network_failure(
self,
func: Callable, *args: Any, **kwargs: Any
) -> Any:
"""Handle network failures with exponential backoff retry logic."""
last_error: Optional[Exception] = None
for attempt in range(self.max_retries):
max_retries = 3
base_delay = 1.0
max_delay = 60.0
exponential_base = 2.0
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except (NetworkError, ConnectionError, TimeoutError) as exc:
last_error = exc
if attempt < self.max_retries - 1:
delay = self._calculate_delay(attempt)
if attempt < max_retries - 1:
delay = base_delay * (exponential_base ** attempt)
delay = min(delay, max_delay)
logger.warning(
"Network error on attempt %d/%d, retrying in %.1fs: %s",
attempt + 1, self.max_retries, delay, exc
attempt + 1, max_retries, delay, exc
)
import time
time.sleep(delay)
@@ -98,22 +104,28 @@ class RecoveryStrategies:
raise last_error
raise NetworkError("Network failure after retries")
@staticmethod
def handle_download_failure(
self,
func: Callable, *args: Any, **kwargs: Any
) -> Any:
"""Handle download failures with exponential backoff retry logic."""
last_error: Optional[Exception] = None
for attempt in range(self.max_retries):
max_retries = 2
base_delay = 1.0
max_delay = 60.0
exponential_base = 2.0
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except DownloadError as exc:
last_error = exc
if attempt < self.max_retries - 1:
delay = self._calculate_delay(attempt)
if attempt < max_retries - 1:
delay = base_delay * (exponential_base ** attempt)
delay = min(delay, max_delay)
logger.warning(
"Download error on attempt %d/%d, retrying in %.1fs: %s",
attempt + 1, self.max_retries, delay, exc
attempt + 1, max_retries, delay, exc
)
import time
time.sleep(delay)