Brian Bjarke JensenandCursor
5149299c1b
Convert is_connected to a method that verifies backend liveness via TTL-cached ping (Redis) or bucket_exists (MinIO), with cache invalidation on connect/disconnect. Co-authored-by: Cursor <[email protected]>
264 lines
9.5 KiB
Python
264 lines
9.5 KiB
Python
"""Definition of MinioAdapter class."""
|
|
|
|
from __future__ import annotations
|
|
import os
|
|
import structlog
|
|
import time
|
|
from io import BytesIO
|
|
from typing import Self
|
|
from python_utils import check_env
|
|
|
|
from python_repositories.interfaces import (
|
|
ConnectionAwareInterface,
|
|
ContextAwareInterface,
|
|
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,
|
|
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
|
|
health_check_ttl_seconds: float = 1.0
|
|
|
|
def __init__(self) -> None:
|
|
# 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.Minio | None = None
|
|
self._bucket_name: str | None = None
|
|
self._health_check_at: float | None = None
|
|
self._health_check_ok: bool = False
|
|
|
|
def __enter__(self) -> Self:
|
|
"""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."""
|
|
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()
|
|
# 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.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
|
|
self._invalidate_health_cache()
|
|
|
|
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
|
|
self._invalidate_health_cache()
|
|
|
|
def _invalidate_health_cache(self) -> None:
|
|
self._health_check_at = None
|
|
self._health_check_ok = False
|
|
|
|
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 is_connected(self) -> bool:
|
|
"""Check if connected to the Minio server."""
|
|
if self._client is None or self._bucket_name is None:
|
|
return False
|
|
now = time.monotonic()
|
|
if self._health_check_at is not None:
|
|
seconds_since_last_health_check = now - self._health_check_at
|
|
cache_is_fresh = (
|
|
seconds_since_last_health_check < self.health_check_ttl_seconds
|
|
)
|
|
if cache_is_fresh:
|
|
return self._health_check_ok
|
|
result = self._probe_connection()
|
|
self._health_check_at = now
|
|
self._health_check_ok = result
|
|
self.logger.debug("Connection status", connected=result)
|
|
return result
|
|
|
|
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
|
|
if not self.is_connected():
|
|
raise ConnectionError("Not connected to Minio")
|
|
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
|
|
if not self.is_connected():
|
|
raise ConnectionError("Not connected to Minio")
|
|
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
|
|
if not self.is_connected():
|
|
raise ConnectionError("Not connected to Minio")
|
|
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
|
|
if not self.is_connected():
|
|
raise ConnectionError("Not connected to Minio")
|
|
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
|