diff --git a/config.yaml b/config.yaml index b51f939..1399cee 100755 --- a/config.yaml +++ b/config.yaml @@ -38,7 +38,7 @@ service: default: input: subdir: "" - extension: IN.gz + extension: in.gz output: subdir: "" extension: out.gz diff --git a/pyinfra/component_factory.py b/pyinfra/component_factory.py index 564c554..85727b1 100644 --- a/pyinfra/component_factory.py +++ b/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()}[ diff --git a/pyinfra/file_descriptor_builder.py b/pyinfra/file_descriptor_builder.py new file mode 100644 index 0000000..3965802 --- /dev/null +++ b/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 diff --git a/pyinfra/file_descriptor_manager.py b/pyinfra/file_descriptor_manager.py index 9f6b810..8c1072a 100644 --- a/pyinfra/file_descriptor_manager.py +++ b/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)