python-repositories/python_repositories/config/minio_config.py

"""MinIO connection configuration."""

from __future__ import annotations

import os
from dataclasses import dataclass

from python_utils import check_env

from python_repositories.config.dotenv_loader import load_dotenv
from python_repositories.config.env_bool import env_bool


@dataclass(frozen=True)
class MinioConfig:
    """Configuration for connecting to MinIO."""

    endpoint: str
    access_key: str
    secret_key: str
    bucket: str
    secure: bool = True
    create_bucket_if_missing: bool = False

    @classmethod
    def from_env(
        cls,
        endpoint_env_var_name: str = "MINIO_ENDPOINT",
        access_key_env_var_name: str = "MINIO_ACCESS_KEY",
        secret_key_env_var_name: str = "MINIO_SECRET_KEY",
        bucket_env_var_name: str = "MINIO_BUCKET",
        secure_env_var_name: str = "MINIO_SECURE",
        create_bucket_if_missing_env_var_name: str = ("MINIO_CREATE_BUCKET_IF_MISSING"),
        *,
        use_dotenv: bool = True,
    ) -> MinioConfig:
        """Load configuration from environment variables."""
        if use_dotenv:
            load_dotenv()
        env_var_names = {
            endpoint_env_var_name,
            access_key_env_var_name,
            secret_key_env_var_name,
            bucket_env_var_name,
        }
        check_env(env_var_names)
        return cls(
            endpoint=str(os.getenv(endpoint_env_var_name)),
            access_key=str(os.getenv(access_key_env_var_name)),
            secret_key=str(os.getenv(secret_key_env_var_name)),
            bucket=str(os.getenv(bucket_env_var_name)),
            secure=env_bool(secure_env_var_name, default=True),
            create_bucket_if_missing=env_bool(
                create_bucket_if_missing_env_var_name,
                default=False,
            ),
        )