renamed

Author: Andrei Neagu

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

@@ -17,7 +17,7 @@
 from models_library.basic_types import SHA256Str
 from models_library.bytes_iters import BytesIter, DataSize
 from pydantic import AnyUrl, ByteSize, TypeAdapter
-from servicelib.bytes_iters import DEFAULT_READ_CHUNK_SIZE, StreamData
+from servicelib.bytes_iters import DEFAULT_READ_CHUNK_SIZE, BytesStreamer
 from servicelib.logging_utils import log_catch, log_context
 from servicelib.s3_utils import FileLikeBytesIterReader
 from servicelib.utils import limited_gather
@@ -473,13 +473,13 @@ async def copy_objects_recursively(
             limit=_MAX_CONCURRENT_COPY,
         )
 
-    async def get_object_stream_data(
+    async def get_bytes_streamer_from_object(
         self,
         bucket_name: S3BucketName,
         object_key: S3ObjectKey,
         *,
         chunk_size: int = DEFAULT_READ_CHUNK_SIZE,
-    ) -> StreamData:
+    ) -> BytesStreamer:
         """stream read an object from S3 chunk by chunk"""
 
         # NOTE `download_fileobj` cannot be used to implement this because
@@ -512,7 +512,7 @@ async def _() -> BytesIter:
 
                 position += chunk_size
 
