minio_conn.py

Overview

minio_conn.py provides a singleton class RAGFlowMinio that encapsulates connection management and operations with a MinIO object storage server. This file is designed to abstract and simplify common MinIO interactions such as uploading, downloading, deleting objects, checking object existence, generating presigned URLs, and bucket management.

The class manages connection lifecycle internally, including automatic reconnection attempts upon failures, making it robust for integration into larger systems needing reliable object storage operations.

Detailed Component Documentation

Class: RAGFlowMinio

A singleton class that manages a persistent connection to a MinIO server and exposes methods for common S3-compatible storage operations.

• Singleton Pattern: Ensured by the @singleton decorator from rag.utils, preventing multiple instances and thus maintaining a single shared connection throughout the application lifecycle.

Properties

• conn (Minio or None): The active Minio client connection instance. Initialized at object creation and reconnected as needed.

Methods

__init__(self)

Initializes the RAGFlowMinio instance, setting up the connection to MinIO server.

• Calls open() to establish a connection.

Usage example:

minio_client = RAGFlowMinio()

__open__(self)

Establishes a connection to the MinIO server using credentials and host info from settings.MINIO.

• If an existing connection exists, it attempts to close it first.

• Creates a new Minio client with secure=False (non-SSL).

• Logs exceptions if connection fails.

Internal method - not intended to be called externally.

__close__(self)

Closes the current connection by deleting the conn attribute and setting it to None.

Internal method.

health(self) -> object

Checks the MinIO server health by:

• Creating a test bucket txtxtxtxt1 if it does not exist.

• Uploading a small test object (b"_t@@@1") into the bucket.

• Returns the result of put_object() call (an object containing info about the upload).

Usage example:

result = minio_client.health()
print(result)  # Information about the test upload

put(self, bucket: str, fnm: str, binary: bytes) -> object or None

Uploads a binary object to the specified bucket and filename.

• Retries up to 3 times in case of failure.

• Creates the bucket if it does not exist.

• Uses BytesIO to wrap the binary data.

• Returns the result of put_object on success or None on failure.

Parameters:

• bucket: Target bucket name.

• fnm: Filename (object name) under which to store the data.

• binary: Data bytes to upload.

Usage example:

data = b"Hello, MinIO!"
minio_client.put("mybucket", "hello.txt", data)

rm(self, bucket: str, fnm: str) -> None

Removes (deletes) the specified object from a bucket.

• Logs exceptions on failure but does not raise.

Parameters:

• bucket: Bucket name.

• fnm: Object name to remove.

Usage example:

minio_client.rm("mybucket", "hello.txt")

get(self, bucket: str, filename: str) -> bytes or None

Retrieves the contents of the specified object.

• Attempts one try to fetch.

• On failure, logs the exception, reopens connection, and waits 1 second.

• Returns the raw bytes of the object on success, or None if failure.

Parameters:

• bucket: Bucket name.

• filename: Object name to retrieve.

Usage example:

content = minio_client.get("mybucket", "hello.txt")
if content:
    print(content.decode())

obj_exist(self, bucket: str, filename: str) -> bool

Checks if an object exists in a bucket.

• Returns False immediately if bucket does not exist.

• Uses stat_object() to check object metadata existence.

• Catches S3Error with specific codes indicating non-existence.

• Returns True if object exists, otherwise False.

Parameters:

• bucket: Bucket name.

• filename: Object name to check.

Usage example:

exists = minio_client.obj_exist("mybucket", "hello.txt")
print("Exists:", exists)

get_presigned_url(self, bucket: str, fnm: str, expires: int) -> str or None

Generates a presigned URL for downloading an object.

• Retries up to 10 times on failure.

• Returns the presigned URL string or None if failed.

Parameters:

• bucket: Bucket name.

• fnm: Object name.

• expires: Expiry time in seconds for the URL validity.

Usage example:

url = minio_client.get_presigned_url("mybucket", "hello.txt", expires=3600)
print(url)

remove_bucket(self, bucket: str) -> None

Deletes all objects inside the bucket and then removes the bucket itself.

• Checks if bucket exists.

• Lists all objects recursively.

• Removes each object.

• Removes the bucket.

• Logs exceptions on failure.

Parameters:

• bucket: Bucket name to remove.

Usage example:

minio_client.remove_bucket("mybucket")

Important Implementation Details

• Singleton Pattern: Ensures only one instance manages the MinIO connection, reducing resource overhead and maintaining a consistent connection state.

• Automatic Reconnection: Methods that interact with MinIO attempt to recover from failure by reopening the connection and retrying after a brief pause.

• Retries: put and get_presigned_url implement retry loops, improving robustness against transient network or service issues.

• Non-secure Connection: The MinIO client is configured with secure=False, meaning it uses HTTP rather than HTTPS; this may be configurable or intentional for local/private deployments.

• Exception Handling: The class logs exceptions instead of raising them, favoring fault tolerance and allowing the calling system to continue functioning.

Interaction with Other System Components

• rag.settings: Provides MinIO connection configuration (host, user, password).

• rag.utils.singleton: Decorator that enforces the singleton pattern on RAGFlowMinio.

• minio library: Core dependency for S3-compatible object storage communication.

• This file acts as a dedicated storage layer interface, likely invoked by higher-level modules needing to store or retrieve data blobs remotely.

Visual Diagram

classDiagram
    class RAGFlowMinio {
        -conn: Minio or None
        +__init__()
        +__open__()
        +__close__()
        +health() object
        +put(bucket: str, fnm: str, binary: bytes) object or None
        +rm(bucket: str, fnm: str) None
        +get(bucket: str, filename: str) bytes or None
        +obj_exist(bucket: str, filename: str) bool
        +get_presigned_url(bucket: str, fnm: str, expires: int) str or None
        +remove_bucket(bucket: str) None
    }

Summary

minio_conn.py provides a robust, singleton-based wrapper around MinIO client operations, abstracting connection management, retries, and error handling. It offers all essential object storage operations with fault tolerance and simple interfaces, integrating tightly with project settings and utilities.

This design simplifies storage access for the rest of the application while ensuring reliability and consistent connection state management.