Merge pull request #119 from negatic/add-minio-streaming

add minio object streaming

Author: HosseinNejatiJavaremi

archipy/adapters/minio/adapters.py

@@ -1,6 +1,6 @@
 import logging
 from collections.abc import Callable
-from typing import Any, NoReturn, TypeVar, override
+from typing import Any, BinaryIO, NoReturn, TypeVar, override
 
 import boto3
 from botocore.client import Config
@@ -811,3 +811,117 @@ def copy_object(
             self._handle_connection_exception(e, "copy_object")
         except Exception as e:
             self._handle_general_exception(e, "copy_object")
+
+    @override
+    def put_object_stream(
+        self,
+        bucket_name: str,
+        object_name: str,
+        data: bytes | BinaryIO,
+        length: int = -1,
+        content_type: str = "application/octet-stream",
+    ) -> None:
+        """Upload data from a bytes buffer or binary stream to a bucket.
+
+        Unlike put_object which requires a local file path, this method accepts
+        in-memory bytes or any binary stream, avoiding the need for a temporary file.
+
+        Args:
+            bucket_name: Destination bucket name.
+            object_name: Object name in the bucket.
+            data: Content to upload as raw bytes or a binary stream (BinaryIO).
+            length: Content length in bytes. If -1, computed automatically for bytes;
+                for streams, providing the exact length avoids buffering overhead.
+            content_type: MIME type of the content. Defaults to "application/octet-stream".
+
+        Raises:
+            InvalidArgumentError: If bucket_name, object_name, or data is invalid.
+            NotFoundError: If the bucket does not exist.
+            PermissionDeniedError: If permission to upload is denied.
+            ResourceExhaustedError: If storage limits are exceeded.
+            ServiceUnavailableError: If the S3 service is unavailable.
+            StorageError: If there's a storage-related error.
+        """
+        try:
+            if not bucket_name or not object_name:
+                raise InvalidArgumentError(
+                    argument_name=(
+                        "bucket_name or object_name"
+                        if not all([bucket_name, object_name])
+                        else "bucket_name"
+                        if not bucket_name
+                        else "object_name"
+                    ),
+                )
+
+            kwargs: dict[str, Any] = {
+                "Bucket": bucket_name,
+                "Key": object_name,
+                "Body": data,
+                "ContentType": content_type,
+            }
+
+            if isinstance(data, bytes):
+                kwargs["ContentLength"] = len(data) if length < 0 else length
+            elif length >= 0:
+                kwargs["ContentLength"] = length
+
+            self._client.put_object(**kwargs)
+            if hasattr(self.list_objects, "clear_cache"):
+                self.list_objects.clear_cache()
+        except InvalidArgumentError:
+            raise
+        except ClientError as e:
+            self._handle_client_exception(e, "put_object_stream")
+        except (ConnectionError, EndpointConnectionError) as e:
+            self._handle_connection_exception(e, "put_object_stream")
+        except Exception as e:
+            self._handle_general_exception(e, "put_object_stream")
+
+    @override
+    def get_object_stream(self, bucket_name: str, object_name: str) -> bytes:
+        """Download an object and return its content as bytes.
+
+        Unlike get_object which requires a local file path, this method returns
+        the object content directly in memory, avoiding a temporary file.
+
+        Args:
+            bucket_name: Source bucket name.
+            object_name: Object name in the bucket.
+
+        Returns:
+            bytes: The full content of the object.
+
+        Raises:
+            InvalidArgumentError: If any required parameter is empty.
+            NotFoundError: If the bucket or object does not exist.
+            PermissionDeniedError: If permission to download is denied.
+            ServiceUnavailableError: If the S3 service is unavailable.
+            StorageError: If there's a storage-related error.
+        """
+        try:
+            if not bucket_name or not object_name:
+                raise InvalidArgumentError(
+                    argument_name=(
+                        "bucket_name or object_name"
+                        if not all([bucket_name, object_name])
+                        else "bucket_name"
+                        if not bucket_name
+                        else "object_name"
+                    ),
+                )
+            response = self._client.get_object(Bucket=bucket_name, Key=object_name)
+        except InvalidArgumentError:
+            raise
+        except ClientError as e:
+            self._handle_client_exception(e, "get_object_stream")
+            raise
+        except (ConnectionError, EndpointConnectionError) as e:
+            self._handle_connection_exception(e, "get_object_stream")
+            raise
+        except Exception as e:
+            self._handle_general_exception(e, "get_object_stream")
+            raise
+        else:
+            content: bytes = response["Body"].read()
+            return content

archipy/adapters/minio/ports.py

@@ -1,7 +1,7 @@
 """MinIO port definitions for ArchiPy."""
 
 from abc import abstractmethod
-from typing import Any
+from typing import Any, BinaryIO
 
 # Define type aliases for better type hinting
 MinioObjectType = dict[str, Any]
@@ -110,3 +110,43 @@ def copy_object(
     ) -> None:
         """Copy an object within or between buckets."""
         raise NotImplementedError
+
+    @abstractmethod
+    def put_object_stream(
+        self,
+        bucket_name: str,
+        object_name: str,
+        data: bytes | BinaryIO,
+        length: int = -1,
+        content_type: str = "application/octet-stream",
+    ) -> None:
+        """Upload data from a bytes buffer or binary stream to a bucket.
+
+        Unlike put_object which requires a local file path, this method accepts
+        in-memory bytes or any binary stream, avoiding the need for a temporary file.
+
+        Args:
+            bucket_name: Destination bucket name.
+            object_name: Object name in the bucket.
+            data: Content to upload as raw bytes or a binary stream (BinaryIO).
+            length: Content length in bytes. If -1, computed automatically for bytes;
+                for streams, providing the exact length avoids buffering overhead.
+            content_type: MIME type of the content. Defaults to "application/octet-stream".
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_object_stream(self, bucket_name: str, object_name: str) -> bytes:
+        """Download an object and return its content as bytes.
+
+        Unlike get_object which requires a local file path, this method returns
+        the object content directly in memory, avoiding a temporary file.
+
+        Args:
+            bucket_name: Source bucket name.
+            object_name: Object name in the bucket.
+
+        Returns:
+            bytes: The full content of the object.
+        """
+        raise NotImplementedError

features/minio_adapter.feature

@@ -45,3 +45,15 @@ Feature: MinIO Operations Testing
     When I copy object "test.txt" from bucket "test-bucket" to "test-copy.txt" in the same bucket
     Then the object "test-copy.txt" should exist in bucket "test-bucket"
     And downloading "test-copy.txt" from "test-bucket" should return content "Hello World"
+
+  Scenario: Upload bytes buffer and retrieve via stream
+    Given a bucket named "test-bucket" exists
+    When I upload the bytes "Hello Streaming World" as "stream-bytes.txt" to bucket "test-bucket"
+    Then the object "stream-bytes.txt" should exist in bucket "test-bucket"
+    And the streaming download of "stream-bytes.txt" from "test-bucket" should return "Hello Streaming World"
+
+  Scenario: Upload binary stream and retrieve via stream
+    Given a bucket named "test-bucket" exists
+    When I upload a binary stream with content "Binary Stream Data" as "stream-bio.txt" to bucket "test-bucket"
+    Then the object "stream-bio.txt" should exist in bucket "test-bucket"
+    And the streaming download of "stream-bio.txt" from "test-bucket" should return "Binary Stream Data"

features/steps/minio_adapter_steps.py

@@ -1,4 +1,5 @@
 # features/steps/minio_steps.py
+import io
 import json
 import os
 import tempfile
@@ -218,3 +219,31 @@ def step_verify_policy(context, bucket_name):
     assert "s3:GetObject" in policy, "Policy doesn't contain read permission"
     assert "s3:PutObject" not in policy, "Policy contains write permission"
     context.logger.info(f"Verified read-only policy for '{bucket_name}'")
+
+
+@when('I upload the bytes "{content}" as "{object_name}" to bucket "{bucket_name}"')
+def step_upload_bytes_stream(context, content, object_name, bucket_name):
+    adapter = get_minio_adapter(context)
+    data = content.encode("utf-8")
+    adapter.put_object_stream(bucket_name, object_name, data, content_type="text/plain; charset=utf-8")
+    context.logger.info(f"Uploaded bytes stream '{object_name}' to '{bucket_name}'")
+
+
+@when('I upload a binary stream with content "{content}" as "{object_name}" to bucket "{bucket_name}"')
+def step_upload_binary_stream(context, content, object_name, bucket_name):
+    adapter = get_minio_adapter(context)
+    raw = content.encode("utf-8")
+    stream = io.BytesIO(raw)
+    adapter.put_object_stream(
+        bucket_name, object_name, stream, length=len(raw), content_type="text/plain; charset=utf-8"
+    )
+    context.logger.info(f"Uploaded binary stream '{object_name}' to '{bucket_name}'")
+
+
+@then('the streaming download of "{object_name}" from "{bucket_name}" should return "{expected_content}"')
+def step_stream_download_verify(context, object_name, bucket_name, expected_content):
+    adapter = get_minio_adapter(context)
+    content_bytes = adapter.get_object_stream(bucket_name, object_name)
+    content = content_bytes.decode("utf-8")
+    assert content == expected_content, f"Content mismatch: expected '{expected_content}', got '{content}'"
+    context.logger.info(f"Verified streaming download content of '{object_name}'")