-        return StreamData(data_size, _)
+        return BytesStreamer(data_size, _)
 
     @s3_exception_handler(_logger)
     async def upload_object_from_bytes_iter(  # TODO: this needs to be based on file interface -> use protocol to expose read

packages/aws-library/tests/test_s3_client.py

@@ -1405,14 +1405,14 @@ async def test_read_object_file_stream(
     tmp_file_name: Path,
 ):
     async with aiofiles.open(tmp_file_name, "wb") as f:
-        stream_data = await simcore_s3_api.get_object_stream_data(
+        bytes_streamer = await simcore_s3_api.get_bytes_streamer_from_object(
             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_bytes_iter(AsyncMock()):
+        assert isinstance(bytes_streamer.data_size, DataSize)
+        async for chunk in bytes_streamer.with_progress_bytes_iter(AsyncMock()):
             await f.write(chunk)
 
-    assert stream_data.data_size == tmp_file_name.stat().st_size
+    assert bytes_streamer.data_size == tmp_file_name.stat().st_size
 
     await assert_same_file_content(with_uploaded_file_on_s3.local_path, tmp_file_name)
 
@@ -1424,13 +1424,13 @@ async def test_upload_object_from_file_stream(
     with_s3_bucket: S3BucketName,
 ):
     object_key = "read_from_s3_write_to_s3"
-    stream_data = await simcore_s3_api.get_object_stream_data(
+    bytes_streamer = await simcore_s3_api.get_bytes_streamer_from_object(
         with_s3_bucket, with_uploaded_file_on_s3.s3_key
     )
-    assert isinstance(stream_data.data_size, DataSize)
+    assert isinstance(bytes_streamer.data_size, DataSize)
 
     await simcore_s3_api.upload_object_from_bytes_iter(
-        with_s3_bucket, object_key, stream_data.with_progress_bytes_iter(AsyncMock())
+        with_s3_bucket, object_key, bytes_streamer.with_progress_bytes_iter(AsyncMock())
     )
 
     await simcore_s3_api.delete_object(bucket=with_s3_bucket, object_key=object_key)
@@ -1555,32 +1555,32 @@ async def test_workflow_compress_s3_objects_and_local_files_in_a_single_archive_
 
     # 1. assemble and upload zip archive
 
-    archive_file_entries: ArchiveEntries = []
+    archive_entries: ArchiveEntries = []
 
     local_files = get_files_info_from_path(path_local_files_for_archive)
     for file_name, file_path in local_files.items():
-        archive_file_entries.append(
+        archive_entries.append(
             (
                 file_name,
-                DiskStreamReader(file_path).get_stream_data(),
+                DiskStreamReader(file_path).get_bytes_streamer(),
             )
         )
 
     s3_files = get_files_info_from_path(path_s3_files_for_archive)
 
     for s3_object_key in s3_files:
-        archive_file_entries.append(
+        archive_entries.append(
             (
                 s3_object_key,
-                await simcore_s3_api.get_object_stream_data(
+                await simcore_s3_api.get_bytes_streamer_from_object(
                     with_s3_bucket, s3_object_key
                 ),
             )
         )
 
     # shuffle order of files in archive.
     # some will be read from S3 and some from the disk
-    random.shuffle(archive_file_entries)
+    random.shuffle(archive_entries)
 
     started = time.time()
 
@@ -1593,7 +1593,7 @@ async def test_workflow_compress_s3_objects_and_local_files_in_a_single_archive_
             with_s3_bucket,
             archive_s3_object_key,
             get_zip_bytes_iter(
-                archive_file_entries,
+                archive_entries,
                 progress_bar=progress_bar,
                 chunk_size=MULTIPART_COPY_THRESHOLD,
             ),

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

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

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

@@ -4,15 +4,15 @@
 from models_library.bytes_iters import BytesIter, DataSize
 
 from ._constants import DEFAULT_READ_CHUNK_SIZE
-from ._models import StreamData
+from ._models import BytesStreamer
 
 
 class DiskStreamReader:
     def __init__(self, file_path: Path, *, chunk_size=DEFAULT_READ_CHUNK_SIZE):
         self.file_path = file_path
         self.chunk_size = chunk_size
 
-    def get_stream_data(self) -> StreamData:
+    def get_bytes_streamer(self) -> BytesStreamer:
         async def _() -> BytesIter:
             async with aiofiles.open(self.file_path, "rb") as f:
                 while True:
@@ -22,4 +22,4 @@ async def _() -> BytesIter:
 
                     yield chunk
 
-        return StreamData(DataSize(self.file_path.stat().st_size), _)
+        return BytesStreamer(DataSize(self.file_path.stat().st_size), _)

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

@@ -6,7 +6,7 @@
 
 
 @dataclass(frozen=True)
-class StreamData:
+class BytesStreamer:
     data_size: DataSize
     bytes_iter_callable: BytesIterCallable
 

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

@@ -7,28 +7,28 @@
 from stream_zip import ZIP_32, AsyncMemberFile, async_stream_zip
 
 from ..progress_bar import ProgressBarData
-from ._models import StreamData
+from ._models import BytesStreamer
 
 FileNameInArchive: TypeAlias = str
-ArchiveFileEntry: TypeAlias = tuple[FileNameInArchive, StreamData]
+ArchiveFileEntry: TypeAlias = tuple[FileNameInArchive, BytesStreamer]
 ArchiveEntries: TypeAlias = list[ArchiveFileEntry]
 
 
 async def _member_files_iter(
-    file_streams: ArchiveEntries, progress_bar: ProgressBarData
+    archive_entries: ArchiveEntries, progress_bar: ProgressBarData
 ) -> AsyncIterable[AsyncMemberFile]:
-    for file_name, stream_info in file_streams:
+    for file_name, byte_streamer in archive_entries:
         yield (
             file_name,
             datetime.now(UTC),
             S_IFREG | 0o600,
             ZIP_32,
-            stream_info.with_progress_bytes_iter(progress_bar=progress_bar),
+            byte_streamer.with_progress_bytes_iter(progress_bar=progress_bar),
         )
 
 
 async def get_zip_bytes_iter(
-    archive_files: ArchiveEntries,
+    archive_entries: ArchiveEntries,
     *,
     progress_bar: ProgressBarData | None = None,
     chunk_size: int,
@@ -39,18 +39,16 @@ async def get_zip_bytes_iter(
         progress_bar = ProgressBarData(num_steps=1, description="zip archive stream")
 
     total_stream_lenth = DataSize(
-        sum(stream_info.data_size for _, stream_info in archive_files)
-    )
-    description = (
-        f"files: count={len(archive_files)}, size={total_stream_lenth.human_readable()}"
+        sum(bytes_streamer.data_size for _, bytes_streamer in archive_entries)
     )
+    description = f"files: count={len(archive_entries)}, size={total_stream_lenth.human_readable()}"
 
     async with progress_bar.sub_progress(
         steps=total_stream_lenth, description=description, progress_unit="Byte"
     ) as sub_progress:
         # NOTE: do not disable compression or the streams will be
         # loaded fully in memory before yielding their content
         async for chunk in async_stream_zip(
-            _member_files_iter(archive_files, sub_progress), chunk_size=chunk_size
+            _member_files_iter(archive_entries, sub_progress), chunk_size=chunk_size
         ):
             yield chunk

packages/service-library/tests/test_bytes_iters.py

@@ -94,7 +94,7 @@ def _progress_cb(*args, **kwargs) -> None:
 
 
 @pytest.mark.parametrize("use_file_like", [True, False])
-async def test_get_zip_data_stream(
+async def test_get_zip_bytes_iter(
     mocked_progress_bar_cb: Mock,
     prepare_content: None,
     local_files_dir: Path,
@@ -107,7 +107,9 @@ async def test_get_zip_data_stream(
     for file in (x for x in local_files_dir.rglob("*") if x.is_file()):
         archive_name = get_relative_to(local_files_dir, file)
 
-        archive_files.append((archive_name, DiskStreamReader(file).get_stream_data()))
+        archive_files.append(
+            (archive_name, DiskStreamReader(file).get_bytes_streamer())
+        )
 
     writer = DiskStreamWriter(local_archive_path)