corrected / cleaned upload and download logic with regard to operations and ids

pyinfra/file_descriptor_builder.py

@@ -42,9 +42,11 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder):
 
         file_pattern = self.operation2file_patterns[operation][end]
 
+        print(f"{end=}")
+
         file_descriptor = {
             **project(queue_item_body, ["dossierId", "fileId", "pages"]),
-            "pages": queue_item_body.get("pages", []),
+            "pages": [queue_item_body["id"]] if end == "output" else [],
             "extension": file_pattern["extension"],
             "subdir": file_pattern["subdir"],
         }
@@ -59,21 +61,25 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder):
             "dossierId", "fileId", "subdir", "pages", "extension"
         )(file_descriptor)
 
-        matcher = os.path.join(dossier_id, make_filename(file_id, subdir, self.__build_page_regex(pages) + extension))
+        matcher = os.path.join(
+            dossier_id, make_filename(file_id, subdir, self.__build_page_regex(pages, subdir) + extension)
+        )
 
         return matcher
 
     @staticmethod
-    def __build_page_regex(pages):
+    def __build_page_regex(pages, subdir):
 
         n_pages = len(pages)
+        print("pages:", pages)
 
         if n_pages > 1:
             page_re = "id:(" + "|".join(map(str, pages)) + ")."
         elif n_pages == 1:
             page_re = f"id:{pages[0]}."
-        else:  # no pages
-            page_re = r"id:\d+."
+        else:  # no pages specified -> either all pages or no pages, depending on whether a subdir is specified
+            print("subdir", subdir)
+            page_re = r"id:\d+." if subdir else ""
 
         return page_re
 

pyinfra/file_descriptor_manager.py

@@ -8,7 +8,8 @@ class FileDescriptorManager:
 
     TODO: This is supposed to be fully decoupled from the concrete file descriptor builder implementation, however some
         bad coupling is still left.
-     """
+    """
+
     def __init__(self, bucket_name, file_descriptor_builder: FileDescriptorBuilder):
         self.bucket_name = bucket_name
         self.operation_file_descriptor_builder = file_descriptor_builder
@@ -20,10 +21,7 @@ class FileDescriptorManager:
         return self.get_object_name(queue_item_body, end="output")
 
     def get_object_name(self, queue_item_body: dict, end):
-
         file_descriptor = self.build_file_descriptor(queue_item_body, end=end)
-        # FIXME: bad coupling
-        file_descriptor["pages"] = [queue_item_body.get("id", 0)]
         object_name = self.__build_matcher(file_descriptor)
 
         return object_name
@@ -48,7 +46,9 @@ class FileDescriptorManager:
         return self.get_object_descriptor(queue_item_body, end="input")
 
     def get_output_object_descriptor(self, storage_upload_info):
-        return self.get_object_descriptor(storage_upload_info, end="output")
+        r = self.get_object_descriptor(storage_upload_info, end="output")
+        print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>", r)
+        return r
 
     def get_object_descriptor(self, queue_item_body, end):
         # TODO: this is complected with the Storage class API

pyinfra/storage/adapters/azure.py

@@ -5,7 +5,7 @@ from azure.storage.blob import ContainerClient, BlobServiceClient
 
 from pyinfra.storage.adapters.adapter import StorageAdapter
 
-logger = logging.getLogger()
+logger = logging.getLogger(__name__)
 logging.getLogger("azure").setLevel(logging.WARNING)
 logging.getLogger("urllib3").setLevel(logging.WARNING)
 

pyinfra/storage/adapters/s3.py

@@ -8,7 +8,7 @@ from minio import Minio
 from pyinfra.exceptions import DataLoadingFailure
 from pyinfra.storage.adapters.adapter import StorageAdapter
 
-logger = logging.getLogger()
+logger = logging.getLogger(__name__)
 
 
 class S3StorageAdapter(StorageAdapter):
@@ -27,6 +27,7 @@ class S3StorageAdapter(StorageAdapter):
         logger.debug(f"Uploading '{object_name}'...")
         data = io.BytesIO(data)
         self.__client.put_object(bucket_name, object_name, data, length=data.getbuffer().nbytes)
+        # input("uploaded")
 
     def get_object(self, bucket_name, object_name):
         logger.debug(f"Downloading '{object_name}'...")

scripts/show_compressed_json.py

@@ -2,6 +2,10 @@ import argparse
 import gzip
 import json
 
