api_utils.py

Overview

api_utils.py is a utility module within the InfiniFlow API backend that provides a comprehensive collection of helper functions, decorators, and routines to facilitate common API-related tasks. These tasks include request validation, response construction, authentication enforcement, error handling, model verification, JSON serialization, timeout management, and interaction with external or internal services.

The file centralizes various cross-cutting concerns that simplify the building and maintenance of RESTful APIs by providing reusable components that are robust, secure, and consistent with the system-wide standards and configurations.

Detailed Description of Functions and Decorators

1. serialize_for_json(obj)

Purpose:
Recursively converts complex Python objects into JSON-serializable structures. Handles objects with __dict__ attributes, classes, lists, tuples, dictionaries, and primitive types. Fallbacks to string representation when serialization fails.

Parameters:

• obj (Any): The object to serialize.

Returns:

• JSON-serializable representation of the input object.

Usage Example:

from api_utils import serialize_for_json

class User:
    def __init__(self, name):
        self.name = name

user = User("Alice")
json_data = serialize_for_json(user)
# Output: {'name': 'Alice'}

2. request(**kwargs)

Purpose:
Wraps the requests library to send HTTP requests with support for custom headers, client authentication (HMAC signature), timeout, and streaming.

Parameters:

• Accepts arbitrary keyword arguments supported by requests.Request (e.g., method, url, headers, data, json, params).

Returns:

• requests.Response object from the executed HTTP request.

Implementation Details:

• Automatically converts headers keys to uppercase with hyphens.

• If client authentication is enabled in settings, generates a signature header using HMAC-SHA1.

• Supports streaming and timeout as optional parameters.

3. get_exponential_backoff_interval(retries, full_jitter=False)

Purpose:
Computes wait time for retry attempts using exponential backoff with optional full jitter, following AWS best practices.

Parameters:

• retries (int): Number of retry attempts already performed.

• full_jitter (bool): If True, returns a random value between 0 and max wait time to avoid thundering herd problem.

Returns:

• int: Wait time in seconds.

4. get_data_error_result(code=settings.RetCode.DATA_ERROR, message="Sorry! Data missing!")

Purpose:
Returns a Flask JSON response indicating data-related errors.

Parameters:

• code (int): Error code, default from settings.

• message (str): Error message.

Returns:

• flask.Response: JSON error response.

5. server_error_response(e)

Purpose:
Generates an appropriate JSON error response based on the type and content of an exception.

Parameters:

• e (Exception): The exception instance.

Returns:

• JSON response with error code and message.

6. error_response(response_code, message=None)

Purpose:
Produces a simple JSON error response with HTTP status code and a message.

Parameters:

• response_code (int): HTTP status code.

• message (str, optional): Error message. Defaults to HTTP status text.

Returns:

• flask.Response: JSON error response.

7. validate_request(*args, **kwargs)

Purpose:
Decorator to validate presence and correctness of request parameters (in JSON or form data).

Parameters:

• *args (str): Required parameter names.

• **kwargs (dict): Parameters with expected values or list/tuple of allowed values.

Usage Example:

@validate_request("user_id", status=["active", "pending"])
def some_api_method():
    ...

Returns:

• Decorated function that returns argument error JSON if validation fails.

8. not_allowed_parameters(*params)

Purpose:
Decorator to reject requests containing specified forbidden parameters.

Parameters:

• *params (str): Parameter names that are disallowed.

9. is_localhost(ip)

Purpose:
Checks if the given IP address corresponds to localhost.

Parameters:

• ip (str): IP address.

Returns:

• bool: True if localhost, else False.

10. send_file_in_mem(data, filename)

Purpose:
Creates a Flask send_file response for sending data held in memory as a file attachment.

Parameters:

• data (str, bytes, or dict): Data to send.

• filename (str): Filename for the attachment.

Returns:

• flask.Response: File attachment response.

11. get_json_result(code=settings.RetCode.SUCCESS, message="success", data=None)

Purpose:
Standardized JSON response with code, message, and optional data.

12. apikey_required(func)

Purpose:
Decorator enforcing API key authentication via Authorization header.

Behavior:

• Extracts token from Authorization header.

• Queries APIToken DB model for validity.

• Injects tenant_id into decorated function's kwargs.

13. build_error_result(code=settings.RetCode.FORBIDDEN, message="success")

Purpose:
Creates a Flask JSON error response with status code.

14. construct_response(code=settings.RetCode.SUCCESS, message="success", data=None, auth=None)

Purpose:
Builds a Flask response with JSON payload and CORS headers. Optionally sets an Authorization header.

15. construct_result(code=settings.RetCode.DATA_ERROR, message="data is missing")

Purpose:
Returns a simple JSON response with code and message (no data).

16. construct_json_result(code=settings.RetCode.SUCCESS, message="success", data=None)

Purpose:
Returns JSON response including data if provided.

17. construct_error_response(e)

Purpose:
Constructs a JSON error response based on exception details.

