added minio adapter and tests

python_repositories/adapters/minio_adapter.py

@@ -0,0 +1,202 @@
+"""Definition of MinioAdapter class."""
+
+from __future__ import annotations
+import os
+from io import BytesIO
+
+import structlog
+from minio import Minio, S3Error
+
+from python_utils import check_env
+
+from python_repositories.interfaces import (
+    ContextAwareInterface,
+    ConnectionAwareInterface,
+)
+
+
+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:
+        # 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,  # type: ignore
+            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,  # type: ignore
+                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,  # type: ignore
+            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,  # type: ignore
+            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