add retiries and degradation processing

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"

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}")
+
+            if last_exception is None:
+                raise RuntimeError(f"Retry loop for {func.__name__} exited without result or exception")
+            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
+
+            for attempt in range(max_retries + 1):
+                try:
+                    return func(*args, **kwargs)
+                except Exception as 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

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")

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")
 

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")

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:

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")
 

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")

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", "")