111 lines
4.3 KiB
Python
111 lines
4.3 KiB
Python
import logging
|
|
from functools import lru_cache
|
|
|
|
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.pipeline_factory import CachedPipelineFactory
|
|
from pyinfra.queue.consumer import Consumer
|
|
from pyinfra.queue.queue_manager.pika_queue_manager import PikaQueueManager
|
|
from pyinfra.server.client_pipeline import ClientPipeline
|
|
from pyinfra.server.dispatcher.dispatchers.rest import RestDispatcher
|
|
from pyinfra.server.interpreter.interpreters.rest_callback import RestPickupStreamer
|
|
from pyinfra.server.packer.packers.rest import RestPacker
|
|
from pyinfra.server.receiver.receivers.rest import RestReceiver
|
|
from pyinfra.storage import storages
|
|
from pyinfra.visitor import QueueVisitor
|
|
from pyinfra.visitor.downloader import Downloader
|
|
from pyinfra.visitor.response_formatter.formatters.default import DefaultResponseFormatter
|
|
from pyinfra.visitor.response_formatter.formatters.identity import IdentityResponseFormatter
|
|
from pyinfra.visitor.strategies.response.aggregation import AggregationStorageStrategy, ProjectingUploadFormatter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ComponentFactory:
|
|
def __init__(self, config):
|
|
self.config = config
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_consumer(self, callback=None):
|
|
callback = callback or self.get_callback()
|
|
return Consumer(self.get_visitor(callback), self.get_queue_manager())
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_callback(self, analysis_base_url=None):
|
|
analysis_base_url = analysis_base_url or self.config.rabbitmq.callback.analysis_endpoint
|
|
|
|
callback = Callback(CachedPipelineFactory(base_url=analysis_base_url, pipeline_factory=self.get_pipeline))
|
|
|
|
def wrapped(body):
|
|
body_repr = project(body, ["dossierId", "fileId", "operation"])
|
|
logger.info(f"Processing {body_repr}...")
|
|
result = callback(body)
|
|
logger.info(f"Completed processing {body_repr}...")
|
|
return result
|
|
|
|
return wrapped
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_visitor(self, callback):
|
|
return QueueVisitor(
|
|
callback=callback,
|
|
data_loader=self.get_downloader(),
|
|
response_strategy=self.get_response_strategy(),
|
|
response_formatter=self.get_response_formatter(),
|
|
)
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_queue_manager(self):
|
|
return PikaQueueManager(self.config.rabbitmq.queues.input, self.config.rabbitmq.queues.output)
|
|
|
|
@staticmethod
|
|
@lru_cache(maxsize=None)
|
|
def get_pipeline(endpoint):
|
|
return ClientPipeline(
|
|
RestPacker(), RestDispatcher(endpoint), RestReceiver(), rcompose(RestPickupStreamer(), RestReceiver())
|
|
)
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_storage(self):
|
|
return storages.get_storage(self.config.storage.backend)
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_response_strategy(self, storage=None):
|
|
return AggregationStorageStrategy(
|
|
storage=storage or self.get_storage(),
|
|
file_descriptor_manager=self.get_file_descriptor_manager(),
|
|
upload_formatter=self.get_upload_formatter(),
|
|
)
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_file_descriptor_manager(self):
|
|
return FileDescriptorManager(
|
|
bucket_name=parse_disjunction_string(self.config.storage.bucket),
|
|
operation2file_patterns=self.get_operation2file_patterns(),
|
|
)
|
|
|
|
@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_response_formatter(self):
|
|
return {"default": DefaultResponseFormatter(), "identity": IdentityResponseFormatter()}[
|
|
self.config.service.response_formatter
|
|
]
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_operation2file_patterns(self):
|
|
return self.config.service.operations
|
|
|
|
@lru_cache(maxsize=None)
|
|
def get_downloader(self, storage=None):
|
|
return Downloader(
|
|
storage=storage or self.get_storage(),
|
|
bucket_name=parse_disjunction_string(self.config.storage.bucket),
|
|
file_descriptor_manager=self.get_file_descriptor_manager(),
|
|
)
|