Brian Bjarke JensenandCursor
3aa91280ec
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 <[email protected]>
220 lines
8.2 KiB
Python
220 lines
8.2 KiB
Python
"""Definition of MinioAdapter class."""
|
|
|
|
from __future__ import annotations
|
|
from io import BytesIO
|
|
|
|
from python_repositories.adapters.connection_aware_adapter import (
|
|
ConnectionAwareAdapter,
|
|
)
|
|
from python_repositories.config import MinioConfig
|
|
from python_repositories.interfaces import ObjectRepositoryInterface
|
|
|
|
try:
|
|
import minio
|
|
except ImportError as exc:
|
|
raise ImportError(
|
|
"MinIO support requires the minio extra. "
|
|
"Install with: pip install python-repositories[minio]"
|
|
) from exc
|
|
|
|
|
|
class MinioAdapter(ObjectRepositoryInterface, ConnectionAwareAdapter):
|
|
"""Minio adapter exposing basic CRUD functionality."""
|
|
|
|
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"
|
|
chunk_size: int = 5 * 2**20 # 5 MiB
|
|
connection_name: str = "Minio"
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
config: MinioConfig | None = None,
|
|
client: minio.Minio | None = None,
|
|
) -> None:
|
|
super().__init__()
|
|
if config is None:
|
|
if client is not None:
|
|
raise ValueError("config is required when client is provided")
|
|
config = MinioConfig.from_env(
|
|
self.endpoint_env_var_name,
|
|
self.access_key_env_var_name,
|
|
self.secret_key_env_var_name,
|
|
self.bucket_env_var_name,
|
|
self.secure_env_var_name,
|
|
)
|
|
self._config = config
|
|
self._client_injected = client is not None
|
|
self._client: minio.Minio | None = client
|
|
self._bucket_name: str | None = config.bucket if client is not None else None
|
|
if self._client_injected:
|
|
self._invalidate_health_cache()
|
|
|
|
def _is_client_ready(self) -> bool:
|
|
return self._client is not None and self._bucket_name is not None
|
|
|
|
def connect(self) -> None:
|
|
"""Connect to the Minio server."""
|
|
if self._client_injected:
|
|
if self._client is not None:
|
|
try:
|
|
_ = self._client.list_buckets()
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
raise ConnectionError(
|
|
f"Could not connect to Minio at {self._config.endpoint}"
|
|
) from exc
|
|
self._invalidate_health_cache()
|
|
return
|
|
if self._client is not None and self.is_connected():
|
|
self.logger.info("Already connected to Minio")
|
|
return
|
|
if self._client is not None:
|
|
self.disconnect()
|
|
endpoint = self._config.endpoint
|
|
access_key = self._config.access_key
|
|
secret_key = self._config.secret_key
|
|
bucket = self._config.bucket
|
|
client = minio.Minio(
|
|
endpoint=endpoint,
|
|
access_key=access_key,
|
|
secret_key=secret_key,
|
|
secure=self._config.secure,
|
|
)
|
|
try:
|
|
_ = client.list_buckets()
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
raise ConnectionError(f"Could not connect to Minio at {endpoint}") from exc
|
|
if not client.bucket_exists(bucket):
|
|
self.logger.info(f"Creating bucket '{bucket}'")
|
|
client.make_bucket(bucket)
|
|
self._client = client
|
|
self._bucket_name = bucket
|
|
self._invalidate_health_cache()
|
|
|
|
def disconnect(self) -> None:
|
|
"""Disconnect from the Minio server."""
|
|
# N.B. Minio client does not have a close method, but we include this for symmetry
|
|
self._client = None
|
|
self._bucket_name = None
|
|
self._invalidate_health_cache()
|
|
|
|
def _probe_connection(self) -> bool:
|
|
assert self._client is not None and self._bucket_name is not None
|
|
try:
|
|
return bool(self._client.bucket_exists(self._bucket_name))
|
|
except Exception: # pylint: disable=broad-except
|
|
return False
|
|
|
|
def put(
|
|
self,
|
|
object_name: str,
|
|
data: BytesIO,
|
|
content_type: str = "application/octet-stream",
|
|
) -> None:
|
|
"""Put an object into the Minio bucket."""
|
|
# Check input
|
|
if not isinstance(object_name, str) or len(object_name) == 0:
|
|
raise ValueError("object_name must be a non-empty string")
|
|
if not isinstance(data, BytesIO) or data.getbuffer().nbytes == 0:
|
|
raise ValueError("data must be a non-empty BytesIO object")
|
|
if not isinstance(content_type, str) or len(content_type) == 0:
|
|
raise ValueError("content_type must be a non-empty string")
|
|
# Check connection
|
|
self._require_connected()
|
|
assert self._client is not None and self._bucket_name is not None
|
|
# Prepare buffer for reading
|
|
num_bytes = data.getbuffer().nbytes
|
|
data.seek(0)
|
|
# Send data to bucket
|
|
# N.B. bucket name is set when connecting
|
|
self._client.put_object(
|
|
bucket_name=self._bucket_name,
|
|
object_name=object_name,
|
|
data=data,
|
|
length=num_bytes,
|
|
part_size=self.chunk_size,
|
|
content_type=content_type,
|
|
)
|
|
self.logger.debug(
|
|
f"Put object '{object_name}' into bucket '{self._bucket_name}'"
|
|
)
|
|
|
|
def get(self, object_name: str) -> BytesIO | None:
|
|
"""Get an object from the Minio bucket."""
|
|
# Check input
|
|
if not isinstance(object_name, str) or len(object_name) == 0:
|
|
raise ValueError("object_name must be a non-empty string")
|
|
# Check connection
|
|
self._require_connected()
|
|
assert self._client is not None and self._bucket_name is not None
|
|
# Get data from bucket
|
|
# N.B. bucket name is set when connecting
|
|
try:
|
|
response = self._client.get_object(
|
|
bucket_name=self._bucket_name,
|
|
object_name=object_name,
|
|
)
|
|
# Get buffered data
|
|
buffer = BytesIO()
|
|
while chunk := response.read(self.chunk_size):
|
|
buffer.write(chunk)
|
|
buffer.seek(0)
|
|
self.logger.debug(
|
|
f"Got object '{object_name}' from bucket '{self._bucket_name}'"
|
|
)
|
|
return buffer
|
|
except minio.S3Error as exc:
|
|
if exc.code == "NoSuchKey":
|
|
self.logger.warning(
|
|
f"Object '{object_name}' not found in bucket '{self._bucket_name}'"
|
|
)
|
|
else:
|
|
self.logger.error(repr(exc))
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
self.logger.error(repr(exc))
|
|
return None
|
|
|
|
def delete(self, object_name: str) -> None:
|
|
"""Delete an object from the Minio bucket."""
|
|
# Check input
|
|
if not isinstance(object_name, str) or len(object_name) == 0:
|
|
raise ValueError("object_name must be a non-empty string")
|
|
# Check connection
|
|
self._require_connected()
|
|
assert self._client is not None and self._bucket_name is not None
|
|
# Delete object from bucket
|
|
# N.B. bucket name is set when connecting
|
|
self._client.remove_object(
|
|
bucket_name=self._bucket_name,
|
|
object_name=object_name,
|
|
)
|
|
self.logger.debug(
|
|
f"Deleted object '{object_name}' from bucket '{self._bucket_name}'"
|
|
)
|
|
|
|
def list_objects(self, prefix: str = "") -> list[str]:
|
|
"""List objects in the Minio bucket with an optional prefix."""
|
|
# Check input
|
|
if not isinstance(prefix, str):
|
|
raise ValueError("prefix must be a string")
|
|
# Check connection
|
|
# N.B. bucket name is set when connecting
|
|
self._require_connected()
|
|
assert self._client is not None and self._bucket_name is not None
|
|
# List objects in bucket
|
|
objects = self._client.list_objects(
|
|
bucket_name=self._bucket_name,
|
|
prefix=prefix,
|
|
recursive=True,
|
|
)
|
|
object_names = [
|
|
obj.object_name for obj in objects if obj.object_name is not None
|
|
]
|
|
self.logger.debug(
|
|
f"Listed {len(object_names)} object(s) in bucket '{self._bucket_name}' with prefix '{prefix}'"
|
|
)
|
|
return object_names
|