From fffc4adf83b6e8278f4b452532d3e2860eeef1f7 Mon Sep 17 00:00:00 2001 From: Somebody Else Date: Thu, 18 Jun 2026 22:10:34 +0300 Subject: [PATCH 1/3] add retiries and degradation processing --- pyproject.toml | 2 +- src/ragflow_orchestrator/retry.py | 134 ++++++++++++++ src/ragflow_orchestrator/templates/bitrix.py | 11 ++ .../templates/confluence_wiki.py | 7 + .../templates/github_template.py | 7 + .../templates/gitlab_template.py | 7 + src/ragflow_orchestrator/templates/jira.py | 7 + src/ragflow_orchestrator/templates/pypi.py | 7 + .../templates/web_crawl.py | 7 + tests/test_retry.py | 174 ++++++++++++++++++ 10 files changed, 362 insertions(+), 1 deletion(-) create mode 100644 src/ragflow_orchestrator/retry.py create mode 100644 tests/test_retry.py diff --git a/pyproject.toml b/pyproject.toml index d15e878..902faa3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "ragflow_orchestrator" -version = "0.1.14" +version = "0.1.15" description = "Universal RAG orchestration layer with pluggable vector DB adapters" readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.11" diff --git a/src/ragflow_orchestrator/retry.py b/src/ragflow_orchestrator/retry.py new file mode 100644 index 0000000..fcda045 --- /dev/null +++ b/src/ragflow_orchestrator/retry.py @@ -0,0 +1,134 @@ +"""Retry utilities with exponential backoff for resilient HTTP and operation handling.""" + +from __future__ import annotations + +import logging +import random +import time +from functools import wraps +from typing import Callable, ParamSpec, TypeVar + +logger = logging.getLogger(__name__) + +P = ParamSpec("P") +T = TypeVar("T") + + +def retry( + max_retries: int = 3, + initial_delay: float = 1.0, + max_delay: float = 32.0, + backoff_factor: float = 2.0, + jitter: bool = True, +) -> Callable[[Callable[P, T]], Callable[P, T]]: + """ + Decorator for retrying functions with exponential backoff. + + Args: + max_retries: Maximum number of retry attempts (total attempts = max_retries + 1). + initial_delay: Initial delay in seconds before first retry. + max_delay: Maximum delay in seconds between retries. + backoff_factor: Multiplier for delay after each failure. + jitter: Whether to add random jitter to delay. + + Example: + @retry(max_retries=3, initial_delay=1.0) + def fetch_url(url: str) -> str: + ... + """ + + def decorator(func: Callable[P, T]) -> Callable[P, T]: + @wraps(func) + def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: + delay = initial_delay + last_exception: Exception | None = None + + for attempt in range(max_retries + 1): + try: + return func(*args, **kwargs) + except Exception as exc: + last_exception = exc + if attempt < max_retries: + if jitter: + jittered_delay = delay * (0.5 + random.random()) + else: + jittered_delay = delay + jittered_delay = min(jittered_delay, max_delay) + logger.warning( + f"Attempt {attempt + 1}/{max_retries + 1} failed for {func.__name__}: {exc}. " + f"Retrying in {jittered_delay:.2f}s..." + ) + time.sleep(jittered_delay) + delay = min(delay * backoff_factor, max_delay) + else: + logger.error(f"All {max_retries + 1} attempts failed for {func.__name__}: {exc}") + + raise last_exception + + return wrapper + + return decorator + + +def retry_or_log( + max_retries: int = 3, + initial_delay: float = 1.0, + max_delay: float = 32.0, + backoff_factor: float = 2.0, + jitter: bool = True, + on_failure: Callable[[Exception, str], None] | None = None, +) -> Callable[[Callable[P, T]], Callable[P, T | None]]: + """ + Decorator for retrying functions with exponential backoff, returning None on final failure. + + Instead of raising an exception, logs the error and returns None after all retries are exhausted. + Useful for gracefully handling non-critical operations. + + Args: + max_retries: Maximum number of retry attempts. + initial_delay: Initial delay in seconds before first retry. + max_delay: Maximum delay in seconds between retries. + backoff_factor: Multiplier for delay after each failure. + jitter: Whether to add random jitter to delay. + on_failure: Optional callback to handle failure (func, error_msg) -> None. + + Example: + @retry_or_log(max_retries=2) + def fetch_url(url: str) -> str | None: + ... + """ + + def decorator(func: Callable[P, T]) -> Callable[P, T | None]: + @wraps(func) + def wrapper(*args: P.args, **kwargs: P.kwargs) -> T | None: + delay = initial_delay + last_exception: Exception | None = None + + for attempt in range(max_retries + 1): + try: + return func(*args, **kwargs) + except Exception as exc: + last_exception = exc + if attempt < max_retries: + if jitter: + jittered_delay = delay * (0.5 + random.random()) + else: + jittered_delay = delay + jittered_delay = min(jittered_delay, max_delay) + logger.warning( + f"Attempt {attempt + 1}/{max_retries + 1} failed for {func.__name__}: {exc}. " + f"Retrying in {jittered_delay:.2f}s..." + ) + time.sleep(jittered_delay) + delay = min(delay * backoff_factor, max_delay) + else: + error_msg = f"All {max_retries + 1} attempts failed for {func.__name__}: {exc}" + logger.error(error_msg) + if on_failure: + on_failure(exc, error_msg) + + return None + + return wrapper + + return decorator diff --git a/src/ragflow_orchestrator/templates/bitrix.py b/src/ragflow_orchestrator/templates/bitrix.py index 56a6bd5..968112c 100644 --- a/src/ragflow_orchestrator/templates/bitrix.py +++ b/src/ragflow_orchestrator/templates/bitrix.py @@ -4,6 +4,7 @@ from typing import Any, Dict, Iterable, List, Tuple from urllib import parse, request +from ragflow_orchestrator.retry import retry from ragflow_orchestrator.templates.base import BaseIngestionTemplate from ragflow_orchestrator.templates.models import ( BitrixConfig, @@ -266,6 +267,16 @@ def _request_json( config: BitrixConfig, method: str, params: Dict[str, Any] | None, + ) -> Dict[str, Any]: + """Fetch JSON from Bitrix API with retry logic.""" + return BitrixTemplate._fetch_json_with_retry(config, method, params) + + @staticmethod + @retry(max_retries=3, initial_delay=1.0, max_delay=10.0, backoff_factor=2.0) + def _fetch_json_with_retry( + config: BitrixConfig, + method: str, + params: Dict[str, Any] | None, ) -> Dict[str, Any]: url = f"{config.base_url}/{method}.json" data_bytes = json.dumps(params or {}).encode("utf-8") diff --git a/src/ragflow_orchestrator/templates/confluence_wiki.py b/src/ragflow_orchestrator/templates/confluence_wiki.py index f130378..3a85eab 100644 --- a/src/ragflow_orchestrator/templates/confluence_wiki.py +++ b/src/ragflow_orchestrator/templates/confluence_wiki.py @@ -4,6 +4,7 @@ import json from urllib import parse, request +from ragflow_orchestrator.retry import retry from ragflow_orchestrator.templates.base import BaseIngestionTemplate from ragflow_orchestrator.templates.models import ConfluenceWikiConfig, IngestionError, TemplateRunReport from ragflow_orchestrator.templates.utils import extract_text_from_html @@ -87,6 +88,12 @@ def _get_page(self, config: ConfluenceWikiConfig, page_id: str) -> dict: @staticmethod def _request_json(config: ConfluenceWikiConfig, url: str) -> dict: + """Fetch JSON from URL with retry logic.""" + return ConfluenceWikiTemplate._fetch_json_with_retry(config, url) + + @staticmethod + @retry(max_retries=3, initial_delay=1.0, max_delay=10.0, backoff_factor=2.0) + def _fetch_json_with_retry(config: ConfluenceWikiConfig, url: str) -> dict: req = request.Request(url=url, method="GET") req.add_header("Accept", "application/json") diff --git a/src/ragflow_orchestrator/templates/github_template.py b/src/ragflow_orchestrator/templates/github_template.py index 3c2b95c..dea59cc 100644 --- a/src/ragflow_orchestrator/templates/github_template.py +++ b/src/ragflow_orchestrator/templates/github_template.py @@ -5,6 +5,7 @@ from urllib import parse, request from ragflow_orchestrator.graph import SqlGraphStore +from ragflow_orchestrator.retry import retry from ragflow_orchestrator.templates.base import BaseIngestionTemplate from ragflow_orchestrator.templates.models import GitHubConfig, IngestionError, TemplateRunReport @@ -128,6 +129,12 @@ def _get_readme(self, config: GitHubConfig, full_name: str) -> str: @staticmethod def _request_json(config: GitHubConfig, url: str) -> object: + """Fetch JSON from URL with retry logic.""" + return GitHubTemplate._fetch_json_with_retry(config, url) + + @staticmethod + @retry(max_retries=3, initial_delay=1.0, max_delay=10.0, backoff_factor=2.0) + def _fetch_json_with_retry(config: GitHubConfig, url: str) -> object: req = request.Request(url=url, method="GET") req.add_header("Accept", "application/vnd.github+json") req.add_header("X-GitHub-Api-Version", "2022-11-28") diff --git a/src/ragflow_orchestrator/templates/gitlab_template.py b/src/ragflow_orchestrator/templates/gitlab_template.py index 8041515..712157e 100644 --- a/src/ragflow_orchestrator/templates/gitlab_template.py +++ b/src/ragflow_orchestrator/templates/gitlab_template.py @@ -6,6 +6,7 @@ from urllib import parse, request from ragflow_orchestrator.graph import SqlGraphStore +from ragflow_orchestrator.retry import retry from ragflow_orchestrator.templates.base import BaseIngestionTemplate from ragflow_orchestrator.templates.models import GitLabConfig, IngestionError, TemplateRunReport @@ -184,6 +185,12 @@ def _get_readme(self, config: GitLabConfig, project_id: str) -> str: @staticmethod def _request_json(config: GitLabConfig, url: str) -> object: + """Fetch JSON from URL with retry logic.""" + return GitLabTemplate._fetch_json_with_retry(config, url) + + @staticmethod + @retry(max_retries=3, initial_delay=1.0, max_delay=10.0, backoff_factor=2.0) + def _fetch_json_with_retry(config: GitLabConfig, url: str) -> object: req = request.Request(url=url, method="GET") req.add_header("Accept", "application/json") if config.auth_mode == "bearer" and config.token: diff --git a/src/ragflow_orchestrator/templates/jira.py b/src/ragflow_orchestrator/templates/jira.py index abacb36..eaf70dd 100644 --- a/src/ragflow_orchestrator/templates/jira.py +++ b/src/ragflow_orchestrator/templates/jira.py @@ -4,6 +4,7 @@ import json from urllib import parse, request +from ragflow_orchestrator.retry import retry from ragflow_orchestrator.templates.base import BaseIngestionTemplate from ragflow_orchestrator.templates.models import IngestionError, JiraConfig, TemplateRunReport @@ -105,6 +106,12 @@ def _flatten_jira_doc(node: object) -> str: @staticmethod def _request_json(config: JiraConfig, url: str) -> dict: + """Fetch JSON from URL with retry logic.""" + return JiraTemplate._fetch_json_with_retry(config, url) + + @staticmethod + @retry(max_retries=3, initial_delay=1.0, max_delay=10.0, backoff_factor=2.0) + def _fetch_json_with_retry(config: JiraConfig, url: str) -> dict: req = request.Request(url=url, method="GET") req.add_header("Accept", "application/json") diff --git a/src/ragflow_orchestrator/templates/pypi.py b/src/ragflow_orchestrator/templates/pypi.py index d697cd4..5fe02cf 100644 --- a/src/ragflow_orchestrator/templates/pypi.py +++ b/src/ragflow_orchestrator/templates/pypi.py @@ -4,6 +4,7 @@ import re from urllib import parse, request +from ragflow_orchestrator.retry import retry from ragflow_orchestrator.templates.base import BaseIngestionTemplate from ragflow_orchestrator.templates.models import IngestionError, PyPIConfig, TemplateRunReport from ragflow_orchestrator.templates.utils import html_to_text @@ -67,6 +68,12 @@ def run(self, config: PyPIConfig) -> TemplateRunReport: @staticmethod def _fetch_package_payload(package: str) -> dict: + """Fetch package JSON from PyPI with retry logic.""" + return PyPITemplate._fetch_package_payload_with_retry(package) + + @staticmethod + @retry(max_retries=3, initial_delay=1.0, max_delay=10.0, backoff_factor=2.0) + def _fetch_package_payload_with_retry(package: str) -> dict: safe_package = parse.quote(package, safe="") url = f"https://pypi.org/pypi/{safe_package}/json" req = request.Request(url=url, method="GET") diff --git a/src/ragflow_orchestrator/templates/web_crawl.py b/src/ragflow_orchestrator/templates/web_crawl.py index b952a78..019a0d8 100644 --- a/src/ragflow_orchestrator/templates/web_crawl.py +++ b/src/ragflow_orchestrator/templates/web_crawl.py @@ -4,6 +4,7 @@ from urllib import request from ragflow_orchestrator.document_pipeline import detect_document_type +from ragflow_orchestrator.retry import retry from ragflow_orchestrator.templates.base import BaseIngestionTemplate from ragflow_orchestrator.templates.models import IngestionError, TemplateRunReport, WebCrawlConfig from ragflow_orchestrator.templates.utils import extract_links, extract_text_from_html, is_same_domain @@ -71,6 +72,12 @@ def run(self, config: WebCrawlConfig) -> TemplateRunReport: @staticmethod def _fetch_html(url: str, timeout: int = 15) -> tuple[str, str] | str: + """Fetch HTML from URL with retry logic.""" + return WebCrawlTemplate._fetch_html_with_retry(url, timeout) + + @staticmethod + @retry(max_retries=3, initial_delay=1.0, max_delay=10.0, backoff_factor=2.0) + def _fetch_html_with_retry(url: str, timeout: int = 15) -> tuple[str, str] | str: req = request.Request(url=url, headers={"User-Agent": "rag-orchestrator/0.1"}) with request.urlopen(req, timeout=timeout) as response: content_type = response.headers.get("Content-Type", "") diff --git a/tests/test_retry.py b/tests/test_retry.py new file mode 100644 index 0000000..e4a0aa8 --- /dev/null +++ b/tests/test_retry.py @@ -0,0 +1,174 @@ +"""Tests for retry utilities with exponential backoff.""" + +from __future__ import annotations + +import time +from unittest.mock import MagicMock, patch + +import pytest + +from ragflow_orchestrator.retry import retry, retry_or_log + + +class TestRetry: + """Test retry decorator with exponential backoff.""" + + def test_retry_succeeds_on_first_attempt(self) -> None: + @retry(max_retries=2, initial_delay=0.01) + def successful_func() -> str: + return "success" + + result = successful_func() + assert result == "success" + + def test_retry_succeeds_after_retries(self) -> None: + call_count = 0 + + @retry(max_retries=2, initial_delay=0.01) + def eventually_succeeds() -> str: + nonlocal call_count + call_count += 1 + if call_count < 2: + raise ValueError("temporary failure") + return "success" + + result = eventually_succeeds() + assert result == "success" + assert call_count == 2 + + def test_retry_exhausts_attempts(self) -> None: + call_count = 0 + + @retry(max_retries=2, initial_delay=0.01) + def always_fails() -> str: + nonlocal call_count + call_count += 1 + raise ValueError("always fails") + + with pytest.raises(ValueError, match="always fails"): + always_fails() + assert call_count == 3 # initial + 2 retries + + def test_retry_respects_max_retries(self) -> None: + call_count = 0 + + @retry(max_retries=1, initial_delay=0.01) + def fails_twice() -> str: + nonlocal call_count + call_count += 1 + raise ValueError("fails") + + with pytest.raises(ValueError): + fails_twice() + assert call_count == 2 # initial + 1 retry + + def test_retry_with_jitter(self) -> None: + @retry(max_retries=1, initial_delay=0.01, jitter=True) + def successful_func() -> str: + return "success" + + result = successful_func() + assert result == "success" + + def test_retry_with_max_delay(self) -> None: + """Test that delay is capped at max_delay.""" + call_count = 0 + delays: list[float] = [] + + @retry(max_retries=3, initial_delay=10.0, max_delay=0.05, backoff_factor=2.0, jitter=False) + def fails_multiple_times() -> str: + nonlocal call_count + call_count += 1 + if call_count < 4: + raise ValueError("fails") + return "success" + + start = time.time() + result = fails_multiple_times() + elapsed = time.time() - start + + # Should have 4 calls (1 initial + 3 retries) + # With max_delay=0.05, total sleep should be <= 0.05 * 3 + assert call_count == 4 + assert elapsed < 0.2 # generous upper bound due to system variability + + def test_retry_with_backoff_factor(self) -> None: + """Test that backoff increases delay exponentially.""" + call_count = 0 + + @retry(max_retries=2, initial_delay=0.01, max_delay=1.0, backoff_factor=2.0, jitter=False) + def fails_twice() -> str: + nonlocal call_count + call_count += 1 + if call_count <= 2: + raise ValueError("fails") + return "success" + + start = time.time() + result = fails_twice() + elapsed = time.time() - start + + # Delays: 0.01, 0.02 = 0.03 total + assert result == "success" + assert elapsed >= 0.03 + + +class TestRetryOrLog: + """Test retry_or_log decorator with graceful fallback.""" + + def test_retry_or_log_succeeds_on_first_attempt(self) -> None: + @retry_or_log(max_retries=2, initial_delay=0.01) + def successful_func() -> str: + return "success" + + result = successful_func() + assert result == "success" + + def test_retry_or_log_succeeds_after_retries(self) -> None: + call_count = 0 + + @retry_or_log(max_retries=2, initial_delay=0.01) + def eventually_succeeds() -> str: + nonlocal call_count + call_count += 1 + if call_count < 2: + raise ValueError("temporary failure") + return "success" + + result = eventually_succeeds() + assert result == "success" + assert call_count == 2 + + def test_retry_or_log_returns_none_on_exhaustion(self) -> None: + call_count = 0 + + @retry_or_log(max_retries=2, initial_delay=0.01) + def always_fails() -> str: + nonlocal call_count + call_count += 1 + raise ValueError("always fails") + + result = always_fails() + assert result is None + assert call_count == 3 # initial + 2 retries + + def test_retry_or_log_with_on_failure_callback(self) -> None: + callback_called = False + callback_exc: Exception | None = None + callback_msg: str | None = None + + def on_failure(exc: Exception, msg: str) -> None: + nonlocal callback_called, callback_exc, callback_msg + callback_called = True + callback_exc = exc + callback_msg = msg + + @retry_or_log(max_retries=1, initial_delay=0.01, on_failure=on_failure) + def always_fails() -> str: + raise ValueError("test failure") + + result = always_fails() + assert result is None + assert callback_called + assert isinstance(callback_exc, ValueError) + assert "test failure" in callback_msg From c01144e3a0d42be317b46c9b76b97392791ab705 Mon Sep 17 00:00:00 2001 From: Somebody Else Date: Thu, 18 Jun 2026 22:33:37 +0300 Subject: [PATCH 2/3] fix ruff suggestions --- src/ragflow_orchestrator/retry.py | 2 -- tests/test_retry.py | 4 +--- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/src/ragflow_orchestrator/retry.py b/src/ragflow_orchestrator/retry.py index fcda045..a6e7931 100644 --- a/src/ragflow_orchestrator/retry.py +++ b/src/ragflow_orchestrator/retry.py @@ -102,13 +102,11 @@ def decorator(func: Callable[P, T]) -> Callable[P, T | None]: @wraps(func) def wrapper(*args: P.args, **kwargs: P.kwargs) -> T | None: delay = initial_delay - last_exception: Exception | None = None for attempt in range(max_retries + 1): try: return func(*args, **kwargs) except Exception as exc: - last_exception = exc if attempt < max_retries: if jitter: jittered_delay = delay * (0.5 + random.random()) diff --git a/tests/test_retry.py b/tests/test_retry.py index e4a0aa8..133252a 100644 --- a/tests/test_retry.py +++ b/tests/test_retry.py @@ -3,7 +3,6 @@ from __future__ import annotations import time -from unittest.mock import MagicMock, patch import pytest @@ -73,7 +72,6 @@ def successful_func() -> str: def test_retry_with_max_delay(self) -> None: """Test that delay is capped at max_delay.""" call_count = 0 - delays: list[float] = [] @retry(max_retries=3, initial_delay=10.0, max_delay=0.05, backoff_factor=2.0, jitter=False) def fails_multiple_times() -> str: @@ -84,7 +82,7 @@ def fails_multiple_times() -> str: return "success" start = time.time() - result = fails_multiple_times() + fails_multiple_times() elapsed = time.time() - start # Should have 4 calls (1 initial + 3 retries) From 2d7573ef0cc1ebc966a696f671f09a7c09612d29 Mon Sep 17 00:00:00 2001 From: Somebody Else Date: Thu, 18 Jun 2026 22:36:35 +0300 Subject: [PATCH 3/3] mypy fixes --- src/ragflow_orchestrator/retry.py | 2 ++ tests/test_retry.py | 1 + 2 files changed, 3 insertions(+) diff --git a/src/ragflow_orchestrator/retry.py b/src/ragflow_orchestrator/retry.py index a6e7931..b459af7 100644 --- a/src/ragflow_orchestrator/retry.py +++ b/src/ragflow_orchestrator/retry.py @@ -63,6 +63,8 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: else: logger.error(f"All {max_retries + 1} attempts failed for {func.__name__}: {exc}") + if last_exception is None: + raise RuntimeError(f"Retry loop for {func.__name__} exited without result or exception") raise last_exception return wrapper diff --git a/tests/test_retry.py b/tests/test_retry.py index 133252a..370981b 100644 --- a/tests/test_retry.py +++ b/tests/test_retry.py @@ -169,4 +169,5 @@ def always_fails() -> str: assert result is None assert callback_called assert isinstance(callback_exc, ValueError) + assert callback_msg is not None assert "test failure" in callback_msg