Merge pull request #122 from OnroerendErfgoed/develop

Develop

HISTORY.rst

@@ -1,3 +1,8 @@
+0.9.0 (22-04-2024)
+------------------
+
+- zipfile with list of keys (#119)
+
 0.8.0 (06-03-2024)
 ------------------
 

augeias/stores/PairTreeFileSystemStore.py

@@ -2,15 +2,14 @@
 This module provide a simple filesystem based store
 """
 import datetime
+import magic
 import os
 from io import BytesIO
-from zipfile import ZipFile
-
-import magic
 from pairtree import ObjectNotFoundException
 from pairtree import PairtreeStorageFactory
 from pairtree import PartNotFoundException
 from pairtree import id2path
+from zipfile import ZipFile
 
 from augeias.stores.StoreInterface import IStore
 from augeias.stores.error import NotFoundException
@@ -122,22 +121,29 @@ def delete_object(self, container_key, object_key):
 
     def get_container_data(self, container_key, translations=None):
         """
-        Find a container and return a zip file of its contents.
+        Find a container and return a zip file of the requested objects.
+        If translations exist, only the files within translations.keys() will be provided
 
         :param container_key: Key of the container which must be retrieved.
         :param translations: Dict of object IDs and file names to use for them.
         :return: a zip file containing all files of the container.
         """
         translations = translations or {}
-        container = self.store.get_object(container_key,
-                                          create_if_doesnt_exist=False)
+        container = self.store.get_object(
+            container_key,
+            create_if_doesnt_exist=False
+        )
+        object_list = translations.keys() if translations else container.list_parts()
         in_memory_file = BytesIO()
-        with ZipFile(in_memory_file, 'w') as zf:
-            for object_key in container.list_parts():
-                zf.writestr(
-                    translations.get(object_key, object_key),
-                    container.get_bytestream(object_key)
-                )
+        try:
+            with ZipFile(in_memory_file, 'w') as zf:
+                for object_key in object_list:
+                    zf.writestr(
+                        translations.get(object_key, object_key),
+                        container.get_bytestream(object_key)
+                    )
+        except PartNotFoundException:
+            raise NotFoundException
         in_memory_file.seek(0)
         return in_memory_file
 

augeias/views.py

@@ -194,7 +194,7 @@ def list_object_keys_for_container(self):
 
     @view_config(route_name='get_container_data', permission='view')
     def get_container_data(self):
-        """Get a container from the data store as zip."""
+        """Get a container or part of its objects from the data store as zip."""
         parameters = self.request.GET
         collection = _retrieve_collection(self.request)
         container_key = self.request.matchdict['container_key']
@@ -203,8 +203,11 @@ def get_container_data(self):
         )
         filename = str(container_key) + '.zip'
         disposition = (f'attachment; filename={filename}')
-        res = Response(content_type='application/zip', status=200,
-                       content_disposition=disposition)
+        res = Response(
+            content_type='application/zip',
+            status=200,
+            content_disposition=disposition
+        )
         res.body = zip_file.read()
         return res
 

setup.py

@@ -17,7 +17,7 @@
 ]
 
 setup(name='augeias',
-      version='0.8.0',
+      version='0.9.0',
       description='Augeias. Stores your files.',
       long_description=README + '\n\n' + HISTORY,
       classifiers=[

tests/test_functional.py

@@ -213,7 +213,8 @@ def test_download_container_zip(self):
             self.assertIn('400x600', filenames)
 
         res = self.testapp.get(
-            '/collections/TEST_COLLECTION/containers/TEST_CONTAINER_ID?200x300=file1.pdf',
+            '/collections/TEST_COLLECTION/containers'
+            '/TEST_CONTAINER_ID?200x300=file1.pdf&400x600=400x600',
             headers={'Accept': 'application/zip'}
         )
         self.assertEqual('200 OK', res.status)
@@ -482,3 +483,48 @@ def test_get_file_from_zip(self):
             expect_errors=True,
         )
         self.assertEqual(res.status_code, 400)
+
+    def test_get_objects_as_zip(self):
+        # create container and add object
+        zip_header = {"Accept": "Application/zip"}
+        cres = self.testapp.put(
+            "/collections/TEST_COLLECTION/containers/TEST_CONTAINER_ID")
+        self.assertEqual("200 OK", cres.status)
+        testdata = os.path.join(here, "../", "fixtures/kasteel.jpg")
+        with open(testdata, "rb") as f:
+            bdata = f.read()
+        self.testapp.put(
+            "/collections/TEST_COLLECTION/containers/TEST_CONTAINER_ID/001", bdata
+        )
+        testdata = os.path.join(here, "../", "fixtures/brug.jpg")
+        with open(testdata, "rb") as f:
+            bdata = f.read()
+        self.testapp.put(
+            "/collections/TEST_COLLECTION/containers/TEST_CONTAINER_ID/002", bdata
+        )
+        # get objects
+        res = self.testapp.get(
+            "/collections/TEST_COLLECTION/containers/TEST_CONTAINER_ID?001=custom_name.jpg",
+            headers=zip_header,
+        )
+        self.assertEqual("200 OK", res.status)
+        zip_file = io.BytesIO(res.body)
+        with zipfile.ZipFile(zip_file, "r") as zip_ref:
+            file_list = zip_ref.namelist()
+        self.assertCountEqual(["custom_name.jpg"], file_list)
+        # get zip objects
+        res = self.testapp.get(
+            "/collections/TEST_COLLECTION/containers/TEST_CONTAINER_ID?002=test.jpg",
+            headers=zip_header,
+        )
+        self.assertEqual("200 OK", res.status)
+        zip_file = io.BytesIO(res.body)
+        with zipfile.ZipFile(zip_file, "r") as zip_ref:
+            file_list = zip_ref.namelist()
+        self.assertCountEqual(["test.jpg"], file_list)
+        res = self.testapp.get(
+            "/collections/TEST_COLLECTION/containers/TEST_CONTAINER_ID?005=brug.jpg",
+            headers=zip_header,
+            expect_errors=True,
+        )
+        self.assertEqual(res.status_code, 404)