Files
python-repositories/python_repositories/config/minio_config.py
T
Brian Bjarke JensenandCursor 3aa91280ec
Code Quality Pipeline / code-quality (pull_request) Successful in 32s
PR Title Check / check-title (pull_request) Successful in 6s
Test Python Package / test (pull_request) Successful in 57s
Load MINIO_SECURE from env and replace adapter asserts with explicit errors.
Default TLS to enabled for production MinIO setups while keeping local and integration tests on plain HTTP via explicit secure=false configuration.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-06 20:45:37 +02:00

66 lines
1.8 KiB
Python

"""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
_TRUTHY = frozenset({"1", "true", "yes", "on"})
_FALSY = frozenset({"0", "false", "no", "off"})
def _env_bool(name: str, default: bool) -> bool:
raw = os.getenv(name)
if raw is None:
return default
normalized = raw.strip().lower()
if normalized in _TRUTHY:
return True
if normalized in _FALSY:
return False
raise ValueError(f"Invalid boolean value for {name}: {raw!r}")
@dataclass(frozen=True)
class MinioConfig:
"""Configuration for connecting to MinIO."""
endpoint: str
access_key: str
secret_key: str
bucket: str
secure: bool = True
@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",
*,
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),
)