Minio Provider

Inputs

Outputs

Source

python

# Input preparation
inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]
endpoint = inp["endpoint"]
access_key = inp["access_key"]
secret_key = inp["secret_key"]
secure = inp.get("secure", False)
default_bucket = inp.get("default_bucket")
region = inp.get("region")

# Compute
from minio import Minio
from minio.error import S3Error
from typing import Optional

class MinioProvider:
    """
    A class to interact with a MinIO object storage server.
    """

    def __init__(self, endpoint: str, access_key: str, secret_key: str, secure: bool = False, bucket_name: Optional[str] = None, region: Optional[str] = None):
        """
        Initialize the MinioProvider instance and create a default bucket if provided.

        :param endpoint: The endpoint of the MinIO server.
        :param access_key: The access key for authentication.
        :param secret_key: The secret key for authentication.
        :param secure: Whether to use HTTPS (True) or HTTP (False).
        :param bucket_name: The name of the bucket to create (optional).
        :param region: The default region for bucket creation (optional).
        """
        self.client = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=secure)
        
        # If bucket_name is provided, call create_default_bucket
        if bucket_name:
            self.create_default_bucket(bucket_name, region)

    def create_default_bucket(self, bucket_name: str, region: Optional[str] = None):
        """
        Create a default bucket if it does not already exist.

        :param bucket_name: The name of the bucket to create.
        :param region: The default region for bucket creation.
        """
        try:
            if not self.client.bucket_exists(bucket_name):
                self.client.make_bucket(bucket_name, location=region)
                print(f"Bucket '{bucket_name}' created successfully.")
            else:
                print(f"Bucket '{bucket_name}' already exists.")
            self.default_bucket = bucket_name
        except S3Error as exc:
            print(f"Error creating bucket: {exc}")

    def get_client(self) -> Minio:
        """
        Get the MinIO client instance.

        :return: The MinIO client.
        """
        return self.client

minio_provider = MinioProvider(
    endpoint=endpoint,
    access_key=access_key,
    secret_key=secret_key,
    secure=secure,
    bucket_name=default_bucket,
    region=region
)

# Output collection (runner reads __ivy_ctx__)
out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]
out["minio_provider"] = minio_provider

Tests

Requires: python:3.11

• basic-connection

  Create MinIO provider with valid credentials (should succeed).

  minioproviderconnectionbasic