diff --git a/config.yaml b/config.yaml index c044f4a..b03a504 100755 --- a/config.yaml +++ b/config.yaml @@ -2,7 +2,7 @@ service: logging_level: $LOGGING_LEVEL_ROOT|INFO # Logging level for service logger response: type: $RESPONSE_TYPE|"file" # Whether the analysis response is stored as file on storage or sent as stream - extension: $RESPONSE_FILE_EXTENSION|".IMAGE_INFO.json.gz" # {.IMAGE_INFO.json.gz | .NER_ENTITIES.json.gz} + extension: $RESPONSE_FILE_EXTENSION|"IMAGE_INFO.json.gz" # {.IMAGE_INFO.json.gz | .NER_ENTITIES.json.gz} key: $RESPONSE_KEY|"imageMetadata" # the key of the result {result, imageMetadata} probing_webserver: @@ -38,7 +38,7 @@ storage: backend: $STORAGE_BACKEND|s3 # The type of storage to use {s3, azure} bucket: $STORAGE_BUCKET|"pyinfra-test-bucket" # The bucket / container to pull files specified in queue requests from # TODO: Caller should specify exact file name, including extension! - target_file_extension: $TARGET_FILE_EXTENSION|".ORIGIN.pdf.gz" # {.TEXT.json.gz | .ORIGIN.pdf.gz} Defines type of file to pull from storage + target_file_extension: $TARGET_FILE_EXTENSION|"ORIGIN.pdf.gz" # {.TEXT.json.gz | .ORIGIN.pdf.gz} Defines type of file to pull from storage s3: endpoint: $STORAGE_ENDPOINT|"http://127.0.0.1:9000" diff --git a/pyinfra/test/unit_tests/conftest.py b/pyinfra/test/unit_tests/conftest.py index 7f72087..0d89c54 100644 --- a/pyinfra/test/unit_tests/conftest.py +++ b/pyinfra/test/unit_tests/conftest.py @@ -22,7 +22,7 @@ logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) -@pytest.fixture +@pytest.fixture(scope="session") def bucket_name(): return "pyinfra-test-bucket" @@ -54,7 +54,7 @@ def mock_make_load_data(): return load_data -@pytest.fixture(params=["minio", "aws"]) +@pytest.fixture(params=["minio", "aws"], scope="session") def storage(client_name, bucket_name, request): logger.debug("Setup for storage") storage = Storage(get_adapter(client_name, request.param)) @@ -86,3 +86,8 @@ def get_queue_manager(queue_manager_name) -> QueueManager: @pytest.fixture(scope="session") def queue_manager(queue_manager_name) -> QueueManager: return get_queue_manager(queue_manager_name) + + +@pytest.fixture +def callback(): + return lambda x: x**2 diff --git a/pyinfra/test/unit_tests/consumer_test.py b/pyinfra/test/unit_tests/consumer_test.py index 8a613b9..ffbede8 100644 --- a/pyinfra/test/unit_tests/consumer_test.py +++ b/pyinfra/test/unit_tests/consumer_test.py @@ -3,11 +3,6 @@ import pytest from pyinfra.pyampq.consumer import Consumer -@pytest.fixture -def callback(): - return lambda x: x**2 - - @pytest.fixture def consumer(queue_manager, callback): return Consumer(callback, queue_manager) diff --git a/pyinfra/test/unit_tests/queue_visitor_test.py b/pyinfra/test/unit_tests/queue_visitor_test.py new file mode 100644 index 0000000..ada1488 --- /dev/null +++ b/pyinfra/test/unit_tests/queue_visitor_test.py @@ -0,0 +1,23 @@ +import pytest + +from pyinfra.visitor import QueueVisitor, get_object_descriptor + + +@pytest.fixture() +def visitor(storage, callback): + return QueueVisitor(storage, callback) + + +@pytest.fixture() +def body(): + return {"dossierId": "folder", "fileId": "file"} + + +@pytest.mark.parametrize("client_name", ["mock", "azure", "s3"], scope="session") +class TestStorage: + + def test_given_a_input_queue_message_callback_pulls_the_data_from_storage(self, visitor, body, storage, bucket_name): + storage.clear_bucket(bucket_name) + storage.put_object(**get_object_descriptor(body), data=b"content") + data_received = visitor.load_data(body) + assert b"content" == data_received diff --git a/pyinfra/test/unit_tests/storage_test.py b/pyinfra/test/unit_tests/storage_test.py index 01a1d93..6e8293f 100644 --- a/pyinfra/test/unit_tests/storage_test.py +++ b/pyinfra/test/unit_tests/storage_test.py @@ -9,7 +9,7 @@ logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) -@pytest.mark.parametrize("client_name", ["mock", "azure", "s3"]) +@pytest.mark.parametrize("client_name", ["mock", "azure", "s3"], scope="session") class TestStorage: def test_clearing_bucket_yields_empty_bucket(self, storage, bucket_name): storage.clear_bucket(bucket_name) @@ -17,26 +17,31 @@ class TestStorage: assert not {*data_received} def test_getting_object_put_in_bucket_is_object(self, storage, bucket_name): + storage.clear_bucket(bucket_name) storage.put_object(bucket_name, "file", b"content") data_received = storage.get_object(bucket_name, "file") assert b"content" == data_received def test_getting_nested_object_put_in_bucket_is_nested_object(self, storage, bucket_name): + storage.clear_bucket(bucket_name) storage.put_object(bucket_name, "folder/file", b"content") data_received = storage.get_object(bucket_name, "folder/file") assert b"content" == data_received def test_getting_objects_put_in_bucket_are_objects(self, storage, bucket_name): + storage.clear_bucket(bucket_name) storage.put_object(bucket_name, "file1", b"content 1") storage.put_object(bucket_name, "folder/file2", b"content 2") data_received = storage.get_all_objects(bucket_name) assert {b"content 1", b"content 2"} == {*data_received} def test_make_bucket_produces_bucket(self, storage, bucket_name): + storage.clear_bucket(bucket_name) storage.make_bucket(bucket_name) assert storage.has_bucket(bucket_name) def test_listing_bucket_files_yields_all_files_in_bucket(self, storage, bucket_name): + storage.clear_bucket(bucket_name) storage.put_object(bucket_name, "file1", b"content 1") storage.put_object(bucket_name, "file2", b"content 2") full_names_received = storage.get_all_object_names(bucket_name) diff --git a/pyinfra/visitor.py b/pyinfra/visitor.py new file mode 100644 index 0000000..fbe901d --- /dev/null +++ b/pyinfra/visitor.py @@ -0,0 +1,35 @@ +from operator import itemgetter +from typing import Callable + +from pyinfra.config import CONFIG +from pyinfra.storage.storage import Storage + + +def get_object_name(body): + dossier_id, file_id = itemgetter("dossierId", "fileId")(body) + object_name = f"{dossier_id}/{file_id}.{CONFIG.target_file_extension}" + return object_name + + +def get_object_descriptor(body): + return { + "bucket_name": CONFIG.storage.bucket, + # TODO: send complete filename in request, as discussed! + "object_name": get_object_name(body) + } + + +class QueueVisitor: + + def __init__(self, storage: Storage, callback: Callable): + self.storage = storage + self.callback = callback + + def load_data(self, body): + return self.storage.get_object(**get_object_descriptor(body)) + + def process_data(self): + pass + + def __call__(self, body): + pass