feat: zip stream now yields predictable chunk sizes

Author: dnlbauer

rocrate/memory_buffer.py

@@ -26,3 +26,6 @@ def read(self, size=-1):
             data = self._buffer[:size]
             self._buffer = self._buffer[size:]
         return data
+
+    def __len__(self):
+        return len(self._buffer)

rocrate/model/data_entity.py

@@ -30,7 +30,7 @@ class DataEntity(Entity):
     def write(self, base_path):
         pass
 
-    def stream(self) -> Generator[tuple[str, bytes], None, None]:
+    def stream(self, chunk_size=8192) -> Generator[tuple[str, bytes], None, None]:
         """ Stream the data from the source. Each chunk of the content is yielded as a tuple
         containing the name of the destination file relative to the crate and the chunk of data.
         The destination file name is required because a DataEntity can be a file or a

rocrate/model/dataset.py

@@ -82,15 +82,15 @@ def write(self, base_path):
         else:
             self._copy_folder(base_path)
 
-    def stream(self) -> Generator[tuple[str, bytes], None, None]:
+    def stream(self, chunk_size=8192) -> Generator[tuple[str, bytes], None, None]:
         if self.source is None:
             return
         elif is_url(str(self.source)):
-            yield from self._stream_folder_from_url()
+            yield from self._stream_folder_from_url(chunk_size)
         else:
-            yield from self._stream_folder_from_path()
+            yield from self._stream_folder_from_path(chunk_size)
 
-    def _stream_folder_from_path(self) -> Generator[tuple[str, bytes], None, None]:
+    def _stream_folder_from_path(self, chunk_size=8192):
         if not Path(str(self.source)).exists():
             raise FileNotFoundError(
                 errno.ENOENT, os.strerror(errno.ENOENT), str(self.source)
@@ -102,10 +102,10 @@ def _stream_folder_from_path(self) -> Generator[tuple[str, bytes], None, None]:
                     source = root / name
                     dest = source.relative_to(Path(self.source).parent)
                     with open(source, 'rb') as f:
-                        for chunk in f:
+                        while chunk := f.read(chunk_size):
                             yield str(dest), chunk
 
-    def _stream_folder_from_url(self) -> Generator[tuple[str, bytes], None, None]:
+    def _stream_folder_from_url(self, chunk_size=8192):
         if not self.fetch_remote:
             if self.validate_url:
                 with urlopen(self.source) as _:
@@ -121,7 +121,6 @@ def _stream_folder_from_url(self) -> Generator[tuple[str, bytes], None, None]:
                     rel_out_path = Path(self.id) / part
 
                     with urlopen(part_uri) as response:
-                        chunk_size = 8192
                         while chunk := response.read(chunk_size):
                             yield str(rel_out_path), chunk
                 except KeyError:

rocrate/model/file.py

@@ -93,7 +93,7 @@ def _stream_from_stream(self, stream):
         if self.record_size:
             self._jsonld['contentSize'] = str(size)
 
-    def _stream_from_url(self, url) -> Generator[tuple[str, bytes], None, None]:
+    def _stream_from_url(self, url, chunk_size=8192):
         if self.fetch_remote or self.validate_url:
             if self.validate_url:
                 if url.startswith("http"):
@@ -109,30 +109,29 @@ def _stream_from_url(self, url) -> Generator[tuple[str, bytes], None, None]:
                 size = 0
                 self._jsonld['contentUrl'] = str(url)
                 with urllib.request.urlopen(url) as response:
-                    chunk_size = 8192
                     while chunk := response.read(chunk_size):
                         yield self.id, chunk
                         size += len(chunk)
 
                 if self.record_size:
                     self._jsonld['contentSize'] = str(size)
 
-    def _stream_from_file(self, path):
+    def _stream_from_file(self, path, chunk_size=8192):
         size = 0
         with open(path, 'rb') as f:
-            for chunk in f:
+            while chunk := f.read(chunk_size):
                 yield self.id, chunk
                 size += len(chunk)
         if self.record_size:
             self._jsonld['contentSize'] = str(size)
 
-    def stream(self) -> Generator[tuple[str, bytes], None, None]:
+    def stream(self, chunk_size=8192) -> Generator[tuple[str, bytes], None, None]:
         if isinstance(self.source, (BytesIO, StringIO)):
             yield from self._stream_from_stream(self.source)
         elif is_url(str(self.source)):
-            yield from self._stream_from_url(self.source)
+            yield from self._stream_from_url(self.source, chunk_size)
         elif self.source is None:
             # Allows to record a File entity whose @id does not exist, see #73
             warnings.warn(f"No source for {self.id}")
         else:
-            yield from self._stream_from_file(self.source)
+            yield from self._stream_from_file(self.source, chunk_size)

rocrate/model/metadata.py

@@ -75,7 +75,7 @@ def generate(self):
             context = context[0]
         return {'@context': context, '@graph': graph}
 
-    def stream(self) -> Generator[tuple[str, bytes], None, None]:
+    def stream(self, chunk_size=8192) -> Generator[tuple[str, bytes], None, None]:
         content = self.generate()
         yield self.id, str.encode(json.dumps(content, indent=4, sort_keys=True), encoding='utf-8')
 

rocrate/model/preview.py

@@ -91,7 +91,7 @@ def is_object_list(a):
         out_html = src.render(crate=self.crate, context=context_entities, data=data_entities)
         return out_html
 
-    def stream(self) -> Generator[tuple[str, bytes], None, None]:
+    def stream(self, chunk_size=8192) -> Generator[tuple[str, bytes], None, None]:
         if self.source:
             yield from super().stream()
         else:

rocrate/rocrate.py

@@ -480,24 +480,26 @@ def write_zip(self, out_path):
             shutil.rmtree(tmp_dir)
         return archive
 
-    def stream_zip(self):
+    def stream_zip(self, chunk_size=8192):
         """ Create a stream of bytes representing the RO-Crate as a ZIP file. """
         with MemoryBuffer() as buffer:
             with zipfile.ZipFile(buffer, mode='w', compression=zipfile.ZIP_DEFLATED) as archive:
                 for writeable_entity in self.data_entities + self.default_entities:
                     current_file_path, current_out_file = None, None
-                    for path, chunk in writeable_entity.stream():
+                    for path, chunk in writeable_entity.stream(chunk_size=chunk_size):
                         if path != current_file_path:
                             if current_out_file:
                                 current_out_file.close()
                             current_file_path = path
                             current_out_file = archive.open(path, mode='w')
                         current_out_file.write(chunk)
-                        yield buffer.read()
+                        while len(buffer) >= chunk_size:
+                            yield buffer.read(chunk_size)
                     if current_out_file:
                         current_out_file.close()
 
-            yield buffer.read()
+            while chunk := buffer.read(chunk_size):
+                yield chunk
 
     def add_workflow(
             self, source=None, dest_path=None, fetch_remote=False, validate_url=False, properties=None,