renamed to bytes_iter

Author: Andrei Neagu

packages/aws-library/src/aws_library/s3/_client.py

@@ -15,11 +15,11 @@
 from botocore.client import Config
 from models_library.api_schemas_storage import ETag, S3BucketName, UploadedPart
 from models_library.basic_types import SHA256Str
-from models_library.data_streams import DataSize, DataStream
+from models_library.data_streams import BytesIter, DataSize
 from pydantic import AnyUrl, ByteSize, TypeAdapter
 from servicelib.data_streams import DEFAULT_READ_CHUNK_SIZE, StreamData
 from servicelib.logging_utils import log_catch, log_context
-from servicelib.s3_utils import FileLikeDataStreamReader
+from servicelib.s3_utils import FileLikeBytesIterReader
 from servicelib.utils import limited_gather
 from settings_library.s3 import S3Settings
 from types_aiobotocore_s3 import S3Client
@@ -492,7 +492,7 @@ async def get_object_stream_data(
         )
         data_size = DataSize(head_response["ContentLength"])
 
-        async def _() -> DataStream:
+        async def _() -> BytesIter:
             # Download the file in chunks
             position = 0
             while position < data_size:
@@ -515,14 +515,14 @@ async def _() -> DataStream:
         return StreamData(data_size, _)
 
     @s3_exception_handler(_logger)
-    async def upload_object_from_file_stream(
+    async def upload_object_from_bytes_iter(  # TODO: this needs to be based on file interface -> use protocol to expose read
         self,
         bucket_name: S3BucketName,
         object_key: S3ObjectKey,
-        file_stream: DataStream,
+        bytes_iter: BytesIter,
     ) -> None:
         """streams write an object in S3 from an AsyncIterable[bytes]"""
-        await self._client.upload_fileobj(FileLikeDataStreamReader(file_stream), bucket_name, object_key)  # type: ignore[arg-type]
+        await self._client.upload_fileobj(FileLikeBytesIterReader(bytes_iter), bucket_name, object_key)  # type: ignore[arg-type]
 
     @staticmethod
     def is_multipart(file_size: ByteSize) -> bool:

packages/aws-library/tests/test_s3_client.py

@@ -60,11 +60,7 @@
 )
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from servicelib.archiving_utils import unarchive_dir
-from servicelib.data_streams import (
-    ArchiveEntries,
-    DiskStreamReader,
-    get_zip_data_stream,
-)
+from servicelib.data_streams import ArchiveEntries, DiskStreamReader, get_zip_bytes_iter
 from servicelib.data_streams._models import DataSize
 from servicelib.file_utils import remove_directory
 from servicelib.progress_bar import ProgressBarData
@@ -1413,7 +1409,7 @@ async def test_read_object_file_stream(
             with_s3_bucket, with_uploaded_file_on_s3.s3_key, chunk_size=1024
         )
         assert isinstance(stream_data.data_size, DataSize)
-        async for chunk in stream_data.with_progress_data_stream(AsyncMock()):
+        async for chunk in stream_data.with_progress_bytes_iter(AsyncMock()):
             await f.write(chunk)
 
     assert stream_data.data_size == tmp_file_name.stat().st_size
@@ -1433,8 +1429,8 @@ async def test_upload_object_from_file_stream(
     )
     assert isinstance(stream_data.data_size, DataSize)
 
-    await simcore_s3_api.upload_object_from_file_stream(
-        with_s3_bucket, object_key, stream_data.with_progress_data_stream(AsyncMock())
+    await simcore_s3_api.upload_object_from_bytes_iter(
+        with_s3_bucket, object_key, stream_data.with_progress_bytes_iter(AsyncMock())
     )
 
     await simcore_s3_api.delete_object(bucket=with_s3_bucket, object_key=object_key)
@@ -1593,10 +1589,10 @@ async def test_workflow_compress_s3_objects_and_local_files_in_a_single_archive_
         progress_report_cb=mocked_progress_bar_cb,
         description="root_bar",
     ) as progress_bar:
