partially decomplected file descriptor manager from concrete and non-generic descriptor code

config.yaml

@@ -38,7 +38,7 @@ service:
     default:
       input:
         subdir: ""
-        extension: IN.gz
+        extension: in.gz
       output:
         subdir: ""
         extension: out.gz

pyinfra/component_factory.py

@@ -6,6 +6,7 @@ from funcy import project, identity, rcompose
 from pyinfra.callback import Callback
 from pyinfra.config import parse_disjunction_string
 from pyinfra.file_descriptor_manager import FileDescriptorManager
+from pyinfra.file_descriptor_builder import FileDescriptorBuilder, RedFileDescriptorBuilder
 from pyinfra.pipeline_factory import CachedPipelineFactory
 from pyinfra.queue.consumer import Consumer
 from pyinfra.queue.queue_manager.pika_queue_manager import PikaQueueManager
@@ -84,13 +85,20 @@ class ComponentFactory:
     def get_file_descriptor_manager(self):
         return FileDescriptorManager(
             bucket_name=parse_disjunction_string(self.config.storage.bucket),
-            operation2file_patterns=self.get_operation2file_patterns(),
+            file_descriptor_builder=self.get_operation_file_descriptor_builder(),
         )
 
     @lru_cache(maxsize=None)
     def get_upload_formatter(self):
         return {"identity": identity, "projecting": ProjectingUploadFormatter()}[self.config.service.upload_formatter]
 