18. token_required(func)

Purpose:
Decorator validating Authorization token, similar to apikey_required but with more detailed error messages.

19. get_result(code=settings.RetCode.SUCCESS, message="", data=None)

Purpose:
General JSON response helper, omitting message when code is success and data is present.

20. get_error_data_result(message="Sorry! Data missing!", code=settings.RetCode.DATA_ERROR)

Purpose:
Returns a data error JSON response.

21. get_error_argument_result(message="Invalid arguments")

Purpose:
Returns argument error JSON response.

22. get_error_permission_result(message="Permission error")

Purpose:
Returns permission error JSON response.

23. get_error_operating_result(message="Operating error")

Purpose:
Returns operation error JSON response.

24. generate_confirmation_token(tenant_id)

Purpose:
Generates a URL-safe confirmation token for a tenant using its UUID and itsdangerous.

25. get_parser_config(chunk_method, parser_config)

Purpose:
Returns parser configuration dictionary for a specified chunking method, merging default and user-provided configurations.

26. get_data_openai(...)

Purpose:
Constructs a dictionary mimicking OpenAI API chat completion response format.

27. check_duplicate_ids(ids, id_type="item")

Purpose:
Checks for duplicates in a list of IDs and returns unique IDs and error messages.

28. verify_embedding_availability(embd_id: str, tenant_id: str)

Purpose:
Checks if an embedding model is available and authorized for a tenant.

Returns:

• Tuple (bool, flask.Response | None) indicating validation result and optional error response.

29. deep_merge(default: dict, custom: dict)

Purpose:
Recursively merges two dictionaries, overriding defaults with custom keys.

30. remap_dictionary_keys(source_data: dict, key_aliases: dict = None)

Purpose:
Transforms dictionary keys according to a mapping schema.

31. get_mcp_tools(mcp_servers: list, timeout: float | int = 10)

Purpose:
Retrieves tool information from a list of MCP servers with timeout support.

32. timeout(seconds: float | int = None, attempts: int = 2, *, exception: Optional[TimeoutException] = None, on_timeout: Optional[OnTimeoutCallback] = None)

Purpose:
Decorator to enforce timeout and retry policy on synchronous and asynchronous functions.

33. is_strong_enough(chat_model, embedding_model)

Purpose:
Asynchronously performs stress tests on chat and embedding models using trio concurrency and the timeout decorator.

Important Implementation Details & Algorithms

• HMAC Authentication for Requests:
  The request function signs outgoing requests using HMAC-SHA1 with a nonce and timestamp to ensure secure communication.

• Exponential Backoff with Jitter:
  get_exponential_backoff_interval implements AWS-style retry backoff to reduce retry storm issues.

• Recursive Deep Merge:
  deep_merge uses a stack-based approach for efficient non-recursive deep dictionary merging, prioritizing custom values.

• Timeout Decorator:
  Supports both synchronous (thread-based) and asynchronous (trio) timeouts with configurable retries and custom exception handling.

• Tenant-Specific Model Authorization:
  verify_embedding_availability enforces multi-tenant model usage policies by querying tenant models and system-wide built-in models.

Interaction with Other System Components

• Database Models:
  Uses APIToken for API key/token validation.

• LLM Services:
  Integrates with LLMService and TenantLLMService to verify availability and authorization of language/embedding models.

• Settings and Constants:
  Uses settings and api.constants for configuration values and return codes.

• Flask Framework:
  Utilizes Flask for request context, response building, and JSON serialization.

• RAG Component:
  Interacts with RAG tools via MCPToolCallSession to fetch tool information.

• Custom JSON Encoder:
  Ensures complex objects are correctly serialized in JSON payloads.

Visual Diagram - Flowchart of Main Utility Functions and Their Relationships

flowchart TD
    A[serialize_for_json] -->|Used by| B[server_error_response]
    C[request] -->|Used for| D[API Calls with Authentication]
    E[get_exponential_backoff_interval] --> F[Retry Logic]
    G[validate_request] --> H[API Endpoint Decorators]
    I[apikey_required] --> H
    J[token_required] --> H
    K[timeout] --> L[Function Execution Control]
    M[verify_embedding_availability] --> N[LLM & Tenant Services]
    O[deep_merge] --> P[get_parser_config]
    Q[remap_dictionary_keys] --> R[Data Transformation]
    S[get_mcp_tools] --> T[MCP Sessions & Tool Retrieval]
    U[is_strong_enough] --> V[Model Stress Testing]

    H --> API[API Endpoints]
    D --> API
    F --> API
    L --> API
    N --> API
    P --> API
    R --> API
    T --> API
    V --> API

Summary

The api_utils.py module is a foundational utilities file for the InfiniFlow API backend. It encapsulates robust mechanisms for request validation, error handling, authentication, JSON serialization, configuration management, and timeout control. These utilities promote code reuse and consistency across the API's diverse endpoints and services, supporting secure, reliable, and maintainable API operation.

End of Documentation for api_utils.py