793a427c50
Merge in RR/pyinfra from RED-6273-multi-tenant-storage to master
Squashed commit of the following:
commit 0fead1f8b59c9187330879b4e48d48355885c27c
Author: Julius Unverfehrt <julius.unverfehrt@iqser.com>
Date: Tue Mar 28 15:02:22 2023 +0200
fix typos
commit 892a803726946876f8b8cd7905a0e73c419b2fb1
Author: Matthias Bisping <matthias.bisping@axbit.com>
Date: Tue Mar 28 14:41:49 2023 +0200
Refactoring
Replace custom storage caching logic with LRU decorator
commit eafcd90260731e3360ce960571f07dee8f521327
Author: Julius Unverfehrt <julius.unverfehrt@iqser.com>
Date: Fri Mar 24 12:50:13 2023 +0100
fix bug in storage connection from endpoint
commit d0c9fb5b7d1c55ae2f90e8faa1efec9f7587c26a
Author: Julius Unverfehrt <julius.unverfehrt@iqser.com>
Date: Fri Mar 24 11:49:34 2023 +0100
add logs to PayloadProcessor
- set log messages to determine if x-tenant
storage connection is working
commit 97309fe58037b90469cf7a3de342d4749a0edfde
Author: Julius Unverfehrt <julius.unverfehrt@iqser.com>
Date: Fri Mar 24 10:41:59 2023 +0100
update PayloadProcessor
- introduce storage cache to make every unique
storage connection only once
- add functionality to pass optional processing
kwargs in queue message like the operation key to
the processing function
commit d48e8108fdc0d463c89aaa0d672061ab7dca83a0
Author: Julius Unverfehrt <julius.unverfehrt@iqser.com>
Date: Wed Mar 22 13:34:43 2023 +0100
add multi-tenant storage connection 1st iteration
- forward x-tenant-id from queue message header to
payload processor
- add functions to receive storage infos from an
endpoint or the config. This enables hashing and
caching of connections created from these infos
- add function to initialize storage connections
from storage infos
- streamline and refactor tests to make them more
readable and robust and to make it easier to add
new tests
- update payload processor with first iteration
of multi tenancy storage connection support
with connection caching and backwards compability
commit 52c047c47b98e62d0b834a9b9b6c0e2bb0db41e5
Author: Julius Unverfehrt <julius.unverfehrt@iqser.com>
Date: Tue Mar 21 15:35:57 2023 +0100
add AES/GCM cipher functions
- decrypt x-tenant storage connection strings
73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
import gzip
|
|
import json
|
|
import logging
|
|
from operator import itemgetter
|
|
|
|
import pika
|
|
|
|
from pyinfra.config import get_config
|
|
from pyinfra.queue.development_queue_manager import DevelopmentQueueManager
|
|
from pyinfra.storage.storages.s3 import get_s3_storage_from_config
|
|
|
|
CONFIG = get_config()
|
|
logging.basicConfig()
|
|
logger = logging.getLogger()
|
|
logger.setLevel(logging.INFO)
|
|
|
|
|
|
def upload_json_and_make_message_body():
|
|
bucket = CONFIG.storage_bucket
|
|
dossier_id, file_id, suffix = "dossier", "file", "json.gz"
|
|
content = {
|
|
"numberOfPages": 7,
|
|
"sectionTexts": "data",
|
|
}
|
|
|
|
object_name = f"{dossier_id}/{file_id}.{suffix}"
|
|
data = gzip.compress(json.dumps(content).encode("utf-8"))
|
|
|
|
storage = get_s3_storage_from_config(CONFIG)
|
|
if not storage.has_bucket(bucket):
|
|
storage.make_bucket(bucket)
|
|
storage.put_object(bucket, object_name, data)
|
|
|
|
message_body = {
|
|
"dossierId": dossier_id,
|
|
"fileId": file_id,
|
|
"targetFileExtension": suffix,
|
|
"responseFileExtension": f"result.{suffix}",
|
|
}
|
|
return message_body
|
|
|
|
|
|
def main():
|
|
development_queue_manager = DevelopmentQueueManager(CONFIG)
|
|
development_queue_manager.clear_queues()
|
|
|
|
message = upload_json_and_make_message_body()
|
|
|
|
development_queue_manager.publish_request(message, pika.BasicProperties(headers={"X-TENANT-ID": "redaction"}))
|
|
logger.info(f"Put {message} on {CONFIG.request_queue}")
|
|
|
|
storage = get_s3_storage_from_config(CONFIG)
|
|
for method_frame, properties, body in development_queue_manager._channel.consume(
|
|
queue=CONFIG.response_queue, inactivity_timeout=15
|
|
):
|
|
if not body:
|
|
break
|
|
response = json.loads(body)
|
|
logger.info(f"Received {response}")
|
|
logger.info(f"Message headers: {properties.headers}")
|
|
development_queue_manager._channel.basic_ack(method_frame.delivery_tag)
|
|
dossier_id, file_id = itemgetter("dossierId", "fileId")(response)
|
|
suffix = message["responseFileExtension"]
|
|
print(f"{dossier_id}/{file_id}.{suffix}")
|
|
result = storage.get_object(CONFIG.storage_bucket, f"{dossier_id}/{file_id}.{suffix}")
|
|
result = json.loads(gzip.decompress(result))
|
|
logger.info(f"Contents of result on storage: {result}")
|
|
development_queue_manager.close_channel()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|