Implementation:Infiniflow Ragflow Connection Utils

Knowledge Sources	Infiniflow_Ragflow
Domains	Infrastructure, Design_Patterns
Last Updated	2026-02-12 06:00 GMT

Overview

Concrete tool providing a timeout decorator and HTTP response construction utilities for the RAGFlow API layer.

Description

The connection_utils module (originally named decorator.py but re-exported) provides a timeout decorator that wraps both synchronous and asynchronous functions with configurable timeout and retry logic, plus construct_response and sync_construct_response functions for building standardized JSON API responses with CORS headers.

Usage

Import the timeout decorator when implementing service methods that call external APIs or perform long-running operations. Import construct_response when building HTTP endpoint handlers that need standardized response formatting.

Code Reference

Source Location

• Repository: Infiniflow_Ragflow
• File: common/connection_utils.py
• Lines: 1-141

Signature

def timeout(seconds: int, attempts: int = 2, exception=None, on_timeout=None):
    """Decorator for sync/async functions with timeout and retry logic."""

async def construct_response(code=RetCode.SUCCESS, message="success", data=None, auth=None):
    """Build JSON response with CORS headers (async)."""

def sync_construct_response(code=RetCode.SUCCESS, message="success", data=None, auth=None):
    """Build JSON response with CORS headers (sync)."""

Import

from common.connection_utils import timeout, construct_response

I/O Contract

Inputs

Name	Type	Required	Description
seconds	int	Yes	Timeout duration in seconds
attempts	int	No	Number of retry attempts (default: 2)
code	RetCode	No	Response status code (default: SUCCESS)
message	str	No	Response message string
data	any	No	Response payload data

Outputs

Name	Type	Description
timeout() returns	decorator	Decorated function with timeout behavior
construct_response() returns	Response	Flask JSON response with CORS headers

Usage Examples

from common.connection_utils import timeout, construct_response

@timeout(seconds=30, attempts=3)
def fetch_remote_data(url):
    """This function will timeout after 30 seconds, retrying up to 3 times."""
    import requests
    return requests.get(url).json()

# Build API response
response = sync_construct_response(data={"result": "ok"})

Related Pages

• Environment:Infiniflow_Ragflow_Python_Runtime