"""Definition of MinioAdapter class.""" from __future__ import annotations import os from io import BytesIO from typing import Self import structlog 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 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 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.""" # 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.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 = self._client is not None self.logger.debug(res) return res 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 self._client is None or self._bucket_name 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, 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 self._client is None or self._bucket_name 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 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 self._client is None or self._bucket_name 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 self._bucket_name 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