"""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" create_bucket_if_missing_env_var_name: str = "MINIO_CREATE_BUCKET_IF_MISSING" 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 client is not None and config is None: raise ValueError("config is required when client is provided") if config is None: 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.create_bucket_if_missing_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): if not self._config.create_bucket_if_missing: raise ConnectionError( f"Bucket '{bucket}' does not exist on Minio at {endpoint}" ) 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 response = None 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}'" ) return None raise finally: if response is not None: response.close() response.release_conn() 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