+    @lru_cache(maxsize=None)
+    def get_operation_file_descriptor_builder(self):
+        return RedFileDescriptorBuilder(
+            operation2file_patterns=self.get_operation2file_patterns(),
+            default_operation_name=self.config.service.operation
+        )
+
     @lru_cache(maxsize=None)
     def get_response_formatter(self):
         return {"default": DefaultResponseFormatter(), "identity": IdentityResponseFormatter()}[

pyinfra/file_descriptor_builder.py

@@ -0,0 +1,87 @@
+import abc
+import os
+from operator import itemgetter
+
+from funcy import project
+
+
+class FileDescriptorBuilder:
+    @abc.abstractmethod
+    def build_file_descriptor(self, queue_item_body, end="input"):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def build_matcher(self, file_descriptor):
+        raise NotImplementedError
+
+    @staticmethod
+    @abc.abstractmethod
+    def build_storage_upload_info(analysis_payload, request_metadata):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def get_path_prefix(self, queue_item_body):
+        raise NotImplementedError
+
+
+class RedFileDescriptorBuilder(FileDescriptorBuilder):
+    """Defines concrete descriptors for storage objects based on queue messages"""
+
+    def __init__(self, operation2file_patterns, default_operation_name):
+
+        self.operation2file_patterns = operation2file_patterns or self.get_default_operation2file_patterns()
+        self.default_operation_name = default_operation_name
+
+    @staticmethod
+    def get_default_operation2file_patterns():
+        return {"default": {"input": {"subdir": "", "extension": ".in"}, "output": {"subdir": "", "extension": ".out"}}}
+
+    def build_file_descriptor(self, queue_item_body, end="input"):
+
+        operation = queue_item_body.get("operation", self.default_operation_name)
+
+        file_pattern = self.operation2file_patterns[operation][end]
+
+        file_descriptor = {
+            **project(queue_item_body, ["dossierId", "fileId", "pages"]),
+            "pages": queue_item_body.get("pages", []),
+            "extension": file_pattern["extension"],
+            "subdir": file_pattern["subdir"],
+        }
+
+        return file_descriptor
+
+    def build_matcher(self, file_descriptor):
+        def make_filename(file_id, subdir, suffix):
+            return os.path.join(file_id, subdir, suffix) if subdir else f"{file_id}.{suffix}"
+
+        dossier_id, file_id, subdir, pages, extension = itemgetter(
+            "dossierId", "fileId", "subdir", "pages", "extension"
+        )(file_descriptor)
+
+        matcher = os.path.join(dossier_id, make_filename(file_id, subdir, self.__build_page_regex(pages) + extension))
+
+        return matcher
+
+    @staticmethod
+    def __build_page_regex(pages):
+
+        n_pages = len(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+."
+
+        return page_re
+
+    @staticmethod
+    def build_storage_upload_info(analysis_payload, request_metadata):
+        storage_upload_info = {**request_metadata, "id": analysis_payload["metadata"].get("id", 0)}
+        return storage_upload_info
+
+    def get_path_prefix(self, queue_item_body):
+        prefix = "/".join(itemgetter("dossierId", "fileId")(self.build_file_descriptor(queue_item_body, end="input")))
+        return prefix

pyinfra/file_descriptor_manager.py

@@ -1,19 +1,17 @@
-import os
-from _operator import itemgetter
-
-from funcy import project
-
-from pyinfra.config import CONFIG
+from pyinfra.file_descriptor_builder import FileDescriptorBuilder
 
 
 class FileDescriptorManager:
-    def __init__(self, bucket_name, operation2file_patterns: dict = None):
-        self.bucket_name = bucket_name
-        self.operation2file_patterns = operation2file_patterns or self.get_default_operation2file_patterns()
+    """Decorates a file descriptor builder with additional convenience functionality and this way provides a
+    comprehensive interface for all file descriptor related operations, while the concrete descriptor logic is
+    implemented in a file descriptor builder.
 
-    @staticmethod
-    def get_default_operation2file_patterns():
-        return {"default": {"input": {"subdir": "", "extension": ".in"}, "output": {"subdir": "", "extension": ".out"}}}
+    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
 
     def get_input_object_name(self, queue_item_body: dict):
         return self.get_object_name(queue_item_body, end="input")
@@ -24,52 +22,14 @@ class FileDescriptorManager:
     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
 
-    def __build_matcher(self, file_descriptor):
-        def make_filename(file_id, subdir, suffix):
-            return os.path.join(file_id, subdir, suffix) if subdir else f"{file_id}.{suffix}"
-
-        dossier_id, file_id, subdir, pages, extension = itemgetter(
-            "dossierId", "fileId", "subdir", "pages", "extension"
-        )(file_descriptor)
-
-        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, subdir):
-        if len(pages) > 1 and subdir:
-            page_re = "id:(" + "|".join(map(str, pages)) + ")."
-        elif not pages and subdir:
-            page_re = r"id:\d+."
-        elif len(pages) == 1 and subdir:
-            page_re = f"id:{pages[0]}."
-        else:
-            page_re = ""
-
-        return page_re
-
     def build_file_descriptor(self, queue_item_body, end="input"):
-        # FIXME: reference to config is misplaced here. Needs to be abstracted in some way.
-        operation = queue_item_body.get("operation", CONFIG.service.operation)
-
-        file_pattern = self.operation2file_patterns[operation][end]
-
-        file_descriptor = {
-            **project(queue_item_body, ["dossierId", "fileId", "pages"]),
-            "pages": queue_item_body.get("pages", []),
-            "extension": file_pattern["extension"],
-            "subdir": file_pattern["subdir"],
-        }
-        return file_descriptor
+        return self.operation_file_descriptor_builder.build_file_descriptor(queue_item_body, end=end)
 
     def build_input_matcher(self, queue_item_body):
         return self.build_matcher(queue_item_body, end="input")
@@ -81,6 +41,9 @@ class FileDescriptorManager:
         file_descriptor = self.build_file_descriptor(queue_item_body, end=end)
         return self.__build_matcher(file_descriptor)
 
+    def __build_matcher(self, file_descriptor):
+        return self.operation_file_descriptor_builder.build_matcher(file_descriptor)
+
     def get_input_object_descriptor(self, queue_item_body):
         return self.get_object_descriptor(queue_item_body, end="input")
 
@@ -88,21 +51,15 @@ class FileDescriptorManager:
         return self.get_object_descriptor(storage_upload_info, end="output")
 
     def get_object_descriptor(self, queue_item_body, end):
+        # TODO: this is complected with the Storage class API
+        # FIXME: bad coupling
         return {
             "bucket_name": self.bucket_name,
             "object_name": self.get_object_name(queue_item_body, end=end),
         }
 
-    @staticmethod
-    def build_storage_upload_info(analysis_payload, request_metadata):
-        storage_upload_info = {**request_metadata, "id": analysis_payload["metadata"].get("id", 0)}
-        return storage_upload_info
+    def build_storage_upload_info(self, analysis_payload, request_metadata):
+        return self.operation_file_descriptor_builder.build_storage_upload_info(analysis_payload, request_metadata)
 
     def get_path_prefix(self, queue_item_body):
-
-        prefix = "/".join(
-            itemgetter("dossierId", "fileId")(
-                self.build_file_descriptor(queue_item_body, end="input"),
-            )
-        )
-        return prefix
+        return self.operation_file_descriptor_builder.get_path_prefix(queue_item_body)