-        await simcore_s3_api.upload_object_from_file_stream(
+        await simcore_s3_api.upload_object_from_bytes_iter(
             with_s3_bucket,
             archive_s3_object_key,
-            get_zip_data_stream(
+            get_zip_bytes_iter(
                 archive_file_entries,
                 progress_bar=progress_bar,
                 chunk_size=MULTIPART_COPY_THRESHOLD,

packages/models-library/src/models_library/data_streams.py

@@ -3,7 +3,7 @@
 
 from pydantic import ByteSize
 
-DataStream: TypeAlias = AsyncIterable[bytes]
+BytesIter: TypeAlias = AsyncIterable[bytes]
 
-DataStreamCallable: TypeAlias = Callable[[], DataStream]
+BytesIterCallable: TypeAlias = Callable[[], BytesIter]
 DataSize: TypeAlias = ByteSize

packages/service-library/src/servicelib/data_streams/__init__.py

@@ -2,14 +2,14 @@
 from ._input import DiskStreamReader
 from ._models import StreamData
 from ._output import DiskStreamWriter
-from ._stream_zip import ArchiveEntries, ArchiveFileEntry, get_zip_data_stream
+from ._stream_zip import ArchiveEntries, ArchiveFileEntry, get_zip_bytes_iter
 
 __all__: tuple[str, ...] = (
     "ArchiveEntries",
     "ArchiveFileEntry",
     "DEFAULT_READ_CHUNK_SIZE",
     "DiskStreamReader",
     "DiskStreamWriter",
-    "get_zip_data_stream",
+    "get_zip_bytes_iter",
     "StreamData",
 )

packages/service-library/src/servicelib/data_streams/_input.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 
 import aiofiles
-from models_library.data_streams import DataSize, DataStream
+from models_library.data_streams import BytesIter, DataSize
 
 from ._constants import DEFAULT_READ_CHUNK_SIZE
 from ._models import StreamData
@@ -13,7 +13,7 @@ def __init__(self, file_path: Path, *, chunk_size=DEFAULT_READ_CHUNK_SIZE):
         self.chunk_size = chunk_size
 
     def get_stream_data(self) -> StreamData:
-        async def _() -> DataStream:
+        async def _() -> BytesIter:
             async with aiofiles.open(self.file_path, "rb") as f:
                 while True:
                     chunk = await f.read(self.chunk_size)

packages/service-library/src/servicelib/data_streams/_models.py

@@ -1,18 +1,18 @@
 from dataclasses import dataclass
 
-from models_library.data_streams import DataSize, DataStream, DataStreamCallable
+from models_library.data_streams import BytesIter, BytesIterCallable, DataSize
 
 from ..progress_bar import ProgressBarData
 
 
 @dataclass(frozen=True)
 class StreamData:
     data_size: DataSize
-    data_stream_callable: DataStreamCallable
+    bytes_iter_callable: BytesIterCallable
 
-    async def with_progress_data_stream(
+    async def with_progress_bytes_iter(
         self, progress_bar: ProgressBarData
-    ) -> DataStream:
-        async for chunk in self.data_stream_callable():
+    ) -> BytesIter:
+        async for chunk in self.bytes_iter_callable():
             await progress_bar.update(len(chunk))
             yield chunk

packages/service-library/src/servicelib/data_streams/_output.py

@@ -1,23 +1,23 @@
 from pathlib import Path
 
 import aiofiles
-from models_library.data_streams import DataStream
+from models_library.data_streams import BytesIter
 
-from ..s3_utils import FileLikeDataStreamReader
+from ..s3_utils import FileLikeBytesIterReader
 
 
 class DiskStreamWriter:
     def __init__(self, destination_path: Path):
         self.destination_path = destination_path
 
-    async def write_from_stream(self, stream: DataStream) -> None:
+    async def write_from_bytes_iter(self, stream: BytesIter) -> None:
         async with aiofiles.open(self.destination_path, "wb") as f:
             async for chunk in stream:
                 await f.write(chunk)
                 await f.flush()
 
     async def write_from_file_like(
-        self, file_like_reader: FileLikeDataStreamReader
+        self, file_like_reader: FileLikeBytesIterReader
     ) -> None:
         async with aiofiles.open(self.destination_path, "wb") as f:
             while True:

packages/service-library/src/servicelib/data_streams/_stream_zip.py

@@ -3,7 +3,7 @@
 from stat import S_IFREG
 from typing import TypeAlias
 
-from models_library.data_streams import DataSize, DataStream
+from models_library.data_streams import BytesIter, DataSize
 from stream_zip import ZIP_32, AsyncMemberFile, async_stream_zip
 
 from ..progress_bar import ProgressBarData
@@ -23,16 +23,16 @@ async def _member_files_iter(
             datetime.now(UTC),
             S_IFREG | 0o600,
             ZIP_32,
-            stream_info.with_progress_data_stream(progress_bar=progress_bar),
+            stream_info.with_progress_bytes_iter(progress_bar=progress_bar),
         )
 
 
-async def get_zip_data_stream(
+async def get_zip_bytes_iter(
     archive_files: ArchiveEntries,
     *,
     progress_bar: ProgressBarData | None = None,
     chunk_size: int,
-) -> DataStream:
+) -> BytesIter:
     # NOTE: this is CPU bound task, even though the loop is not blocked,
     # the CPU is still used for compressing the content.
     if progress_bar is None:

packages/service-library/src/servicelib/s3_utils.py

@@ -1,14 +1,14 @@
-from models_library.data_streams import DataStream
+from models_library.data_streams import BytesIter
 
 
-class FileLikeDataStreamReader:
-    def __init__(self, data_stream: DataStream):
-        self._data_stream = data_stream
+class FileLikeBytesIterReader:
+    def __init__(self, bytes_iter: BytesIter):
+        self._bytes_iter = bytes_iter
         self._buffer = bytearray()
         self._async_iterator = self._get_iterator()
 
     async def _get_iterator(self):
-        async for chunk in self._data_stream:
+        async for chunk in self._bytes_iter:
             yield chunk
 
     async def read(self, size: int) -> bytes:

packages/service-library/tests/test_data_streams.py

@@ -19,11 +19,11 @@
     ArchiveEntries,
     DiskStreamReader,
     DiskStreamWriter,
-    get_zip_data_stream,
+    get_zip_bytes_iter,
 )
 from servicelib.file_utils import remove_directory
 from servicelib.progress_bar import ProgressBarData
-from servicelib.s3_utils import FileLikeDataStreamReader
+from servicelib.s3_utils import FileLikeBytesIterReader
 
 
 def _ensure_dir(path: Path) -> Path:
@@ -116,14 +116,14 @@ async def test_get_zip_data_stream(
         progress_report_cb=mocked_progress_bar_cb,
         description="root_bar",
     ) as root:
-        file_stream = get_zip_data_stream(
+        bytes_iter = get_zip_bytes_iter(
             archive_files, progress_bar=root, chunk_size=1024
         )
 
         if use_file_like:
-            await writer.write_from_file_like(FileLikeDataStreamReader(file_stream))
+            await writer.write_from_file_like(FileLikeBytesIterReader(bytes_iter))
         else:
-            await writer.write_from_stream(file_stream)
+            await writer.write_from_bytes_iter(bytes_iter)
 
     # 2. extract archive using exiting tools
     await unarchive_dir(local_archive_path, local_unpacked_archive)