222 lines
7.9 KiB
Python
222 lines
7.9 KiB
Python
"""Definition of MinioAdapter class."""
|
|
|
|
from __future__ import annotations
|
|
import os
|
|
from io import BytesIO
|
|
from typing import TYPE_CHECKING
|
|
import structlog
|
|
from python_utils import check_env
|
|
|
|
from python_repositories.interfaces import (
|
|
ContextAwareInterface,
|
|
ConnectionAwareInterface,
|
|
)
|
|
|
|
# Import for mypy type checking only
|
|
if TYPE_CHECKING:
|
|
from minio import Minio, S3Error
|
|
|
|
|
|
class MinioAdapter(
|
|
ContextAwareInterface,
|
|
ConnectionAwareInterface,
|
|
):
|
|
"""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"
|
|
chunk_size: int = 5 * 2**20 # 5 MiB
|
|
|
|
def __init__(self) -> None:
|
|
# Import here to avoid hard dependency if MinioAdapter is not used
|
|
try:
|
|
from minio import Minio, S3Error
|
|
except ImportError as e:
|
|
raise RuntimeError("MinioAdapter dependencies missing") from e
|
|
# Setup logger
|
|
self.logger = structlog.get_logger(
|
|
self.__class__.__name__,
|
|
)
|
|
# Check environment variables
|
|
check_env(
|
|
{
|
|
self.endpoint_env_var_name,
|
|
self.access_key_env_var_name,
|
|
self.secret_key_env_var_name,
|
|
self.bucket_env_var_name,
|
|
},
|
|
)
|
|
# Prepare internal variables
|
|
self._client: Minio | None = None
|
|
self._bucket_name: str | None = None
|
|
|
|
def __enter__(self) -> MinioAdapter:
|
|
"""Enter the context."""
|
|
self.connect()
|
|
return self
|
|
|
|
def __exit__(
|
|
self, exc_type: type | None, exc_val: object | None, exc_tb: object | None
|
|
) -> None:
|
|
"""Exit the context."""
|
|
ctx_info = {"exc_type": exc_type, "exc_val": exc_val, "exc_tb": exc_tb}
|
|
if any(
|
|
(
|
|
exc_type is not None,
|
|
exc_val is not None,
|
|
exc_tb is not None,
|
|
),
|
|
):
|
|
self.logger.error("Error while exiting context", **ctx_info)
|
|
self.disconnect()
|
|
|
|
def connect(self) -> None:
|
|
"""Connect to the Minio server."""
|
|
# Stop if already connected
|
|
if self.is_connected:
|
|
self.logger.info("Already connected to Minio")
|
|
return
|
|
# Prepare arguments
|
|
endpoint = str(os.getenv(self.endpoint_env_var_name))
|
|
access_key = str(os.getenv(self.access_key_env_var_name))
|
|
secret_key = str(os.getenv(self.secret_key_env_var_name))
|
|
bucket = str(os.getenv(self.bucket_env_var_name))
|
|
# Connect client
|
|
client = Minio(
|
|
endpoint=endpoint,
|
|
access_key=access_key,
|
|
secret_key=secret_key,
|
|
secure=False,
|
|
)
|
|
# Test the connection by listing buckets (will raise if connection fails)
|
|
try:
|
|
_ = client.list_buckets()
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
raise ConnectionError(f"Could not connect to Minio at {endpoint}") from exc
|
|
# Ensure bucket exists
|
|
if not client.bucket_exists(bucket):
|
|
self.logger.info(f"Creating bucket '{bucket}'")
|
|
client.make_bucket(bucket)
|
|
# Persist information
|
|
self._client = client
|
|
self._bucket_name = bucket
|
|
|
|
def disconnect(self) -> None:
|
|
"""Disconnect from the Minio server."""
|
|
# Close connection
|
|
# N.B. Minio client does not have a close method, but we include this for symmetry with other adapters
|
|
# Reset client
|
|
self._client = None
|
|
self._bucket_name = None
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
"""Check if connected to Minio server."""
|
|
res = bool(isinstance(self._client, Minio))
|
|
self.logger.debug(res)
|
|
return res
|
|
|
|
def _put(self, object_name: str, data: BytesIO) -> 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")
|
|
# Check connection
|
|
if self._client is None or not self.is_connected:
|
|
raise ConnectionError("Not connected to Minio")
|
|
# 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,
|
|
)
|
|
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
|
|
if self._client is None or not self.is_connected:
|
|
raise ConnectionError("Not connected to Minio")
|
|
# 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 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
|
|
if self._client is None or not self.is_connected:
|
|
raise ConnectionError("Not connected to Minio")
|
|
# 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
|
|
if self._client is None or not self.is_connected:
|
|
raise ConnectionError("Not connected to Minio")
|
|
# 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
|