+from funcy import lmap
+
+from pyinfra.server.packing import string_to_bytes
+
 
 def parse_args():
     parser = argparse.ArgumentParser()
@@ -9,14 +13,20 @@ def parse_args():
     return parser.parse_args()
 
 
+def interpret(parsed):
+    return {**parsed, "data": str(string_to_bytes(parsed["data"]))}
+
+
 def main(fp):
     with open(fp, "rb") as f:
         compressed_json_path = f.read()
 
     json_str = gzip.decompress(compressed_json_path)
-    json_dict = json.loads(json_str)
+    parsed = json.loads(json_str)
+    parsed = [parsed] if isinstance(parsed, dict) else parsed
+    parsed = lmap(interpret, parsed)
 
-    print(json.dumps(json_dict, indent=2))
+    print(json.dumps(parsed, indent=2))
 
 
 if __name__ == "__main__":

test/config.yaml

@@ -1,19 +1,56 @@
 service:
   response_formatter: identity
   operations:
-    multi_inp_op:
+#    multi_inp_op:
+#      input:
+#        subdir: op_inp_files
+#        extension: IN.gz
+#      output:
+#        subdir: op_outp_files
+#        extension: OUT.gz
+    upper:
       input:
-        subdir: op_inp_files
-        extension: IN.gz
+        subdir: ""
+        extension: up_in.gz
       output:
-        subdir: op_outp_files
-        extension: OUT.gz
+        subdir: ""
+        extension: up_out.gz
+    extract:
+      input:
+        subdir: ""
+        extension: extr_in.gz
+      output:
+        subdir: "extractions"
+        extension: gz
+    rotate:
+      input:
+        subdir: ""
+        extension: rot_in.gz
+      output:
+        subdir: ""
+        extension: rot_out.gz
+    classify:
+      input:
+        subdir: ""
+        extension: cls_in.gz
+      output:
+        subdir: ""
+        extension: cls_out.gz
+    stream_pages:
+      input:
+        subdir: ""
+        extension: pgs_in.gz
+      output:
+        subdir: "pages"
+        extension: pgs_out.gz
+
+
     default:
       input:
         subdir: ""
         extension: IN.gz
       output:
-        subdir: op_outp_files
+        subdir: ""
         extension: OUT.gz
 
 storage:
@@ -41,5 +78,5 @@ webserver:
 
 mock_analysis_endpoint: "http://127.0.0.1:5000"
 
-use_docker_fixture: 1
-logging: 0
+use_docker_fixture: 0
+logging: 1

test/conftest.py

@@ -89,8 +89,8 @@ def mock_make_load_data():
     return load_data
 
 
-# def pytest_make_parametrize_id(config, val, argname):
-#     return f"\n\t{argname}={val}\n"
+def pytest_make_parametrize_id(config, val, argname):
+    return f"\n\t{argname}={val}\n"
 
 
 @pytest.fixture

test/fixtures/consumer.py

@@ -26,8 +26,8 @@ def pair_data_with_queue_message(data: Iterable[bytes]):
     def inner():
         for i, d in enumerate(data):
             body = {
-                "dossierId": "folder",
-                "fileId": f"file{i}",
+                "dossierId": "dossier_id",
+                "fileId": f"file_id_{i}",
                 "targetFileExtension": "in.gz",
                 "responseFileExtension": "out.gz",
                 "pages": [0, 2, 3],

test/fixtures/input.py

@@ -6,6 +6,7 @@ import pytest
 from PIL import Image
 from funcy import lmap, compose, flatten, lflatten, omit, second, first, lzip, merge
 
+from pyinfra.server.debugging import inspect
 from pyinfra.server.dispatcher.dispatcher import Nothing
 from pyinfra.server.normalization import normalize_item
 from pyinfra.server.packing import pack, unpack
@@ -80,7 +81,7 @@ def targets(data_message_pairs, input_data_items, operation, metadata, server_si
     if operation is Nothing:
         return Nothing
 
-    op = compose(lift(star(pack)), normalize_item, operation)
+    op = compose(lift(star(pack)), normalize_item, inspect("A"), operation)
 
     try:
         response_data, response_metadata = zip(*map(unpack, flatten(starmap(op, zip(input_data_items, metadata)))))
@@ -147,15 +148,19 @@ def metadata(n_items, many_to_n):
 
 @pytest.fixture
 def queue_message_metadata(n_items, operation_name):
+    print()
+    print("n_items:", n_items)
     def metadata(i):
-        return merge(
+        r = merge(
             {
-                "dossierId": "folder",
-                "fileId": f"file{i}",
-                "pages": [0, 2, 3],
+                "dossierId": "dossier_id",
+                "fileId": f"file_id_{i}",
             },
             ({"operation": operation_name} if operation_name else {}),
+            ({"pages": [0, 2, 3]} if n_items > 1 else {}),
         )
+        print("metadata:", r)
+        return r
 
     return lmap(metadata, range(n_items))
 

test/fixtures/server.py

@@ -2,6 +2,7 @@ import io
 import logging
 import socket
 from collections import Counter
+from itertools import count
 from multiprocessing import Process
 from operator import itemgetter
 from typing import Generator
@@ -51,8 +52,10 @@ def server(server_stream_function, buffer_size, operation_name):
 
 
 @pytest.fixture
-def operation_name(many_to_n):
-    return "multi_inp_op" if many_to_n else ""
+def operation_name(core_operation):
+    operation_name = core_operation.__name__
+    print("OPERATION NAME", operation_name)
+    return operation_name
 
 
 @pytest.fixture
@@ -105,11 +108,18 @@ def core_operation(item_type, one_to_many, analysis_task):
             yield upper(string, metadata), metadata
 
     def upper(string: bytes, metadata):
-        return string.decode().upper().encode(), metadata
+        print("upper>>>>>>>>>>>", metadata)
+        print("???????????????????????????????", string)
+        r = string.decode().upper().encode(), metadata
+        print("!!!!!!!!!!!!!!!!!!!!!", r)
+        return r
 
     def extract(string: bytes, metadata):
-        for i, c in project(dict(enumerate(string.decode())), metadata["pages"]).items():
+        print("extract>>>>>>>>>>>", metadata)
+        print(string.decode())
+        for i, c in project(dict(enumerate(string.decode())), metadata.get("pages", range(len(string.decode())))).items():
             metadata["id"] = i
+            print(i)
             yield c.encode(), metadata
 
     def rotate(im: bytes, metadata):

test/integration_tests/serve_test.py

@@ -37,7 +37,7 @@ from test.config import CONFIG as TEST_CONFIG
 @pytest.mark.parametrize(
     "n_items",
     [
-        1,
+        # 1,
         3,
     ],
 )
@@ -78,11 +78,15 @@ from test.config import CONFIG as TEST_CONFIG
 @pytest.mark.parametrize(
     "many_to_n",
     [
-        True,
-        # False,
+        # True,
+        False,
     ],
 )
-def test_serving(server_process, bucket_name, components, targets, data_message_pairs, n_items, many_to_n):
+def test_serving(operation_name, server_process, bucket_name, components, targets, data_message_pairs, n_items, many_to_n):
+
+    print(f"{operation_name=}")
+    print()
+    print()
 
     storage, queue_manager, consumer, file_descriptor_manager = components
 
@@ -108,6 +112,12 @@ def test_serving(server_process, bucket_name, components, targets, data_message_
 
     # TODO: correctness of target should be validated as well, since production has become non-trivial
     assert sorted(outputs) == sorted(targets)
+    print(sorted(targets))
+    print(sorted(outputs))
+
+    # sleep(2)
+    # print("---")
+    # input()
 
 
 @pytest.fixture
@@ -146,8 +156,9 @@ def upload_data_to_folder_in_storage_and_publish_single_request_to_queue(
     pages = ref_message["pages"]
 
     for data, page in zip(map(first, data_message_pairs), pages):
-        object_descriptor = file_descriptor_manager.get_input_object_descriptor(ref_message)
+        object_descriptor = file_descriptor_manager.get_input_object_descriptor({**ref_message, "id": page})
         object_descriptor["object_name"] = build_filepath(object_descriptor, page)
+        print(f"{object_descriptor=}")
 
         storage.put_object(**object_descriptor, data=compress(data))
 

test/unit_tests/queue_visitor_test.py

@@ -9,7 +9,7 @@ from test.utils.storage import pack_for_upload
 
 @pytest.fixture()
 def body():
-    return {"dossierId": "folder", "fileId": "file"}
+    return {"dossierId": "dossier_id", "fileId": "file_id"}
 
 
 @pytest.mark.parametrize("client_name", ["mock", "azure", "s3"], scope="session")