chore: fuzzy pin kn-utils to allow for future updates

feat:BREAKING CHANGE: download callback no forwards all files as bytes

See merge request knecon/research/pyinfra!108

.dvc/.gitignore

@@ -0,0 +1,2 @@
+/config.local
+/cache

.dvc/config

@@ -0,0 +1,5 @@
+[core]
+    remote = azure
+['remote "azure"']
+    url = azure://pyinfra-dvc
+    connection_string =

.dvcignore

@@ -0,0 +1,3 @@
+# Add patterns of files dvc should ignore, which could improve
+# the performance. Learn more at
+# https://dvc.org/doc/user-guide/dvcignore

.gitignore

@@ -31,7 +31,6 @@ __pycache__/
 # file extensions
 *.log
 *.csv
-*.json
 *.pkl
 *.profile
 *.cbm

.gitlab-ci.yml

@@ -1,11 +1,23 @@
+# CI for services, check gitlab repo for python package CI
 include:
   - project: "Gitlab/gitlab"
-    ref: 0.3.0
-    file: "/ci-templates/research/python_pkg_venv_test_build_release_gitlab-ci.yml"
+    ref: main
+    file: "/ci-templates/research/python_pkg-test-build-release.gitlab-ci.yml"
 
-default:
-  image: python:3.10
+# set project variables here
+variables:
+  NEXUS_PROJECT_DIR: research # subfolder in Nexus docker-gin where your container will be stored
+  IMAGENAME: $CI_PROJECT_NAME # if the project URL is gitlab.example.com/group-name/project-1, CI_PROJECT_NAME is project-1
+  REPORTS_DIR: reports
+  FF_USE_FASTZIP: "true" # enable fastzip - a faster zip implementation that also supports level configuration.
+  ARTIFACT_COMPRESSION_LEVEL: default # can also be set to fastest, fast, slow and slowest. If just enabling fastzip is not enough try setting this to fastest or fast.
+  CACHE_COMPRESSION_LEVEL: default # same as above, but for caches
+  # TRANSFER_METER_FREQUENCY: 5s # will display transfer progress every 5 seconds for artifacts and remote caches. For debugging purposes.
 
-run-tests:
-  script:
-    - echo "Disabled until we have an automated way to run docker compose before tests."
+
+############
+# UNIT TESTS
+unit-tests:
+  variables:
+    ###### UPDATE/EDIT ######
+    UNIT_TEST_DIR: "tests/unit_test"

.pre-commit-config.yaml

@@ -1,42 +1,55 @@
 # See https://pre-commit.com for more information
 # See https://pre-commit.com/hooks.html for more hooks
-exclude: ^(docs/|notebooks/|data/|src/secrets/|src/static/|src/templates/|tests)
+exclude: ^(docs/|notebooks/|data/|src/configs/|tests/|.hooks/)
 default_language_version:
-  python: python3.8
+  python: python3.10
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v5.0.0
     hooks:
       - id: trailing-whitespace
       - id: end-of-file-fixer
       - id: check-yaml
-        exclude: bamboo-specs/bamboo.yml
+        name: Check Gitlab CI (unsafe)
+        args: [--unsafe]
+        files: .gitlab-ci.yml
+      - id: check-yaml
+        exclude: .gitlab-ci.yml
+      - id: check-toml
+      - id: detect-private-key
+      - id: check-added-large-files
+        args: ['--maxkb=10000']
+      - id: check-case-conflict
+      - id: mixed-line-ending
 
-  # - repo: https://github.com/pycqa/pylint
-  #   rev: v2.16.1
-  #   hooks:
-  #     - id: pylint
-  #       args:
-  #         ["--max-line-length=120", "--errors-only", "--ignore-imports=true", ]
+  - repo: https://github.com/pre-commit/mirrors-pylint
+    rev: v3.0.0a5
+    hooks:
+      - id: pylint
+        language: system
+        args:
+          - --disable=C0111,R0903
+          - --max-line-length=120
 
   - repo: https://github.com/pre-commit/mirrors-isort
     rev: v5.10.1
     hooks:
       - id: isort
-        args: ["--profile", "black"]
+        args:
+          - --profile black
 
   - repo: https://github.com/psf/black
-    rev: 23.1.0
+    rev: 24.10.0
     hooks:
       - id: black
         # exclude: ^(docs/|notebooks/|data/|src/secrets/)
         args:
           - --line-length=120
-# -   repo: local
-#     hooks:
-#     -   id: system
-#         name: PyLint
-#         entry: poetry run pylint
-#         language: system
-#         exclude: ^alembic/
-#         files: \.py$
+
+  - repo: https://github.com/compilerla/conventional-pre-commit
+    rev: v3.6.0
+    hooks:
+      - id: conventional-pre-commit
+        pass_filenames: false
+        stages: [commit-msg]
+        # args: [] # optional: list of Conventional Commits types to allow e.g. [feat, fix, ci, chore, test]

.python-version

@@ -1 +1 @@
-3.10.12
+3.10

README.md

@@ -6,13 +6,13 @@
 4. [ Module Installation ](#module-installation)
 5. [ Scripts ](#scripts)
 6. [ Tests ](#tests)
-
+7. [ Opentelemetry protobuf dependency hell ](#opentelemetry-protobuf-dependency-hell)
 
 ## About
 
 Shared library for the research team, containing code related to infrastructure and communication with other services.
 Offers a simple interface for processing data and sending responses via AMQP, monitoring via Prometheus and storage
-access via S3 or Azure.
+access via S3 or Azure. Also export traces via OpenTelemetry for queue messages and webserver requests.
 
 To start, see the [complete example](pyinfra/examples.py) which shows how to use all features of the service and can be
 imported and used directly for default research service pipelines (data ID in message, download data from storage,
@@ -31,33 +31,68 @@ The following table shows all necessary settings. You can find a preconfigured s
 bitbucket. These are the complete settings, you only need all if using all features of the service as described in
 the [complete example](pyinfra/examples.py).
 
-| Environment Variable               | Internal / .toml Name            | Description                                                                                                                                                                                                                                                             |
-|------------------------------------|----------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| LOGGING__LEVEL                     | logging.level                    | Log level                                                                                                                                                                                                                                                               |
-| METRICS__PROMETHEUS__ENABLED       | metrics.prometheus.enabled       | Enable Prometheus metrics collection                                                                                                                                                                                                                                    |
-| METRICS__PROMETHEUS__PREFIX        | metrics.prometheus.prefix        | Prefix for Prometheus metrics (e.g. {product}-{service})                                                                                                                                                                                                                |
-| WEBSERVER__HOST                    | webserver.host                   | Host of the webserver (offering e.g. /prometheus, /ready and /health endpoints)                                                                                                                                                                                         |
-| WEBSERVER__PORT                    | webserver.port                   | Port of the webserver                                                                                                                                                                                                                                                   |
-| RABBITMQ__HOST                     | rabbitmq.host                    | Host of the RabbitMQ server                                                                                                                                                                                                                                             |
-| RABBITMQ__PORT                     | rabbitmq.port                    | Port of the RabbitMQ server                                                                                                                                                                                                                                             |
-| RABBITMQ__USERNAME                 | rabbitmq.username                | Username for the RabbitMQ server                                                                                                                                                                                                                                        |
-| RABBITMQ__PASSWORD                 | rabbitmq.password                | Password for the RabbitMQ server                                                                                                                                                                                                                                        |
-| RABBITMQ__HEARTBEAT                | rabbitmq.heartbeat               | Heartbeat for the RabbitMQ server                                                                                                                                                                                                                                       |
-| RABBITMQ__CONNECTION_SLEEP         | rabbitmq.connection_sleep        | Sleep time intervals during message processing. Has to be a divider of heartbeat, and shouldn't be too big, since only in these intervals queue interactions happen (like receiving new messages) This is also the minimum time the service needs to process a message. |
-| RABBITMQ__INPUT_QUEUE              | rabbitmq.input_queue             | Name of the input queue                                                                                                                                                                                                                                                 |
-| RABBITMQ__OUTPUT_QUEUE             | rabbitmq.output_queue            | Name of the output queue                                                                                                                                                                                                                                                |
-| RABBITMQ__DEAD_LETTER_QUEUE        | rabbitmq.dead_letter_queue       | Name of the dead letter queue                                                                                                                                                                                                                                           |
-| STORAGE__BACKEND                   | storage.backend                  | Storage backend to use (currently only "s3" and "azure" are supported)                                                                                                                                                                                                  |
-| STORAGE__CACHE_SIZE                | storage.cache_size               | Number of cached storage connection (to reduce connection stops and reconnects for multi tenancy).                                                                                                                                                                      |
-| STORAGE__S3__BUCKET_NAME           | storage.s3.bucket_name           | Name of the S3 bucket                                                                                                                                                                                                                                                   |
-| STORAGE__S3__ENDPOINT              | storage.s3.endpoint              | Endpoint of the S3 server                                                                                                                                                                                                                                               |
-| STORAGE__S3__KEY                   | storage.s3.key                   | Access key for the S3 server                                                                                                                                                                                                                                            |
-| STORAGE__S3__SECRET                | storage.s3.secret                | Secret key for the S3 server                                                                                                                                                                                                                                            |
-| STORAGE__S3__REGION                | storage.s3.region                | Region of the S3 server                                                                                                                                                                                                                                                 |
-| STORAGE__AZURE__CONTAINER          | storage.azure.container_name     | Name of the Azure container                                                                                                                                                                                                                                             |
-| STORAGE__AZURE__CONNECTION_STRING  | storage.azure.connection_string  | Connection string for the Azure server                                                                                                                                                                                                                                  |
-| STORAGE__TENANT_SERVER__PUBLIC_KEY | storage.tenant_server.public_key | Public key of the tenant server                                                                                                                                                                                                                                         |
-| STORAGE__TENANT_SERVER__ENDPOINT   | storage.tenant_server.endpoint   | Endpoint of the tenant server                                                                                                                                                                                                                                           |
+| Environment Variable                       | Internal / .toml Name                   | Description                                                                                                                                                                                                                                                             |
+| ------------------------------------------ | --------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| LOGGING\_\_LEVEL                           | logging.level                           | Log level                                                                                                                                                                                                                                                               |
+| DYNAMIC_TENANT_QUEUES\_\_ENABLED                     | dynamic_tenant_queues.enabled                     | Enable queues per tenant that are dynamically created mode                                                                                                                                                                                                                                          |
+| METRICS\_\_PROMETHEUS\_\_ENABLED           | metrics.prometheus.enabled              | Enable Prometheus metrics collection                                                                                                                                                                                                                                    |
+| METRICS\_\_PROMETHEUS\_\_PREFIX            | metrics.prometheus.prefix               | Prefix for Prometheus metrics (e.g. {product}-{service})                                                                                                                                                                                                                |
+| WEBSERVER\_\_HOST                          | webserver.host                          | Host of the webserver (offering e.g. /prometheus, /ready and /health endpoints)                                                                                                                                                                                         |
+| WEBSERVER\_\_PORT                          | webserver.port                          | Port of the webserver                                                                                                                                                                                                                                                   |
+| RABBITMQ\_\_HOST                           | rabbitmq.host                           | Host of the RabbitMQ server                                                                                                                                                                                                                                             |
+| RABBITMQ\_\_PORT                           | rabbitmq.port                           | Port of the RabbitMQ server                                                                                                                                                                                                                                             |
+| RABBITMQ\_\_USERNAME                       | rabbitmq.username                       | Username for the RabbitMQ server                                                                                                                                                                                                                                        |
+| RABBITMQ\_\_PASSWORD                       | rabbitmq.password                       | Password for the RabbitMQ server                                                                                                                                                                                                                                        |
+| RABBITMQ\_\_HEARTBEAT                      | rabbitmq.heartbeat                      | Heartbeat for the RabbitMQ server                                                                                                                                                                                                                                       |
+| RABBITMQ\_\_CONNECTION_SLEEP               | rabbitmq.connection_sleep               | Sleep time intervals during message processing. Has to be a divider of heartbeat, and shouldn't be too big, since only in these intervals queue interactions happen (like receiving new messages) This is also the minimum time the service needs to process a message. |
+| RABBITMQ\_\_INPUT_QUEUE                    | rabbitmq.input_queue                    | Name of the input queue in single queue setting                                                                                                                                                                                                                         |
+| RABBITMQ\_\_OUTPUT_QUEUE                   | rabbitmq.output_queue                   | Name of the output queue in single queue setting                                                                                                                                                                                                                        |
+| RABBITMQ\_\_DEAD_LETTER_QUEUE              | rabbitmq.dead_letter_queue              | Name of the dead letter queue in single queue setting                                                                                                                                                                                                                   |
+| RABBITMQ\_\_TENANT_EVENT_QUEUE_SUFFIX      | rabbitmq.tenant_event_queue_suffix      | Suffix for the tenant event queue in multi tenant/queue setting                                                                                                                                                                                                         |
+| RABBITMQ\_\_TENANT_EVENT_DLQ_SUFFIX        | rabbitmq.tenant_event_dlq_suffix        | Suffix for the dead letter queue in multi tenant/queue setting                                                                                                                                                                                                          |
+| RABBITMQ\_\_TENANT_EXCHANGE_NAME           | rabbitmq.tenant_exchange_name           | Name of tenant exchange in multi tenant/queue setting                                                                                                                                                                                                                   |
+| RABBITMQ\_\_QUEUE_EXPIRATION_TIME          | rabbitmq.queue_expiration_time          | Time until queue expiration in multi tenant/queue setting                                                                                                                                                                                                               |
+| RABBITMQ\_\_SERVICE_REQUEST_QUEUE_PREFIX   | rabbitmq.service_request_queue_prefix   | Service request queue prefix in multi tenant/queue setting                                                                                                                                                                                                              |
+| RABBITMQ\_\_SERVICE_REQUEST_EXCHANGE_NAME  | rabbitmq.service_request_exchange_name  | Service request exchange name in multi tenant/queue setting                                                                                                                                                                                                             |
+| RABBITMQ\_\_SERVICE_RESPONSE_EXCHANGE_NAME | rabbitmq.service_response_exchange_name | Service response exchange name in multi tenant/queue setting                                                                                                                                                                                                            |
+| RABBITMQ\_\_SERVICE_DLQ_NAME               | rabbitmq.service_dlq_name               | Service dead letter queue name in multi tenant/queue setting                                                                                                                                                                                                            |
+| STORAGE\_\_BACKEND                         | storage.backend                         | Storage backend to use (currently only "s3" and "azure" are supported)                                                                                                                                                                                                  |
+| STORAGE\_\_S3\_\_BUCKET                    | storage.s3.bucket                       | Name of the S3 bucket                                                                                                                                                                                                                                                   |
+| STORAGE\_\_S3\_\_ENDPOINT                  | storage.s3.endpoint                     | Endpoint of the S3 server                                                                                                                                                                                                                                               |
+| STORAGE\_\_S3\_\_KEY                       | storage.s3.key                          | Access key for the S3 server                                                                                                                                                                                                                                            |
+| STORAGE\_\_S3\_\_SECRET                    | storage.s3.secret                       | Secret key for the S3 server                                                                                                                                                                                                                                            |
+| STORAGE\_\_S3\_\_REGION                    | storage.s3.region                       | Region of the S3 server                                                                                                                                                                                                                                                 |
+| STORAGE\_\_AZURE\_\_CONTAINER              | storage.azure.container_name            | Name of the Azure container                                                                                                                                                                                                                                             |
+| STORAGE\_\_AZURE\_\_CONNECTION_STRING      | storage.azure.connection_string         | Connection string for the Azure server                                                                                                                                                                                                                                  |
+| STORAGE\_\_TENANT_SERVER\_\_PUBLIC_KEY     | storage.tenant_server.public_key        | Public key of the tenant server                                                                                                                                                                                                                                         |
+| STORAGE\_\_TENANT_SERVER\_\_ENDPOINT       | storage.tenant_server.endpoint          | Endpoint of the tenant server                                                                                                                                                                                                                                           |
+| TRACING\_\_ENABLED                         | tracing.enabled                         | Enable tracing                                                                                                                                                                                                                                                          |
+| TRACING\_\_TYPE                            | tracing.type                            | Tracing mode - possible values: "opentelemetry", "azure_monitor" (Excpects APPLICATIONINSIGHTS_CONNECTION_STRING environment variable.)                                                                                                                                 |
+| TRACING\_\_OPENTELEMETRY\_\_ENDPOINT       | tracing.opentelemetry.endpoint          | Endpoint to which OpenTelemetry traces are exported                                                                                                                                                                                                                     |
+| TRACING\_\_OPENTELEMETRY\_\_SERVICE_NAME   | tracing.opentelemetry.service_name      | Name of the service as displayed in the traces collected                                                                                                                                                                                                                |
+| TRACING\_\_OPENTELEMETRY\_\_EXPORTER       | tracing.opentelemetry.exporter          | Name of exporter                                                                                                                                                                                                                                                        |
+| KUBERNETES\_\_POD_NAME                     | kubernetes.pod_name                     | Service pod name                                                                                                                                                                                                                                                        |
+
+## Setup
+**IMPORTANT** you need to set the following environment variables before running the setup script:
+- ``$NEXUS_USER`` your Nexus user (usually equal to firstname.lastname@knecon.com)
+- ``$NEXUS_PASSWORD`` your Nexus password (usually equal to your Azure Login)
+
+```shell
+# create venv and activate it
+source ./scripts/setup/devenvsetup.sh {{ cookiecutter.python_version }} $NEXUS_USER $NEXUS_PASSWORD
+source .venv/bin/activate
+```
+
+### OpenTelemetry
+
+Open telemetry (vis its Python SDK) is set up to be as unobtrusive as possible; for typical use cases it can be
+configured
+from environment variables, without additional work in the microservice app, although additional confiuration is
+possible.
+
+`TRACING__OPENTELEMETRY__ENDPOINT` should typically be set
+to `http://otel-collector-opentelemetry-collector.otel-collector:4318/v1/traces`.
 
 ## Queue Manager
 
@@ -66,7 +101,7 @@ to the output queue. The default callback also downloads data from the storage a
 The response message does not contain the data itself, but the identifiers from the input message (including headers
 beginning with "X-").
 
-Usage:
+### Standalone Usage
 
 ```python
 from pyinfra.queue.manager import QueueManager
@@ -77,7 +112,32 @@ settings = load_settings("path/to/settings")
 processing_function: DataProcessor  # function should expect a dict (json) or bytes (pdf) as input and should return a json serializable object.
 
 queue_manager = QueueManager(settings)
-queue_manager.start_consuming(make_download_process_upload_callback(processing_function, settings))
+callback = make_download_process_upload_callback(processing_function, settings)
+queue_manager.start_consuming(make_download_process_upload_callback(callback, settings))
+```
+
+### Usage in a Service
+
+This is the recommended way to use the module. This includes the webserver, Prometheus metrics and health endpoints.
+Custom endpoints can be added by adding a new route to the `app` object beforehand. Settings are loaded from files
+specified as CLI arguments (e.g. `--settings-path path/to/settings.toml`). The values can also be set or overriden via
+environment variables (e.g. `LOGGING__LEVEL=DEBUG`).
+
+The callback can be replaced with a custom one, for example if the data to process is contained in the message itself
+and not on the storage.
+
+```python
+from pyinfra.config.loader import load_settings, parse_settings_path
+from pyinfra.examples import start_standard_queue_consumer
+from pyinfra.queue.callback import make_download_process_upload_callback, DataProcessor
+
+processing_function: DataProcessor
+
+arguments = parse_settings_path()
+settings = load_settings(arguments.settings_path)
+
+callback = make_download_process_upload_callback(processing_function, settings)
+start_standard_queue_consumer(callback, settings)  # optionally also pass a fastAPI app object with preconfigured routes
 ```
 
 ### AMQP input message:
@@ -139,7 +199,7 @@ $ cd tests && docker compose up
 **Shell 2**: Start pyinfra with callback mock
 
 ```bash
-$ python scripts/start_pyinfra.py 
+$ python scripts/start_pyinfra.py
 ```
 
 **Shell 3**: Upload dummy content on storage and publish message
@@ -152,3 +212,9 @@ $ python scripts/send_request.py
 
 Tests require a running minio and rabbitmq container, meaning you have to run `docker compose up` in the tests folder
 before running the tests.
+
+## OpenTelemetry Protobuf Dependency Hell
+
+**Note**: Status 2025/01/09: the currently used `opentelemetry-exporter-otlp-proto-http` version `1.25.0` requires
+a `protobuf` version < `5.x.x` and is not compatible with the latest protobuf version `5.27.x`. This is an [open issue](https://github.com/open-telemetry/opentelemetry-python/issues/3958) in opentelemetry, because [support for 4.25.x ends in Q2 '25](https://protobuf.dev/support/version-support/#python).
+Therefore, we should keep this in mind and update the dependency once opentelemetry includes support for `protobuf 5.27.x`.

pyinfra/config/loader.py

@@ -1,5 +1,6 @@
 import argparse
 import os
+from functools import partial
 from pathlib import Path
 from typing import Union
 
@@ -7,20 +8,28 @@ from dynaconf import Dynaconf, ValidationError, Validator
 from funcy import lflatten
 from kn_utils.logging import logger
 
+# This path is ment for testing purposes and convenience. It probably won't reflect the actual root path when pyinfra is
+# installed as a package, so don't use it in production code, but define your own root path as described in load config.
+local_pyinfra_root_path = Path(__file__).parents[2]
 
-def load_settings(settings_path: Union[str, Path] = None, validators: list[Validator] = None):
-    settings_path = Path(settings_path) if settings_path else None
-    validators = validators or get_all_validators()
 
-    if not settings_path:
-        logger.info("No settings path specified, only loading .env end ENVs.")
-        settings_files = []
-    elif os.path.isdir(settings_path):
-        logger.info(f"Settings path is a directory, loading all .toml files in the directory: {settings_path}")
-        settings_files = list(settings_path.glob("*.toml"))
-    else:
-        logger.info(f"Settings path is a file, loading only the specified file: {settings_path}")
-        settings_files = [settings_path]
+def load_settings(
+    settings_path: Union[str, Path, list] = "config/",
+    root_path: Union[str, Path] = None,
+    validators: list[Validator] = None,
+):
+    """Load settings from .toml files, .env and environment variables. Also ensures a ROOT_PATH environment variable is
+    set. If ROOT_PATH is not set and no root_path argument is passed, the current working directory is used as root.
+    Settings paths can be a single .toml file, a folder containing .toml files or a list of .toml files and folders.
+    If a ROOT_PATH environment variable is set, it is not overwritten by the root_path argument.
+    If a folder is passed, all .toml files in the folder are loaded. If settings path is None, only .env and
+    environment variables are loaded. If settings_path are relative paths, they are joined with the root_path argument.
+    """
+
+    root_path = get_or_set_root_path(root_path)
+    validators = validators or get_pyinfra_validators()
+
+    settings_files = normalize_to_settings_files(settings_path, root_path)
 
     settings = Dynaconf(
         load_dotenv=True,
@@ -34,10 +43,63 @@ def load_settings(settings_path: Union[str, Path] = None, validators: list[Valid
     return settings
 
 
-pyinfra_config_path = Path(__file__).resolve().parents[2] / "config/"
+def normalize_to_settings_files(settings_path: Union[str, Path, list], root_path: Union[str, Path]):
+    if settings_path is None:
+        logger.info("No settings path specified, only loading .env end ENVs.")
+        settings_files = []
+    elif isinstance(settings_path, str) or isinstance(settings_path, Path):
+        settings_files = [settings_path]
+    elif isinstance(settings_path, list):
+        settings_files = settings_path
+    else:
+        raise ValueError(f"Invalid settings path: {settings_path=}")
+
+    settings_files = lflatten(map(partial(_normalize_and_verify, root_path=root_path), settings_files))
+    logger.debug(f"Normalized settings files: {settings_files}")
+
+    return settings_files
 
 
-def get_all_validators():
+def _normalize_and_verify(settings_path: Path, root_path: Path):
+    settings_path = Path(settings_path)
+    root_path = Path(root_path)
+
+    if not settings_path.is_absolute():
+        logger.debug(f"Settings path is not absolute, joining with root path: {root_path}")
+        settings_path = root_path / settings_path
+
+    if settings_path.is_dir():
+        logger.debug(f"Settings path is a directory, loading all .toml files in the directory: {settings_path}")
+        settings_files = list(settings_path.glob("*.toml"))
+    elif settings_path.is_file():
+        logger.debug(f"Settings path is a file, loading specified file: {settings_path}")
+        settings_files = [settings_path]
+    else:
+        raise ValueError(f"Invalid settings path: {settings_path=}, {root_path=}")
+
+    return settings_files
+
+
+def get_or_set_root_path(root_path: Union[str, Path] = None):
+    env_root_path = os.environ.get("ROOT_PATH")
+
+    if env_root_path:
+        root_path = env_root_path
+        logger.debug(f"'ROOT_PATH' environment variable is set to {root_path}.")
+
+    elif root_path:
+        logger.info(f"'ROOT_PATH' environment variable is not set, setting to {root_path}.")
+        os.environ["ROOT_PATH"] = str(root_path)
+
+    else:
+        root_path = Path.cwd()
+        logger.info(f"'ROOT_PATH' environment variable is not set, defaulting to working directory {root_path}.")
+        os.environ["ROOT_PATH"] = str(root_path)
+
+    return root_path
+
+
+def get_pyinfra_validators():
     import pyinfra.config.validators
 
     return lflatten(
@@ -61,13 +123,11 @@ def validate_settings(settings: Dynaconf, validators):
     logger.debug("Settings validated.")
 
 
-def parse_args():
+def parse_settings_path():
     parser = argparse.ArgumentParser()
     parser.add_argument(
-        "--settings_path",
-        "-s",
-        type=Path,
-        default=pyinfra_config_path,
-        help="Path to settings file or folder. Must be a .toml file or a folder containing .toml files.",
+        "settings_path",
+        help="Path to settings file(s) or folder(s). Must be .toml file(s) or a folder(s) containing .toml files.",
+        nargs="+",
     )
-    return parser.parse_args()
+    return parser.parse_args().settings_path

pyinfra/config/validators.py

@@ -44,3 +44,14 @@ webserver_validators = [
     Validator("webserver.host", must_exist=True, is_type_of=str),
     Validator("webserver.port", must_exist=True, is_type_of=int),
 ]
+
+tracing_validators = [
+    Validator("tracing.enabled", must_exist=True, is_type_of=bool),
+    Validator("tracing.type", must_exist=True, is_type_of=str)
+]
+
+opentelemetry_validators = [
+    Validator("tracing.opentelemetry.endpoint", must_exist=True, is_type_of=str),
+    Validator("tracing.opentelemetry.service_name", must_exist=True, is_type_of=str),
+    Validator("tracing.opentelemetry.exporter", must_exist=True, is_type_of=str)
+]

pyinfra/examples.py

@@ -1,38 +1,169 @@
+import asyncio
+import signal
+import sys
+
+import aiohttp
+from aiormq.exceptions import AMQPConnectionError
 from dynaconf import Dynaconf
 from fastapi import FastAPI
 from kn_utils.logging import logger
 
-from pyinfra.queue.callback import make_download_process_upload_callback, DataProcessor
+from pyinfra.config.loader import get_pyinfra_validators, validate_settings
+from pyinfra.queue.async_manager import AsyncQueueManager, RabbitMQConfig
+from pyinfra.queue.callback import Callback
 from pyinfra.queue.manager import QueueManager
-from pyinfra.webserver.prometheus import add_prometheus_endpoint, \
-    make_prometheus_processing_time_decorator_from_settings
-from pyinfra.webserver.utils import add_health_check_endpoint, create_webserver_thread_from_settings
+from pyinfra.utils.opentelemetry import instrument_app, instrument_pika, setup_trace
+from pyinfra.webserver.prometheus import (
+    add_prometheus_endpoint,
+    make_prometheus_processing_time_decorator_from_settings,
+)
+from pyinfra.webserver.utils import (
+    add_health_check_endpoint,
+    create_webserver_thread_from_settings,
+    run_async_webserver,
+)
+
+shutdown_flag = False
 
 
-def start_queue_consumer_with_prometheus_and_health_endpoints(process_fn: DataProcessor, settings: Dynaconf):
+async def graceful_shutdown(manager: AsyncQueueManager, queue_task, webserver_task):
+    global shutdown_flag
+    shutdown_flag = True
+    logger.info("SIGTERM received, shutting down gracefully...")
+
+    if queue_task and not queue_task.done():
+        queue_task.cancel()
+
+    # await queue manager shutdown
+    await asyncio.gather(queue_task, manager.shutdown(), return_exceptions=True)
+
+    if webserver_task and not webserver_task.done():
+        webserver_task.cancel()
+
+    # await webserver shutdown
+    await asyncio.gather(webserver_task, return_exceptions=True)
+
+    logger.info("Shutdown complete.")
+
+
+async def run_async_queues(manager: AsyncQueueManager, app, port, host):
+    """Run the async webserver and the async queue manager concurrently."""
+    queue_task = None
+    webserver_task = None
+    tenant_api_available = True
+
+    # add signal handler for SIGTERM and SIGINT
+    loop = asyncio.get_running_loop()
+    loop.add_signal_handler(
+        signal.SIGTERM, lambda: asyncio.create_task(graceful_shutdown(manager, queue_task, webserver_task))
+    )
+    loop.add_signal_handler(
+        signal.SIGINT, lambda: asyncio.create_task(graceful_shutdown(manager, queue_task, webserver_task))
+    )
+
+    try:
+        active_tenants = await manager.fetch_active_tenants()
+
+        queue_task = asyncio.create_task(manager.run(active_tenants=active_tenants), name="queues")
+        webserver_task = asyncio.create_task(run_async_webserver(app, port, host), name="webserver")
+        await asyncio.gather(queue_task, webserver_task)
+    except asyncio.CancelledError:
+        logger.info("Main task was cancelled, initiating shutdown.")
+    except AMQPConnectionError as e:
+        logger.warning(f"AMQPConnectionError: {e} - shutting down.")
+    except (aiohttp.ClientResponseError, aiohttp.ClientConnectorError):
+        logger.warning("Tenant server did not answer - shutting down.")
+        tenant_api_available = False
+    except Exception as e:
+        logger.error(f"An error occurred while running async queues: {e}", exc_info=True)
+        sys.exit(1)
+    finally:
+        if shutdown_flag:
+            logger.debug("Graceful shutdown already in progress.")
+        else:
+            logger.warning("Initiating shutdown due to error or manual interruption.")
+            if not tenant_api_available:
+                sys.exit(0)
+            if queue_task and not queue_task.done():
+                queue_task.cancel()
+
+            if webserver_task and not webserver_task.done():
+                webserver_task.cancel()
+
+            await asyncio.gather(queue_task, manager.shutdown(), webserver_task, return_exceptions=True)
+            logger.info("Shutdown complete.")
+
+
+def start_standard_queue_consumer(
+    callback: Callback,
+    settings: Dynaconf,
+    app: FastAPI = None,
+):
     """Default serving logic for research services.
 
-    Supplies /health, /ready and /prometheus endpoints. The process_fn is monitored for processing time per call.
-    Workload is only received via queue messages. The message contains a file path to the data to be processed, which
-    gets downloaded from the storage. The data and the message are then passed to the process_fn. The process_fn should
-    return a json serializable object. This object is then uploaded to the storage. The response message is just the
-    original message.
-
-    Adapt as needed.
+    Supplies /health, /ready and /prometheus endpoints (if enabled). The callback is monitored for processing time per
+    message. Also traces the queue messages via openTelemetry (if enabled).
+    Workload is received via queue messages and processed by the callback function (see pyinfra.queue.callback for
+    callbacks).
     """
-    logger.info(f"Starting webserver and queue consumer...")
+    validate_settings(settings, get_pyinfra_validators())
 
-    app = FastAPI()
+    logger.info("Starting webserver and queue consumer...")
 
-    app = add_prometheus_endpoint(app)
-    process_fn = make_prometheus_processing_time_decorator_from_settings(settings)(process_fn)
+    app = app or FastAPI()
 
-    queue_manager = QueueManager(settings)
+    if settings.metrics.prometheus.enabled:
+        logger.info("Prometheus metrics enabled.")
+        app = add_prometheus_endpoint(app)
+        callback = make_prometheus_processing_time_decorator_from_settings(settings)(callback)
 
-    app = add_health_check_endpoint(app, queue_manager.is_ready)
+    if settings.tracing.enabled:
+        setup_trace(settings)
 
-    webserver_thread = create_webserver_thread_from_settings(app, settings)
-    webserver_thread.start()
+        instrument_pika(dynamic_queues=settings.dynamic_tenant_queues.enabled)
+        instrument_app(app)
 
-    callback = make_download_process_upload_callback(process_fn, settings)
-    queue_manager.start_consuming(callback)
+    if settings.dynamic_tenant_queues.enabled:
+        logger.info("Dynamic tenant queues enabled. Running async queues.")
+        config = RabbitMQConfig(
+            host=settings.rabbitmq.host,
+            port=settings.rabbitmq.port,
+            username=settings.rabbitmq.username,
+            password=settings.rabbitmq.password,
+            heartbeat=settings.rabbitmq.heartbeat,
+            input_queue_prefix=settings.rabbitmq.service_request_queue_prefix,
+            tenant_event_queue_suffix=settings.rabbitmq.tenant_event_queue_suffix,
+            tenant_exchange_name=settings.rabbitmq.tenant_exchange_name,
+            service_request_exchange_name=settings.rabbitmq.service_request_exchange_name,
+            service_response_exchange_name=settings.rabbitmq.service_response_exchange_name,
+            service_dead_letter_queue_name=settings.rabbitmq.service_dlq_name,
+            queue_expiration_time=settings.rabbitmq.queue_expiration_time,
+            pod_name=settings.kubernetes.pod_name,
+        )
+        manager = AsyncQueueManager(
+            config=config,
+            tenant_service_url=settings.storage.tenant_server.endpoint,
+            message_processor=callback,
+            max_concurrent_tasks=(
+                settings.asyncio.max_concurrent_tasks if hasattr(settings.asyncio, "max_concurrent_tasks") else 10
+            ),
+        )
+    else:
+        logger.info("Dynamic tenant queues disabled. Running sync queues.")
+        manager = QueueManager(settings)
+
+    app = add_health_check_endpoint(app, manager.is_ready)
+
+    if isinstance(manager, AsyncQueueManager):
+        asyncio.run(run_async_queues(manager, app, port=settings.webserver.port, host=settings.webserver.host))
+
+    elif isinstance(manager, QueueManager):
+        webserver = create_webserver_thread_from_settings(app, settings)
+        webserver.start()
+        try:
+            manager.start_consuming(callback)
+        except Exception as e:
+            logger.error(f"An error occurred while consuming messages: {e}", exc_info=True)
+            sys.exit(1)
+    else:
+        logger.warning(f"Behavior for type {type(manager)} is not defined")

pyinfra/queue/async_manager.py

@@ -0,0 +1,329 @@
+import asyncio
+import concurrent.futures
+import json
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, Set
+
+import aiohttp
+from aio_pika import ExchangeType, IncomingMessage, Message, connect
+from aio_pika.abc import (
+    AbstractChannel,
+    AbstractConnection,
+    AbstractExchange,
+    AbstractIncomingMessage,
+    AbstractQueue,
+)
+from aio_pika.exceptions import (
+    ChannelClosed,
+    ChannelInvalidStateError,
+    ConnectionClosed,
+)
+from aiormq.exceptions import AMQPConnectionError
+from kn_utils.logging import logger
+from kn_utils.retry import retry
+
+
+@dataclass
+class RabbitMQConfig:
+    host: str
+    port: int
+    username: str
+    password: str
+    heartbeat: int
+    input_queue_prefix: str
+    tenant_event_queue_suffix: str
+    tenant_exchange_name: str
+    service_request_exchange_name: str
+    service_response_exchange_name: str
+    service_dead_letter_queue_name: str
+    queue_expiration_time: int
+    pod_name: str
+
+    connection_params: Dict[str, object] = field(init=False)
+
+    def __post_init__(self):
+        self.connection_params = {
+            "host": self.host,
+            "port": self.port,
+            "login": self.username,
+            "password": self.password,
+            "client_properties": {"heartbeat": self.heartbeat},
+        }
+
+
+class AsyncQueueManager:
+    def __init__(
+        self,
+        config: RabbitMQConfig,
+        tenant_service_url: str,
+        message_processor: Callable[[Dict[str, Any]], Dict[str, Any]],
+        max_concurrent_tasks: int = 10,
+    ):
+        self.config = config
+        self.tenant_service_url = tenant_service_url
+        self.message_processor = message_processor
+        self.semaphore = asyncio.Semaphore(max_concurrent_tasks)
+
+        self.connection: AbstractConnection | None = None
+        self.channel: AbstractChannel | None = None
+        self.tenant_exchange: AbstractExchange | None = None
+        self.input_exchange: AbstractExchange | None = None
+        self.output_exchange: AbstractExchange | None = None
+        self.tenant_exchange_queue: AbstractQueue | None = None
+        self.tenant_queues: Dict[str, AbstractChannel] = {}
+        self.consumer_tags: Dict[str, str] = {}
+
+        self.message_count: int = 0
+
+    @retry(tries=5, exceptions=AMQPConnectionError, reraise=True, logger=logger)
+    async def connect(self) -> None:
+        logger.info("Attempting to connect to RabbitMQ...")
+        self.connection = await connect(**self.config.connection_params)
+        self.connection.close_callbacks.add(self.on_connection_close)
+        self.channel = await self.connection.channel()
+        await self.channel.set_qos(prefetch_count=1)
+        logger.info("Successfully connected to RabbitMQ")
+
+    async def on_connection_close(self, sender, exc):
+        """This is a callback for unexpected connection closures."""
+        logger.debug(f"Sender: {sender}")
+        if isinstance(exc, ConnectionClosed):
+            logger.warning("Connection to RabbitMQ lost. Attempting to reconnect...")
+            try:
+                active_tenants = await self.fetch_active_tenants()
+                await self.run(active_tenants=active_tenants)
+                logger.debug("Reconnected to RabbitMQ successfully")
+            except Exception as e:
+                logger.warning(f"Failed to reconnect to RabbitMQ: {e}")
+                # cancel queue manager and webserver to shutdown service
+                tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
+                [task.cancel() for task in tasks if task.get_name() in ["queues", "webserver"]]
+        else:
+            logger.debug("Connection closed on purpose.")
+
+    async def is_ready(self) -> bool:
+        if self.connection is None or self.connection.is_closed:
+            try:
+                await self.connect()
+            except Exception as e:
+                logger.error(f"Failed to connect to RabbitMQ: {e}")
+                return False
+        return True
+
+    @retry(tries=5, exceptions=(AMQPConnectionError, ChannelInvalidStateError), reraise=True, logger=logger)
+    async def setup_exchanges(self) -> None:
+        self.tenant_exchange = await self.channel.declare_exchange(
+            self.config.tenant_exchange_name, ExchangeType.TOPIC, durable=True
+        )
+        self.input_exchange = await self.channel.declare_exchange(
+            self.config.service_request_exchange_name, ExchangeType.DIRECT, durable=True
+        )
+        self.output_exchange = await self.channel.declare_exchange(
+            self.config.service_response_exchange_name, ExchangeType.DIRECT, durable=True
+        )
+
+        # we must declare DLQ to handle error messages
+        self.dead_letter_queue = await self.channel.declare_queue(
+            self.config.service_dead_letter_queue_name, durable=True
+        )
+
+    @retry(tries=5, exceptions=(AMQPConnectionError, ChannelInvalidStateError), reraise=True, logger=logger)
+    async def setup_tenant_queue(self) -> None:
+        self.tenant_exchange_queue = await self.channel.declare_queue(
+            f"{self.config.pod_name}_{self.config.tenant_event_queue_suffix}",
+            durable=True,
+            arguments={
+                "x-dead-letter-exchange": "",
+                "x-dead-letter-routing-key": self.config.service_dead_letter_queue_name,
+                "x-expires": self.config.queue_expiration_time,
+            },
+        )
+        await self.tenant_exchange_queue.bind(self.tenant_exchange, routing_key="tenant.*")
+        self.consumer_tags["tenant_exchange_queue"] = await self.tenant_exchange_queue.consume(
+            self.process_tenant_message
+        )
+
+    async def process_tenant_message(self, message: AbstractIncomingMessage) -> None:
+        try:
+            async with message.process():
+                message_body = json.loads(message.body.decode())
+                logger.debug(f"Tenant message received: {message_body}")
+                tenant_id = message_body["tenantId"]
+                routing_key = message.routing_key
+
+                if routing_key == "tenant.created":
+                    await self.create_tenant_queues(tenant_id)
+                elif routing_key == "tenant.delete":
+                    await self.delete_tenant_queues(tenant_id)
+        except Exception as e:
+            logger.error(e, exc_info=True)
+
+    async def create_tenant_queues(self, tenant_id: str) -> None:
+        queue_name = f"{self.config.input_queue_prefix}_{tenant_id}"
+        logger.info(f"Declaring queue: {queue_name}")
+        try:
+            input_queue = await self.channel.declare_queue(
+                queue_name,
+                durable=True,
+                arguments={
+                    "x-dead-letter-exchange": "",
+                    "x-dead-letter-routing-key": self.config.service_dead_letter_queue_name,
+                },
+            )
+            await input_queue.bind(self.input_exchange, routing_key=tenant_id)
+            self.consumer_tags[tenant_id] = await input_queue.consume(self.process_input_message)
+            self.tenant_queues[tenant_id] = input_queue
+            logger.info(f"Created and started consuming queue for tenant {tenant_id}")
+        except Exception as e:
+            logger.error(e, exc_info=True)
+
+    async def delete_tenant_queues(self, tenant_id: str) -> None:
+        if tenant_id in self.tenant_queues:
+            # somehow queue.delete() does not work here
+            await self.channel.queue_delete(f"{self.config.input_queue_prefix}_{tenant_id}")
+            del self.tenant_queues[tenant_id]
+            del self.consumer_tags[tenant_id]
+            logger.info(f"Deleted queues for tenant {tenant_id}")
+
+    async def process_input_message(self, message: IncomingMessage) -> None:
+        async def process_message_body_and_await_result(unpacked_message_body):
+            async with self.semaphore:
+                loop = asyncio.get_running_loop()
+                with concurrent.futures.ThreadPoolExecutor(max_workers=1) as thread_pool_executor:
+                    logger.info("Processing payload in a separate thread.")
+                    result = await loop.run_in_executor(
+                        thread_pool_executor, self.message_processor, unpacked_message_body
+                    )
+                    return result
+
+        async with message.process(ignore_processed=True):
+            if message.redelivered:
+                logger.warning(f"Declining message with {message.delivery_tag=} due to it being redelivered.")
+                await message.nack(requeue=False)
+                return
+
+            if message.body.decode("utf-8") == "STOP":
+                logger.info("Received stop signal, stopping consumption...")
+                await message.ack()
+                # TODO: shutdown is probably not the right call here - align w/ Dev what should happen on stop signal
+                await self.shutdown()
+                return
+
+            self.message_count += 1
+
+            try:
+                tenant_id = message.routing_key
+
+                filtered_message_headers = (
+                    {k: v for k, v in message.headers.items() if k.lower().startswith("x-")} if message.headers else {}
+                )
+
+                logger.debug(f"Processing message with {filtered_message_headers=}.")
+
+                result: dict = await (
+                    process_message_body_and_await_result({**json.loads(message.body), **filtered_message_headers})
+                    or {}
+                )
+
+                if result:
+                    await self.publish_to_output_exchange(tenant_id, result, filtered_message_headers)
+                    await message.ack()
+                    logger.debug(f"Message with {message.delivery_tag=} acknowledged.")
+                else:
+                    raise ValueError(f"Could not process message with {message.body=}.")
+
+            except json.JSONDecodeError:
+                await message.nack(requeue=False)
+                logger.error(f"Invalid JSON in input message: {message.body}", exc_info=True)
+            except FileNotFoundError as e:
+                logger.warning(f"{e}, declining message with {message.delivery_tag=}.", exc_info=True)
+                await message.nack(requeue=False)
+            except Exception as e:
+                await message.nack(requeue=False)
+                logger.error(f"Error processing input message: {e}", exc_info=True)
+            finally:
+                self.message_count -= 1
+
+    async def publish_to_output_exchange(self, tenant_id: str, result: Dict[str, Any], headers: Dict[str, Any]) -> None:
+        await self.output_exchange.publish(
+            Message(body=json.dumps(result).encode(), headers=headers),
+            routing_key=tenant_id,
+        )
+        logger.info(f"Published result to queue {tenant_id}.")
+
+    @retry(tries=5, exceptions=(aiohttp.ClientResponseError, aiohttp.ClientConnectorError), reraise=True, logger=logger)
+    async def fetch_active_tenants(self) -> Set[str]:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(self.tenant_service_url) as response:
+                response.raise_for_status()
+                if response.headers["content-type"].lower() == "application/json":
+                    data = await response.json()
+                    return {tenant["tenantId"] for tenant in data}
+                else:
+                    logger.error(
+                        f"Failed to fetch active tenants. Content type is not JSON: {response.headers['content-type'].lower()}"
+                    )
+                    return set()
+
+    @retry(
+        tries=5,
+        exceptions=(
+            AMQPConnectionError,
+            ChannelInvalidStateError,
+        ),
+        reraise=True,
+        logger=logger,
+    )
+    async def initialize_tenant_queues(self, active_tenants: set) -> None:
+        for tenant_id in active_tenants:
+            await self.create_tenant_queues(tenant_id)
+
+    async def run(self, active_tenants: set) -> None:
+
+        await self.connect()
+        await self.setup_exchanges()
+        await self.initialize_tenant_queues(active_tenants=active_tenants)
+        await self.setup_tenant_queue()
+
+        logger.info("RabbitMQ handler is running. Press CTRL+C to exit.")
+
+    async def close_channels(self) -> None:
+        try:
+            if self.channel and not self.channel.is_closed:
+                # Cancel queues to stop fetching messages
+                logger.debug("Cancelling queues...")
+                for tenant, queue in self.tenant_queues.items():
+                    await queue.cancel(self.consumer_tags[tenant])
+                if self.tenant_exchange_queue:
+                    await self.tenant_exchange_queue.cancel(self.consumer_tags["tenant_exchange_queue"])
+                while self.message_count != 0:
+                    logger.debug(f"Messages are still being processed: {self.message_count=} ")
+                    await asyncio.sleep(2)
+                await self.channel.close(exc=asyncio.CancelledError)
+                logger.debug("Channel closed.")
+            else:
+                logger.debug("No channel to close.")
+        except ChannelClosed:
+            logger.warning("Channel was already closed.")
+        except ConnectionClosed:
+            logger.warning("Connection was lost, unable to close channel.")
+        except Exception as e:
+            logger.error(f"Error during channel shutdown: {e}")
+
+    async def close_connection(self) -> None:
+        try:
+            if self.connection and not self.connection.is_closed:
+                await self.connection.close(exc=asyncio.CancelledError)
+                logger.debug("Connection closed.")
+            else:
+                logger.debug("No connection to close.")
+        except ConnectionClosed:
+            logger.warning("Connection was already closed.")
+        except Exception as e:
+            logger.error(f"Error closing connection: {e}")
+
+    async def shutdown(self) -> None:
+        logger.info("Shutting down RabbitMQ handler...")
+        await self.close_channels()
+        await self.close_connection()
+        logger.info("RabbitMQ handler shut down successfully.")

pyinfra/queue/callback.py

@@ -1,23 +1,27 @@
-from typing import Callable, Union
+from typing import Callable
 
 from dynaconf import Dynaconf
 from kn_utils.logging import logger
 
 from pyinfra.storage.connection import get_storage
-from pyinfra.storage.utils import download_data_as_specified_in_message, upload_data_as_specified_in_message
+from pyinfra.storage.utils import (
+    download_data_bytes_as_specified_in_message,
+    upload_data_as_specified_in_message,
+    DownloadedData,
+)
 
-DataProcessor = Callable[[Union[dict, bytes], dict], dict]
+DataProcessor = Callable[[dict[str, DownloadedData] | DownloadedData, dict], dict | list | str]
+Callback = Callable[[dict], dict]
 
 
-def make_download_process_upload_callback(data_processor: DataProcessor, settings: Dynaconf):
+def make_download_process_upload_callback(data_processor: DataProcessor, settings: Dynaconf) -> Callback:
     """Default callback for processing queue messages.
+
     Data will be downloaded from the storage as specified in the message. If a tenant id is specified, the storage
     will be configured to use that tenant id, otherwise the storage is configured as specified in the settings.
     The data is the passed to the dataprocessor, together with the message. The dataprocessor should return a
-    json serializable object. This object is then uploaded to the storage as specified in the message.
-
-    The response message is just the original message.
-    Adapt as needed.
+    json serializable object. This object is then uploaded to the storage as specified in the message. The response
+    message is just the original message.
     """
 
     def inner(queue_message_payload: dict) -> dict:
@@ -25,7 +29,9 @@ def make_download_process_upload_callback(data_processor: DataProcessor, setting
 
         storage = get_storage(settings, queue_message_payload.get("X-TENANT-ID"))
 
-        data = download_data_as_specified_in_message(storage, queue_message_payload)
+        data: dict[str, DownloadedData] | DownloadedData = download_data_bytes_as_specified_in_message(
+            storage, queue_message_payload
+        )
 
         result = data_processor(data, queue_message_payload)
 

pyinfra/queue/manager.py

@@ -4,17 +4,17 @@ import json
 import logging
 import signal
 import sys
-from typing import Union, Callable
+from typing import Callable, Union
 
 import pika
 import pika.exceptions
 from dynaconf import Dynaconf
 from kn_utils.logging import logger
+from kn_utils.retry import retry
 from pika.adapters.blocking_connection import BlockingChannel, BlockingConnection
-from retry import retry
 
-from pyinfra.config.validators import queue_manager_validators
 from pyinfra.config.loader import validate_settings
+from pyinfra.config.validators import queue_manager_validators
 
 pika_logger = logging.getLogger("pika")
 pika_logger.setLevel(logging.WARNING)  # disables non-informative pika log clutter
@@ -35,11 +35,16 @@ class QueueManager:
         self.connection: Union[BlockingConnection, None] = None
         self.channel: Union[BlockingChannel, None] = None
         self.connection_sleep = settings.rabbitmq.connection_sleep
+        self.processing_callback = False
+        self.received_signal = False
 
         atexit.register(self.stop_consuming)
         signal.signal(signal.SIGTERM, self._handle_stop_signal)
         signal.signal(signal.SIGINT, self._handle_stop_signal)
 
+        self.max_retries = settings.rabbitmq.max_retries or 5
+        self.max_delay = settings.rabbitmq.max_delay or 60
+
     @staticmethod
     def create_connection_parameters(settings: Dynaconf):
         credentials = pika.PlainCredentials(username=settings.rabbitmq.username, password=settings.rabbitmq.password)
@@ -52,9 +57,12 @@ class QueueManager:
 
         return pika.ConnectionParameters(**pika_connection_params)
 
-    @retry(tries=3, delay=5, jitter=(1, 3), logger=logger)
+    @retry(
+        tries=5,
+        exceptions=(pika.exceptions.AMQPConnectionError, pika.exceptions.ChannelClosedByBroker),
+        reraise=True,
+    )
     def establish_connection(self):
-        # TODO: set sensible retry parameters
         if self.connection and self.connection.is_open:
             logger.debug("Connection to RabbitMQ already established.")
             return
@@ -77,19 +85,31 @@ class QueueManager:
         logger.info("Connection to RabbitMQ established, channel open.")
 
     def is_ready(self):
-        self.establish_connection()
-        return self.channel.is_open
+        try:
+            self.establish_connection()
+            return self.channel.is_open
+        except Exception as e:
+            logger.error(f"Failed to establish connection: {e}")
+            return False
 
-    @retry(exceptions=pika.exceptions.AMQPConnectionError, tries=3, delay=5, jitter=(1, 3), logger=logger)
+    @retry(
+        tries=5,
+        exceptions=pika.exceptions.AMQPConnectionError,
+        reraise=True,
+    )
     def start_consuming(self, message_processor: Callable):
         on_message_callback = self._make_on_message_callback(message_processor)
 
         try:
             self.establish_connection()
             self.channel.basic_consume(self.input_queue, on_message_callback)
+            logger.info("Starting to consume messages...")
             self.channel.start_consuming()
-        except Exception:
-            logger.error("An unexpected error occurred while consuming messages. Consuming will stop.", exc_info=True)
+        except pika.exceptions.AMQPConnectionError as e:
+            logger.error(f"AMQP Connection Error: {e}")
+            raise
+        except Exception as e:
+            logger.error(f"An unexpected error occurred while consuming messages: {e}", exc_info=True)
             raise
         finally:
             self.stop_consuming()
@@ -141,17 +161,17 @@ class QueueManager:
                 logger.info("Processing payload in separate thread.")
                 future = thread_pool_executor.submit(message_processor, unpacked_message_body)
 
-                # FIXME: This block is probably not necessary, but kept since the implications of removing it are
+                # TODO: This block is probably not necessary, but kept since the implications of removing it are
                 #  unclear. Remove it in a future iteration where less changes are being made to the code base.
                 while future.running():
                     logger.debug("Waiting for payload processing to finish...")
-                    self.connection.process_data_events()
                     self.connection.sleep(self.connection_sleep)
 
                 return future.result()
 
         def on_message_callback(channel, method, properties, body):
             logger.info(f"Received message from queue with delivery_tag {method.delivery_tag}.")
+            self.processing_callback = True
 
             if method.redelivered:
                 logger.warning(f"Declining message with {method.delivery_tag=} due to it being redelivered.")
@@ -185,14 +205,25 @@ class QueueManager:
 
                 channel.basic_ack(delivery_tag=method.delivery_tag)
                 logger.debug(f"Message with {method.delivery_tag=} acknowledged.")
+            except FileNotFoundError as e:
+                logger.warning(f"{e}, declining message with {method.delivery_tag=}.")
+                channel.basic_nack(method.delivery_tag, requeue=False)
             except Exception:
                 logger.warning(f"Failed to process message with {method.delivery_tag=}, declining...", exc_info=True)
                 channel.basic_nack(method.delivery_tag, requeue=False)
                 raise
 
+            finally:
+                self.processing_callback = False
+                if self.received_signal:
+                    self.stop_consuming()
+                    sys.exit(0)
+
         return on_message_callback
 
     def _handle_stop_signal(self, signum, *args, **kwargs):
         logger.info(f"Received signal {signum}, stopping consuming...")
-        self.stop_consuming()
-        sys.exit(0)
+        self.received_signal = True
+        if not self.processing_callback:
+            self.stop_consuming()
+            sys.exit(0)

pyinfra/storage/connection.py

@@ -4,92 +4,86 @@ import requests
 from dynaconf import Dynaconf
 from kn_utils.logging import logger
 
+from pyinfra.config.loader import validate_settings
+from pyinfra.config.validators import (
+    multi_tenant_storage_validators,
+    storage_validators,
+)
 from pyinfra.storage.storages.azure import get_azure_storage_from_settings
 from pyinfra.storage.storages.s3 import get_s3_storage_from_settings
 from pyinfra.storage.storages.storage import Storage
 from pyinfra.utils.cipher import decrypt
-from pyinfra.config.validators import storage_validators, multi_tenant_storage_validators
-from pyinfra.config.loader import validate_settings
 
 
 def get_storage(settings: Dynaconf, tenant_id: str = None) -> Storage:
-    """Get storage connection based on settings.
-    If tenant_id is provided, gets storage connection information from tenant server instead.
-    The connections are cached based on the settings.cache_size value.
-
-    In the future, when the default storage from config is no longer needed (only multi-tenant storage will be used),
-    get_storage_from_tenant_id can replace this function directly.
+    """Establishes a storage connection.
+    If tenant_id is provided, gets storage connection information from tenant server. These connections are cached.
+    Otherwise, gets storage connection information from settings.
     """
     logger.info("Establishing storage connection...")
 
     if tenant_id:
         logger.info(f"Using tenant storage for {tenant_id}.")
-        return get_storage_from_tenant_id(tenant_id, settings)
-    else:
-        logger.info("Using default storage.")
-        return get_storage_from_settings(settings)
+        validate_settings(settings, multi_tenant_storage_validators)
 
+        return get_storage_for_tenant(
+            tenant_id,
+            settings.storage.tenant_server.endpoint,
+            settings.storage.tenant_server.public_key,
+        )
 
-def get_storage_from_settings(settings: Dynaconf) -> Storage:
+    logger.info("Using default storage.")
     validate_settings(settings, storage_validators)
 
-    @lru_cache(maxsize=settings.storage.cache_size)
-    def _get_storage(backend: str) -> Storage:
-        return storage_dispatcher[backend](settings)
-
-    return _get_storage(settings.storage.backend)
-
-
-def get_storage_from_tenant_id(tenant_id: str, settings: Dynaconf) -> Storage:
-    validate_settings(settings, multi_tenant_storage_validators)
-
-    @lru_cache(maxsize=settings.storage.cache_size)
-    def _get_storage(tenant: str, endpoint: str, public_key: str) -> Storage:
-        response = requests.get(f"{endpoint}/{tenant}").json()
-
-        maybe_azure = response.get("azureStorageConnection")
-        maybe_s3 = response.get("s3StorageConnection")
-        assert (maybe_azure or maybe_s3) and not (maybe_azure and maybe_s3), "Only one storage backend can be used."
-
-        if maybe_azure:
-            connection_string = decrypt(public_key, maybe_azure["connectionString"])
-            backend = "azure"
-            storage_info = {
-                "storage": {
-                    "azure": {
-                        "connection_string": connection_string,
-                        "container": maybe_azure["containerName"],
-                    },
-                }
-            }
-        elif maybe_s3:
-            secret = decrypt(public_key, maybe_s3["secret"])
-            backend = "s3"
-            storage_info = {
-                "storage": {
-                    "s3": {
-                        "endpoint": maybe_s3["endpoint"],
-                        "key": maybe_s3["key"],
-                        "secret": secret,
-                        "region": maybe_s3["region"],
-                        "bucket": maybe_s3["bucketName"],
-                    },
-                }
-            }
-        else:
-            raise Exception(f"Unknown storage backend in {response}.")
-
-        storage_settings = Dynaconf()
-        storage_settings.update(storage_info)
-
-        storage = storage_dispatcher[backend](storage_settings)
-
-        return storage
-
-    return _get_storage(tenant_id, settings.storage.tenant_server.endpoint, settings.storage.tenant_server.public_key)
+    return storage_dispatcher[settings.storage.backend](settings)
 
 
 storage_dispatcher = {
     "azure": get_azure_storage_from_settings,
     "s3": get_s3_storage_from_settings,
 }
+
+
+@lru_cache(maxsize=10)
+def get_storage_for_tenant(tenant: str, endpoint: str, public_key: str) -> Storage:
+    response = requests.get(f"{endpoint}/{tenant}").json()
+
+    maybe_azure = response.get("azureStorageConnection")
+    maybe_s3 = response.get("s3StorageConnection")
+
+    assert (maybe_azure or maybe_s3) and not (maybe_azure and maybe_s3), "Only one storage backend can be used."
+
+    if maybe_azure:
+        connection_string = decrypt(public_key, maybe_azure["connectionString"])
+        backend = "azure"
+        storage_info = {
+            "storage": {
+                "azure": {
+                    "connection_string": connection_string,
+                    "container": maybe_azure["containerName"],
+                },
+            }
+        }
+    elif maybe_s3:
+        secret = decrypt(public_key, maybe_s3["secret"])
+        backend = "s3"
+        storage_info = {
+            "storage": {
+                "s3": {
+                    "endpoint": maybe_s3["endpoint"],
+                    "key": maybe_s3["key"],
+                    "secret": secret,
+                    "region": maybe_s3["region"],
+                    "bucket": maybe_s3["bucketName"],
+                },
+            }
+        }
+    else:
+        raise Exception(f"Unknown storage backend in {response}.")
+
+    storage_settings = Dynaconf()
+    storage_settings.update(storage_info)
+
+    storage = storage_dispatcher[backend](storage_settings)
+
+    return storage

pyinfra/storage/storages/azure.py

@@ -7,9 +7,9 @@ from dynaconf import Dynaconf
 from kn_utils.logging import logger
 from retry import retry
 
-from pyinfra.storage.storages.storage import Storage
-from pyinfra.config.validators import azure_storage_validators
 from pyinfra.config.loader import validate_settings
+from pyinfra.config.validators import azure_storage_validators
+from pyinfra.storage.storages.storage import Storage
 
 logging.getLogger("azure").setLevel(logging.WARNING)
 logging.getLogger("urllib3").setLevel(logging.WARNING)

pyinfra/storage/storages/s3.py

@@ -7,9 +7,9 @@ from kn_utils.logging import logger
 from minio import Minio
 from retry import retry
 
-from pyinfra.storage.storages.storage import Storage
-from pyinfra.config.validators import s3_storage_validators
 from pyinfra.config.loader import validate_settings
+from pyinfra.config.validators import s3_storage_validators
+from pyinfra.storage.storages.storage import Storage
 from pyinfra.utils.url_parsing import validate_and_parse_s3_endpoint
 
 

pyinfra/storage/utils.py

@@ -1,6 +1,7 @@
 import gzip
 import json
-from typing import Union
+from functools import singledispatch
+from typing import TypedDict
 
 from kn_utils.logging import logger
 from pydantic import BaseModel, ValidationError
@@ -18,6 +19,17 @@ class DossierIdFileIdDownloadPayload(BaseModel):
         return f"{self.dossierId}/{self.fileId}.{self.targetFileExtension}"
 
 
+class TenantIdDossierIdFileIdDownloadPayload(BaseModel):
+    tenantId: str
+    dossierId: str
+    fileId: str
+    targetFileExtension: str
+
+    @property
+    def targetFilePath(self):
+        return f"{self.tenantId}/{self.dossierId}/{self.fileId}.{self.targetFileExtension}"
+
+
 class DossierIdFileIdUploadPayload(BaseModel):
     dossierId: str
     fileId: str
@@ -28,50 +40,79 @@ class DossierIdFileIdUploadPayload(BaseModel):
         return f"{self.dossierId}/{self.fileId}.{self.responseFileExtension}"
 
 
+class TenantIdDossierIdFileIdUploadPayload(BaseModel):
+    tenantId: str
+    dossierId: str
+    fileId: str
+    responseFileExtension: str
+
+    @property
+    def responseFilePath(self):
+        return f"{self.tenantId}/{self.dossierId}/{self.fileId}.{self.responseFileExtension}"
+
+
 class TargetResponseFilePathDownloadPayload(BaseModel):
-    targetFilePath: str
+    targetFilePath: str | dict[str, str]
 
 
 class TargetResponseFilePathUploadPayload(BaseModel):
     responseFilePath: str
 
 
-def download_data_as_specified_in_message(storage: Storage, raw_payload: dict) -> Union[dict, bytes]:
+class DownloadedData(TypedDict):
+    data: bytes
+    file_path: str
+
+
+def download_data_bytes_as_specified_in_message(
+    storage: Storage, raw_payload: dict
+) -> dict[str, DownloadedData] | DownloadedData:
     """Convenience function to download a file specified in a message payload.
-    Supports both legacy and new payload formats.
-
-    If the content is compressed with gzip (.gz), it will be decompressed (-> bytes).
-    If the content is a json file, it will be decoded (-> dict).
-    If no file is specified in the payload or the file does not exist in storage, an exception will be raised.
-    In all other cases, the content will be returned as is (-> bytes).
-
-    This function can be extended in the future as needed (e.g. handling of more file types), but since further
-    requirements are not specified at this point in time, and it is unclear what these would entail, the code is kept
-    simple for now to improve readability, maintainability and avoid refactoring efforts of generic solutions that
-    weren't as generic as they seemed.
-
+    Supports both legacy and new payload formats. Also supports downloading multiple files at once, which should
+    be specified in a dictionary under the 'targetFilePath' key with the file path as value.
+    The data is downloaded as bytes and returned as a dictionary with the file path as key and the data as value.
+    In case of several download targets, a nested dictionary is returned with the same keys and dictionaries with
+    the file path and data as values.
     """
 
     try:
-        if "dossierId" in raw_payload:
+        if "tenantId" in raw_payload and "dossierId" in raw_payload:
+            payload = TenantIdDossierIdFileIdDownloadPayload(**raw_payload)
+        elif "tenantId" not in raw_payload and "dossierId" in raw_payload:
             payload = DossierIdFileIdDownloadPayload(**raw_payload)
         else:
             payload = TargetResponseFilePathDownloadPayload(**raw_payload)
     except ValidationError:
         raise ValueError("No download file path found in payload, nothing to download.")
 
-    if not storage.exists(payload.targetFilePath):
-        raise FileNotFoundError(f"File '{payload.targetFilePath}' does not exist in storage.")
-
-    data = storage.get_object(payload.targetFilePath)
-
-    data = gzip.decompress(data) if ".gz" in payload.targetFilePath else data
-    data = json.loads(data.decode("utf-8")) if ".json" in payload.targetFilePath else data
-    logger.info(f"Downloaded {payload.targetFilePath} from storage.")
+    data = _download(payload.targetFilePath, storage)
 
     return data
 
 
+@singledispatch
+def _download(
+    file_path_or_file_path_dict: str | dict[str, str], storage: Storage
+) -> dict[str, DownloadedData] | DownloadedData:
+    pass
+
+
+@_download.register(str)
+def _download_single_file(file_path: str, storage: Storage) -> DownloadedData:
+    if not storage.exists(file_path):
+        raise FileNotFoundError(f"File '{file_path}' does not exist in storage.")
+
+    data = storage.get_object(file_path)
+    logger.info(f"Downloaded {file_path} from storage.")
+
+    return DownloadedData(data=data, file_path=file_path)
+
+
+@_download.register(dict)
+def _download_multiple_files(file_path_dict: dict, storage: Storage) -> dict[str, DownloadedData]:
+    return {key: _download(value, storage) for key, value in file_path_dict.items()}
+
+
 def upload_data_as_specified_in_message(storage: Storage, raw_payload: dict, data):
     """Convenience function to upload a file specified in a message payload. For now, only json serializable data is
     supported. The storage json consists of the raw_payload, which is extended with a 'data' key, containing the
@@ -87,7 +128,9 @@ def upload_data_as_specified_in_message(storage: Storage, raw_payload: dict, dat
     """
 
     try:
-        if "dossierId" in raw_payload:
+        if "tenantId" in raw_payload and "dossierId" in raw_payload:
+            payload = TenantIdDossierIdFileIdUploadPayload(**raw_payload)
+        elif "tenantId" not in raw_payload and "dossierId" in raw_payload:
             payload = DossierIdFileIdUploadPayload(**raw_payload)
         else:
             payload = TargetResponseFilePathUploadPayload(**raw_payload)

pyinfra/utils/opentelemetry.py

@@ -0,0 +1,96 @@
+import json
+
+from azure.monitor.opentelemetry import configure_azure_monitor
+from dynaconf import Dynaconf
+from fastapi import FastAPI
+from kn_utils.logging import logger
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.instrumentation.aio_pika import AioPikaInstrumentor
+from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
+from opentelemetry.instrumentation.pika import PikaInstrumentor
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import (
+    BatchSpanProcessor,
+    ConsoleSpanExporter,
+    SpanExporter,
+    SpanExportResult,
+)
+
+from pyinfra.config.loader import validate_settings
+from pyinfra.config.validators import opentelemetry_validators
+
+
+class JsonSpanExporter(SpanExporter):
+    def __init__(self):
+        self.traces = []
+
+    def export(self, spans):
+        for span in spans:
+            self.traces.append(json.loads(span.to_json()))
+        return SpanExportResult.SUCCESS
+
+    def shutdown(self):
+        pass
+
+
+def setup_trace(settings: Dynaconf, service_name: str = None, exporter: SpanExporter = None):
+    tracing_type = settings.tracing.type
+    if tracing_type == "azure_monitor":
+        # Configure OpenTelemetry to use Azure Monitor with the
+        # APPLICATIONINSIGHTS_CONNECTION_STRING environment variable.
+        try:
+            configure_azure_monitor()
+            logger.info("Azure Monitor tracing enabled.")
+        except Exception as exception:
+            logger.warning(f"Azure Monitor tracing could not be enabled: {exception}")
+    elif tracing_type == "opentelemetry":
+        configure_opentelemtry_tracing(settings, service_name, exporter)
+        logger.info("OpenTelemetry tracing enabled.")
+    else:
+        logger.warning(f"Unknown tracing type: {tracing_type}. Tracing could not be enabled.")
+
+
+def configure_opentelemtry_tracing(settings: Dynaconf, service_name: str = None, exporter: SpanExporter = None):
+    service_name = service_name or settings.tracing.opentelemetry.service_name
+    exporter = exporter or get_exporter(settings)
+
+    resource = Resource(attributes={"service.name": service_name})
+    provider = TracerProvider(resource=resource, shutdown_on_exit=True)
+
+    processor = BatchSpanProcessor(exporter)
+    provider.add_span_processor(processor)
+
+    # TODO: trace.set_tracer_provider produces a warning if trying to set the provider twice.
+    #  "WARNING  opentelemetry.trace:__init__.py:521 Overriding of current TracerProvider is not allowed"
+    #  This doesn't seem to affect the functionality since we only want to use the tracer provided set in the beginning.
+    #  We work around the log message by using the protected method with log=False.
+    trace._set_tracer_provider(provider, log=False)
+
+
+def get_exporter(settings: Dynaconf):
+    validate_settings(settings, validators=opentelemetry_validators)
+
+    if settings.tracing.opentelemetry.exporter == "json":
+        return JsonSpanExporter()
+    elif settings.tracing.opentelemetry.exporter == "otlp":
+        return OTLPSpanExporter(endpoint=settings.tracing.opentelemetry.endpoint)
+    elif settings.tracing.opentelemetry.exporter == "console":
+        return ConsoleSpanExporter()
+    else:
+        raise ValueError(
+            f"Invalid OpenTelemetry exporter {settings.tracing.opentelemetry.exporter}. "
+            f"Valid values are 'json', 'otlp' and 'console'."
+        )
+
+
+def instrument_pika(dynamic_queues: bool):
+    if dynamic_queues:
+        AioPikaInstrumentor().instrument()
+    else:
+        PikaInstrumentor().instrument()
+
+
+def instrument_app(app: FastAPI, excluded_urls: str = "/health,/ready,/prometheus"):
+    FastAPIInstrumentor().instrument_app(app, excluded_urls=excluded_urls)

pyinfra/webserver/prometheus.py

@@ -4,11 +4,11 @@ from typing import Callable, TypeVar
 from dynaconf import Dynaconf
 from fastapi import FastAPI
 from funcy import identity
-from prometheus_client import generate_latest, CollectorRegistry, REGISTRY, Summary
+from prometheus_client import REGISTRY, CollectorRegistry, Summary, generate_latest
 from starlette.responses import Response
 
-from pyinfra.config.validators import prometheus_validators
 from pyinfra.config.loader import validate_settings
+from pyinfra.config.validators import prometheus_validators
 
 
 def add_prometheus_endpoint(app: FastAPI, registry: CollectorRegistry = REGISTRY) -> FastAPI:
@@ -36,16 +36,11 @@ def make_prometheus_processing_time_decorator_from_settings(
     postfix: str = "processing_time",
     registry: CollectorRegistry = REGISTRY,
 ) -> Decorator:
-    """Make a decorator for monitoring the processing time of a function. The decorator is only applied if the
-    prometheus metrics are enabled in the settings.
-    This, and other metrics should follow the convention
-        {product name}_{service name}_{processing step / parameter to monitor}.
+    """Make a decorator for monitoring the processing time of a function. This, and other metrics should follow the
+    convention {product name}_{service name}_{processing step / parameter to monitor}.
     """
     validate_settings(settings, validators=prometheus_validators)
 
-    if not settings.metrics.prometheus.enabled:
-        return identity
-
     processing_time_sum = Summary(
         f"{settings.metrics.prometheus.prefix}_{postfix}",
         "Summed up processing time per call.",

pyinfra/webserver/utils.py

@@ -1,18 +1,35 @@
+import asyncio
+import inspect
 import logging
+import signal
 import threading
+import time
 from typing import Callable
 
 import uvicorn
 from dynaconf import Dynaconf
 from fastapi import FastAPI
+from kn_utils.logging import logger
+from kn_utils.retry import retry
 
-from pyinfra.config.validators import webserver_validators
 from pyinfra.config.loader import validate_settings
+from pyinfra.config.validators import webserver_validators
 
 
+class PyInfraUvicornServer(uvicorn.Server):
+    # this is a workaround to enable custom signal handlers
+    # https://github.com/encode/uvicorn/issues/1579
+    def install_signal_handlers(self):
+        pass
+
+
+@retry(
+    tries=5,
+    exceptions=Exception,
+    reraise=True,
+)
 def create_webserver_thread_from_settings(app: FastAPI, settings: Dynaconf) -> threading.Thread:
     validate_settings(settings, validators=webserver_validators)
-
     return create_webserver_thread(app=app, port=settings.webserver.port, host=settings.webserver.host)
 
 
@@ -20,11 +37,43 @@ def create_webserver_thread(app: FastAPI, port: int, host: str) -> threading.Thr
     """Creates a thread that runs a FastAPI webserver. Start with thread.start(), and join with thread.join().
     Note that the thread is a daemon thread, so it will be terminated when the main thread is terminated.
     """
-    thread = threading.Thread(target=lambda: uvicorn.run(app, port=port, host=host, log_level=logging.WARNING))
+
+    def run_server():
+        retries = 5
+        for attempt in range(retries):
+            try:
+                uvicorn.run(app, port=port, host=host, log_level=logging.WARNING)
+                break
+            except Exception as e:
+                if attempt < retries - 1:  # if it's not the last attempt
+                    logger.warning(f"Attempt {attempt + 1} failed to start the server: {e}. Retrying...")
+                    time.sleep(2**attempt)  # exponential backoff
+                else:
+                    logger.error(f"Failed to start the server after {retries} attempts: {e}")
+                    raise
+
+    thread = threading.Thread(target=run_server)
     thread.daemon = True
     return thread
 
 
+async def run_async_webserver(app: FastAPI, port: int, host: str):
+    """Run the FastAPI web server async."""
+    config = uvicorn.Config(app, host=host, port=port, log_level=logging.WARNING)
+    server = PyInfraUvicornServer(config)
+
+    try:
+        await server.serve()
+    except asyncio.CancelledError:
+        logger.debug("Webserver was cancelled.")
+        server.should_exit = True
+        await server.shutdown()
+    except Exception as e:
+        logger.error(f"Error while running the webserver: {e}", exc_info=True)
+    finally:
+        logger.info("Webserver has been shut down.")
+
+
 HealthFunction = Callable[[], bool]
 
 
@@ -32,13 +81,23 @@ def add_health_check_endpoint(app: FastAPI, health_function: HealthFunction) ->
     """Add a health check endpoint to the app. The health function should return True if the service is healthy,
     and False otherwise. The health function is called when the endpoint is hit.
     """
+    if inspect.iscoroutinefunction(health_function):
 
-    @app.get("/health")
-    @app.get("/ready")
-    def check_health():
-        if health_function():
-            return {"status": "OK"}, 200
-        else:
+        @app.get("/health")
+        @app.get("/ready")
+        async def async_check_health():
+            alive = await health_function()
+            if alive:
+                return {"status": "OK"}, 200
+            return {"status": "Service Unavailable"}, 503
+
+    else:
+
+        @app.get("/health")
+        @app.get("/ready")
+        def check_health():
+            if health_function():
+                return {"status": "OK"}, 200
             return {"status": "Service Unavailable"}, 503
 
     return app

pyproject.toml

@@ -1,10 +1,9 @@
 [tool.poetry]
 name = "pyinfra"
-version = "1.10.0"
+version = "4.1.0"
 description = ""
 authors = ["Team Research <research@knecon.com>"]
 license = "All rights reseverd"
-readme = "README.md"
 
 [tool.poetry.dependencies]
 python = ">=3.10,<3.11"
@@ -19,18 +18,43 @@ azure-storage-blob = "^12.13"
 # misc utils
 funcy = "^2"
 pycryptodome = "^3.19"
-# research shared packages
-kn-utils = { version = "^0.2.4.dev112", source = "gitlab-research" }
 fastapi = "^0.109.0"
 uvicorn = "^0.26.0"
 
+# DONT USE GROUPS BECAUSE THEY ARE NOT INSTALLED FOR PACKAGES
+# [tool.poetry.group.internal.dependencies] <<< THIS IS NOT WORKING
+kn-utils = { version = ">=0.4.0", source = "nexus" }
+# We set all opentelemetry dependencies to lower bound because the image classification service depends on a protobuf version <4, but does not use proto files.
+# Therefore, we allow latest possible protobuf version in the services which use proto files. As soon as the dependency issue is fixed set this to the latest possible opentelemetry version
+opentelemetry-instrumentation-pika = ">=0.46b0,<0.50"
+opentelemetry-exporter-otlp = ">=1.25.0,<1.29"
+opentelemetry-instrumentation = ">=0.46b0,<0.50"
+opentelemetry-api = ">=1.25.0,<1.29"
+opentelemetry-sdk = ">=1.25.0,<1.29"
+opentelemetry-exporter-otlp-proto-http = ">=1.25.0,<1.29"
+opentelemetry-instrumentation-flask = ">=0.46b0,<0.50"
+opentelemetry-instrumentation-requests = ">=0.46b0,<0.50"
+opentelemetry-instrumentation-fastapi = ">=0.46b0,<0.50"
+opentelemetry-instrumentation-aio-pika = ">=0.46b0,<0.50"
+wcwidth = "<=0.2.12"
+azure-monitor-opentelemetry = "^1.6.0"
+aio-pika = "^9.4.2"
+aiohttp = "^3.9.5"
+
+# THIS IS NOT AVAILABLE FOR SERVICES THAT IMPLEMENT PYINFRA
 [tool.poetry.group.dev.dependencies]
 pytest = "^7"
 ipykernel = "^6.26.0"
-black = "^23.10"
+black = "^24.10"
 pylint = "^3"
 coverage = "^7.3"
 requests = "^2.31"
+pre-commit = "^3.6.0"
+cyclonedx-bom = "^4.1.1"
+dvc = "^3.51.2"
+dvc-azure = "^3.1.0"
+deepdiff = "^7.0.1"
+pytest-cov = "^5.0.0"
 
 [tool.pytest.ini_options]
 minversion = "6.0"
@@ -39,13 +63,39 @@ testpaths = ["tests", "integration"]
 log_cli = 1
 log_cli_level = "DEBUG"
 
+[tool.mypy]
+exclude = ['.venv']
+
+[tool.black]
+line-length = 120
+target-version = ["py310"]
+
+[tool.isort]
+profile = "black"
+
+[tool.pylint.format]
+max-line-length = 120
+disable = [
+    "C0114",
+    "C0325",
+    "R0801",
+    "R0902",
+    "R0903",
+    "R0904",
+    "R0913",
+    "R0914",
+    "W0511",
+]
+docstring-min-length = 3
+
 [[tool.poetry.source]]
-name = "PyPI"
+name = "pypi-proxy"
+url = "https://nexus.knecon.com/repository/pypi-proxy/simple"
 priority = "primary"
 
 [[tool.poetry.source]]
-name = "gitlab-research"
-url = "https://gitlab.knecon.com/api/v4/groups/19/-/packages/pypi/simple"
+name = "nexus"
+url = "https://nexus.knecon.com/repository/python/simple"
 priority = "explicit"
 
 [build-system]

scripts/send_async_request.py

@@ -0,0 +1,150 @@
+import asyncio
+import gzip
+import json
+from operator import itemgetter
+from typing import Any, Dict
+
+from aio_pika import Message
+from aio_pika.abc import AbstractIncomingMessage
+from kn_utils.logging import logger
+
+from pyinfra.config.loader import load_settings, local_pyinfra_root_path
+from pyinfra.queue.async_manager import AsyncQueueManager, RabbitMQConfig
+from pyinfra.storage.storages.s3 import S3Storage, get_s3_storage_from_settings
+
+settings = load_settings(local_pyinfra_root_path / "config/")
+
+
+async def dummy_message_processor(message: Dict[str, Any]) -> Dict[str, Any]:
+    logger.info(f"Processing message: {message}")
+    # await asyncio.sleep(1)  # Simulate processing time
+
+    storage = get_s3_storage_from_settings(settings)
+    tenant_id, dossier_id, file_id = itemgetter("tenantId", "dossierId", "fileId")(message)
+    suffix = message["responseFileExtension"]
+
+    object_name = f"{tenant_id}/{dossier_id}/{file_id}.{message['targetFileExtension']}"
+    original_content = json.loads(gzip.decompress(storage.get_object(object_name)))
+    processed_content = {
+        "processedPages": original_content["numberOfPages"],
+        "processedSectionTexts": f"Processed: {original_content['sectionTexts']}",
+    }
+
+    processed_object_name = f"{tenant_id}/{dossier_id}/{file_id}.{suffix}"
+    processed_data = gzip.compress(json.dumps(processed_content).encode("utf-8"))
+    storage.put_object(processed_object_name, processed_data)
+
+    processed_message = message.copy()
+    processed_message["processed"] = True
+    processed_message["processor_message"] = "This message was processed by the dummy processor"
+
+    logger.info(f"Finished processing message. Result: {processed_message}")
+    return processed_message
+
+
+async def on_response_message_callback(storage: S3Storage):
+    async def on_message(message: AbstractIncomingMessage) -> None:
+        async with message.process(ignore_processed=True):
+            if not message.body:
+                raise ValueError
+            response = json.loads(message.body)
+            logger.info(f"Received {response}")
+            logger.info(f"Message headers: {message.properties.headers}")
+            await message.ack()
+            tenant_id, dossier_id, file_id = itemgetter("tenantId", "dossierId", "fileId")(response)
+            suffix = response["responseFileExtension"]
+            result = storage.get_object(f"{tenant_id}/{dossier_id}/{file_id}.{suffix}")
+            result = json.loads(gzip.decompress(result))
+            logger.info(f"Contents of result on storage: {result}")
+
+    return on_message
+
+
+def upload_json_and_make_message_body(tenant_id: str):
+    dossier_id, file_id, suffix = "dossier", "file", "json.gz"
+    content = {
+        "numberOfPages": 7,
+        "sectionTexts": "data",
+    }
+
+    object_name = f"{tenant_id}/{dossier_id}/{file_id}.{suffix}"
+    data = gzip.compress(json.dumps(content).encode("utf-8"))
+
+    storage = get_s3_storage_from_settings(settings)
+    if not storage.has_bucket():
+        storage.make_bucket()
+    storage.put_object(object_name, data)
+
+    message_body = {
+        "tenantId": tenant_id,
+        "dossierId": dossier_id,
+        "fileId": file_id,
+        "targetFileExtension": suffix,
+        "responseFileExtension": f"result.{suffix}",
+    }
+    return message_body, storage
+
+
+async def test_rabbitmq_handler() -> None:
+    tenant_service_url = settings.storage.tenant_server.endpoint
+
+    config = RabbitMQConfig(
+        host=settings.rabbitmq.host,
+        port=settings.rabbitmq.port,
+        username=settings.rabbitmq.username,
+        password=settings.rabbitmq.password,
+        heartbeat=settings.rabbitmq.heartbeat,
+        input_queue_prefix=settings.rabbitmq.service_request_queue_prefix,
+        tenant_event_queue_suffix=settings.rabbitmq.tenant_event_queue_suffix,
+        tenant_exchange_name=settings.rabbitmq.tenant_exchange_name,
+        service_request_exchange_name=settings.rabbitmq.service_request_exchange_name,
+        service_response_exchange_name=settings.rabbitmq.service_response_exchange_name,
+        service_dead_letter_queue_name=settings.rabbitmq.service_dlq_name,
+        queue_expiration_time=settings.rabbitmq.queue_expiration_time,
+        pod_name=settings.kubernetes.pod_name,
+    )
+
+    handler = AsyncQueueManager(config, tenant_service_url, dummy_message_processor)
+
+    await handler.connect()
+    await handler.setup_exchanges()
+
+    tenant_id = "test_tenant"
+
+    # Test tenant creation
+    create_message = {"tenantId": tenant_id}
+    await handler.tenant_exchange.publish(
+        Message(body=json.dumps(create_message).encode()), routing_key="tenant.created"
+    )
+    logger.info(f"Sent create tenant message for {tenant_id}")
+    await asyncio.sleep(0.5)  # Wait for queue creation
+
+    # Prepare service request
+    service_request, storage = upload_json_and_make_message_body(tenant_id)
+
+    # Test service request
+    await handler.input_exchange.publish(Message(body=json.dumps(service_request).encode()), routing_key=tenant_id)
+    logger.info(f"Sent service request for {tenant_id}")
+    await asyncio.sleep(5)  # Wait for message processing
+
+    # Consume service request
+    response_queue = await handler.channel.declare_queue(name=f"response_queue_{tenant_id}")
+    await response_queue.bind(exchange=handler.output_exchange, routing_key=tenant_id)
+    callback = await on_response_message_callback(storage)
+    await response_queue.consume(callback=callback)
+
+    await asyncio.sleep(5)  # Wait for message processing
+
+    # Test tenant deletion
+    delete_message = {"tenantId": tenant_id}
+    await handler.tenant_exchange.publish(
+        Message(body=json.dumps(delete_message).encode()), routing_key="tenant.delete"
+    )
+    logger.info(f"Sent delete tenant message for {tenant_id}")
+    await asyncio.sleep(0.5)  # Wait for queue deletion
+
+    await handler.connection.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(test_rabbitmq_handler())

scripts/send_request.py

@@ -4,11 +4,11 @@ from operator import itemgetter
 
 from kn_utils.logging import logger
 
-from pyinfra.config.loader import load_settings, pyinfra_config_path
+from pyinfra.config.loader import load_settings, local_pyinfra_root_path
 from pyinfra.queue.manager import QueueManager
 from pyinfra.storage.storages.s3 import get_s3_storage_from_settings
 
-settings = load_settings(pyinfra_config_path)
+settings = load_settings(local_pyinfra_root_path / "config/")
 
 
 def upload_json_and_make_message_body():

scripts/send_sigterm.py

@@ -0,0 +1,17 @@
+import os
+import signal
+import time
+
+# BE CAREFUL WITH THIS SCRIPT - THIS SIMULATES A SIGTERM FROM KUBERNETES
+target_pid = int(input("Enter the PID of the target script: "))
+
+print(f"Sending SIGTERM to PID {target_pid}...")
+time.sleep(1)
+
+try:
+    os.kill(target_pid, signal.SIGTERM)
+    print("SIGTERM sent.")
+except ProcessLookupError:
+    print("Process not found.")
+except PermissionError:
+    print("Permission denied. Are you trying to signal a process you don't own?")

scripts/setup/devenvsetup.sh

@@ -0,0 +1,39 @@
+#!/bin/bash
+python_version=$1
+nexus_user=$2
+nexus_password=$3
+
+# cookiecutter https://gitlab.knecon.com/knecon/research/template-python-project.git --checkout master
+# latest_dir=$(ls -td -- */ | head -n 1)  # should be the dir cookiecutter just created
+
+# cd $latest_dir
+
+pyenv install $python_version
+pyenv local $python_version
+pyenv shell $python_version
+
+# install poetry globally (PREFERRED), only need to install it once
+# curl -sSL https://install.python-poetry.org | python3 -
+
+# remember to update poetry once in a while
+poetry self update
+
+# install poetry in current python environment, can lead to multiple instances of poetry being installed on one system (DISPREFERRED)
+# pip install --upgrade pip
+# pip install poetry
+
+poetry config virtualenvs.in-project true
+poetry config installer.max-workers 10
+poetry config repositories.pypi-proxy "https://nexus.knecon.com/repository/pypi-proxy/simple"
+poetry config http-basic.pypi-proxy ${nexus_user} ${nexus_password}
+poetry config repositories.nexus https://nexus.knecon.com/repository/python/simple
+poetry config http-basic.nexus ${nexus_user} ${nexus_password}
+
+poetry env use $(pyenv which python)
+poetry install --with=dev
+poetry update
+
+source .venv/bin/activate
+
+pre-commit install
+pre-commit autoupdate

scripts/start_pyinfra.py

@@ -1,7 +1,8 @@
 import time
 
-from pyinfra.config.loader import load_settings, parse_args
-from pyinfra.examples import start_queue_consumer_with_prometheus_and_health_endpoints
+from pyinfra.config.loader import load_settings, parse_settings_path
+from pyinfra.examples import start_standard_queue_consumer
+from pyinfra.queue.callback import make_download_process_upload_callback
 
 
 def processor_mock(_data: dict, _message: dict) -> dict:
@@ -10,5 +11,8 @@ def processor_mock(_data: dict, _message: dict) -> dict:
 
 
 if __name__ == "__main__":
-    settings = load_settings(parse_args().settings_path)
-    start_queue_consumer_with_prometheus_and_health_endpoints(processor_mock, settings)
+    arguments = parse_settings_path()
+    settings = load_settings(arguments)
+
+    callback = make_download_process_upload_callback(processor_mock, settings)
+    start_standard_queue_consumer(callback, settings)

tests/conftest.py

@@ -1,20 +1,48 @@
+import json
+
 import pytest
 
-from pyinfra.config.loader import load_settings, pyinfra_config_path
-from pyinfra.storage.connection import get_storage_from_settings
+from pyinfra.config.loader import load_settings, local_pyinfra_root_path
+from pyinfra.queue.manager import QueueManager
+from pyinfra.storage.connection import get_storage
 
 
 @pytest.fixture(scope="session")
 def settings():
-    return load_settings(pyinfra_config_path)
+    return load_settings(local_pyinfra_root_path / "config/")
 
 
 @pytest.fixture(scope="class")
 def storage(storage_backend, settings):
     settings.storage.backend = storage_backend
 
-    storage = get_storage_from_settings(settings)
+    storage = get_storage(settings)
     storage.make_bucket()
 
     yield storage
     storage.clear_bucket()
+
+
+@pytest.fixture(scope="session")
+def queue_manager(settings):
+    settings.rabbitmq_heartbeat = 10
+    settings.connection_sleep = 5
+    settings.rabbitmq.max_retries = 3
+    settings.rabbitmq.max_delay = 10
+    queue_manager = QueueManager(settings)
+    yield queue_manager
+
+
+@pytest.fixture
+def input_message():
+    return json.dumps(
+        {
+            "targetFilePath": "test/target.json.gz",
+            "responseFilePath": "test/response.json.gz",
+        }
+    )
+
+
+@pytest.fixture
+def stop_message():
+    return "STOP"

tests/data.dvc

@@ -0,0 +1,6 @@
+outs:
+- md5: 75cc98b7c8fcf782a7d4941594e6bc12.dir
+  size: 134913
+  nfiles: 9
+  hash: md5
+  path: data

tests/docker-compose.yml

@@ -1,31 +1,41 @@
-version: '2'
+version: '3.8'
 services:
   minio:
-    image: minio/minio:RELEASE.2022-06-11T19-55-32Z
+    image: minio/minio:latest
+    container_name: minio
     ports:
       - "9000:9000"
     environment:
       - MINIO_ROOT_PASSWORD=password
       - MINIO_ROOT_USER=root
     volumes:
-      - /tmp/minio_store:/data
+      - /tmp/data/minio_store:/data
     command: server /data
-    network_mode: "bridge" 
+    network_mode: "bridge"
+    extra_hosts:
+        - "host.docker.internal:host-gateway"
   rabbitmq:
-    image: docker.io/bitnami/rabbitmq:3.9.8
+    image: docker.io/bitnami/rabbitmq:latest
+    container_name: rabbitmq
     ports:
-      - '4369:4369'
-      - '5551:5551'
-      - '5552:5552'
+      # - '4369:4369'
+      # - '5551:5551'
+      # - '5552:5552'
       - '5672:5672'
-      - '25672:25672'
       - '15672:15672'
+      # - '25672:25672'
     environment:
       - RABBITMQ_SECURE_PASSWORD=yes
       - RABBITMQ_VM_MEMORY_HIGH_WATERMARK=100%
       - RABBITMQ_DISK_FREE_ABSOLUTE_LIMIT=20Gi
+      - RABBITMQ_MANAGEMENT_ALLOW_WEB_ACCESS=true
     network_mode: "bridge"
     volumes:
-      - /opt/bitnami/rabbitmq/.rabbitmq/:/data/bitnami
-volumes:
-  mdata:
+      - /tmp/bitnami/rabbitmq/.rabbitmq/:/data/bitnami
+    healthcheck:
+      test: [ "CMD", "curl", "-f", "http://localhost:15672" ]
+      interval: 30s
+      timeout: 10s
+      retries: 5
+    extra_hosts:
+        - "host.docker.internal:host-gateway"

tests/integration_test/opentelemetry_test.py

@@ -0,0 +1,41 @@
+from time import sleep
+
+import pytest
+
+from pyinfra.utils.opentelemetry import get_exporter, instrument_pika, setup_trace
+
+
+@pytest.fixture(scope="session")
+def exporter(settings):
+    settings.tracing.opentelemetry.exporter = "json"
+    return get_exporter(settings)
+
+
+@pytest.fixture(autouse=True)
+def setup_test_trace(settings, exporter, tracing_type):
+    settings.tracing.type = tracing_type
+    setup_trace(settings, exporter=exporter)
+
+
+class TestOpenTelemetry:
+    @pytest.mark.xfail(
+        reason="Azure Monitor requires a connection string. Therefore the test is allowed to fail in this case."
+    )
+    @pytest.mark.parametrize("tracing_type", ["opentelemetry", "azure_monitor"])
+    def test_queue_messages_are_traced(self, queue_manager, input_message, stop_message, settings, exporter):
+        instrument_pika()
+
+        queue_manager.purge_queues()
+        queue_manager.publish_message_to_input_queue(input_message)
+        queue_manager.publish_message_to_input_queue(stop_message)
+
+        def callback(_):
+            sleep(2)
+            return {"flat": "earth"}
+
+        queue_manager.start_consuming(callback)
+
+        for exported_trace in exporter.traces:
+            assert (
+                exported_trace["resource"]["attributes"]["service.name"] == settings.tracing.opentelemetry.service_name
+            )

tests/integration_test/prometheus_monitoring_test.py

@@ -5,7 +5,10 @@ import pytest
 import requests
 from fastapi import FastAPI
 
-from pyinfra.webserver.prometheus import add_prometheus_endpoint, make_prometheus_processing_time_decorator_from_settings
+from pyinfra.webserver.prometheus import (
+    add_prometheus_endpoint,
+    make_prometheus_processing_time_decorator_from_settings,
+)
 from pyinfra.webserver.utils import create_webserver_thread_from_settings
 
 

tests/integration_test/queue_test.py

@@ -3,11 +3,8 @@ from sys import stdout
 from time import sleep
 
 import pika
-import pytest
 from kn_utils.logging import logger
 
-from pyinfra.queue.manager import QueueManager
-
 logger.remove()
 logger.add(sink=stdout, level="DEBUG")
 
@@ -20,28 +17,21 @@ def make_callback(process_time):
     return callback
 
 
-@pytest.fixture(scope="session")
-def queue_manager(settings):
-    settings.rabbitmq_heartbeat = 10
-    settings.connection_sleep = 5
-    queue_manager = QueueManager(settings)
-    yield queue_manager
-
-
-@pytest.fixture
-def input_message():
-    return json.dumps({
-        "targetFilePath": "test/target.json.gz",
-        "responseFilePath": "test/response.json.gz",
-    })
-
-
-@pytest.fixture
-def stop_message():
-    return "STOP"
+def file_not_found_callback(x):
+    raise FileNotFoundError("File not found")
 
 
 class TestQueueManager:
+    def test_not_available_file_leads_to_message_rejection_without_crashing(
+            self, queue_manager, input_message, stop_message
+    ):
+        queue_manager.purge_queues()
+
+        queue_manager.publish_message_to_input_queue(input_message)
+        queue_manager.publish_message_to_input_queue(stop_message)
+
+        queue_manager.start_consuming(file_not_found_callback)
+
     def test_processing_of_several_messages(self, queue_manager, input_message, stop_message):
         queue_manager.purge_queues()
 

tests/integration_test/storage_test.py

@@ -5,8 +5,11 @@ from time import sleep
 import pytest
 from fastapi import FastAPI
 
-from pyinfra.storage.connection import get_storage_from_tenant_id
-from pyinfra.storage.utils import download_data_as_specified_in_message, upload_data_as_specified_in_message
+from pyinfra.storage.connection import get_storage_for_tenant
+from pyinfra.storage.utils import (
+    download_data_bytes_as_specified_in_message,
+    upload_data_as_specified_in_message,
+)
 from pyinfra.utils.cipher import encrypt
 from pyinfra.webserver.utils import create_webserver_thread
 
@@ -103,7 +106,11 @@ class TestMultiTenantStorage:
         self, tenant_id, tenant_server_mock, settings, tenant_server_host, tenant_server_port
     ):
         settings["storage"]["tenant_server"]["endpoint"] = f"http://{tenant_server_host}:{tenant_server_port}"
-        storage = get_storage_from_tenant_id(tenant_id, settings)
+        storage = get_storage_for_tenant(
+            tenant_id,
+            settings["storage"]["tenant_server"]["endpoint"],
+            settings["storage"]["tenant_server"]["public_key"],
+        )
 
         storage.put_object("file", b"content")
         data_received = storage.get_object("file")
@@ -125,23 +132,35 @@ def payload(payload_type):
             "targetFileExtension": "target.json.gz",
             "responseFileExtension": "response.json.gz",
         }
+    elif payload_type == "target_file_dict":
+        return {
+            "targetFilePath": {"file_1": "test/file.target.json.gz", "file_2": "test/file.target.json.gz"},
+            "responseFilePath": "test/file.response.json.gz",
+        }
 
 
-@pytest.mark.parametrize("payload_type", ["target_response_file_path", "dossier_id_file_id"], scope="class")
+@pytest.mark.parametrize(
+    "payload_type",
+    [
+        "target_response_file_path",
+        "dossier_id_file_id",
+        "target_file_dict",
+    ],
+    scope="class",
+)
 @pytest.mark.parametrize("storage_backend", ["azure", "s3"], scope="class")
 class TestDownloadAndUploadFromMessage:
-    def test_download_and_upload_from_message(self, storage, payload):
+    def test_download_and_upload_from_message(self, storage, payload, payload_type):
         storage.clear_bucket()
 
-        input_data = {"data": "success"}
+        result = {"process_result": "success"}
+        storage_data = {**payload, "data": result}
+        packed_data = gzip.compress(json.dumps(storage_data).encode())
 
-        storage.put_object("test/file.target.json.gz", gzip.compress(json.dumps(input_data).encode()))
+        storage.put_object("test/file.target.json.gz", packed_data)
 
-        data = download_data_as_specified_in_message(storage, payload)
-
-        assert data == input_data
-
-        upload_data_as_specified_in_message(storage, payload, input_data)
+        _ = download_data_bytes_as_specified_in_message(storage, payload)
+        upload_data_as_specified_in_message(storage, payload, result)
         data = json.loads(gzip.decompress(storage.get_object("test/file.response.json.gz")).decode())
 
-        assert data == {**payload, "data": input_data}
+        assert data == storage_data

tests/unit_test/config_test.py

@@ -1,9 +1,10 @@
 import os
+from pathlib import Path
 
 import pytest
 from dynaconf import Validator
 
-from pyinfra.config.loader import load_settings
+from pyinfra.config.loader import load_settings, local_pyinfra_root_path, normalize_to_settings_files
 from pyinfra.config.validators import webserver_validators
 
 
@@ -22,7 +23,7 @@ class TestConfig:
 
         validators = webserver_validators
 
-        test_settings = load_settings(validators=validators)
+        test_settings = load_settings(root_path=local_pyinfra_root_path, validators=validators)
 
         assert test_settings.webserver.host == "localhost"
 
@@ -30,7 +31,25 @@ class TestConfig:
         os.environ["TEST__VALUE__INT"] = "1"
         os.environ["TEST__VALUE__STR"] = "test"
 
-        test_settings = load_settings(validators=test_validators)
+        test_settings = load_settings(root_path=local_pyinfra_root_path, validators=test_validators)
 
         assert test_settings.test.value.int == 1
         assert test_settings.test.value.str == "test"
+
+    @pytest.mark.parametrize(
+        "settings_path,expected_file_paths",
+        [
+            (None, []),
+            ("config", [f"{local_pyinfra_root_path}/config/settings.toml"]),
+            ("config/settings.toml", [f"{local_pyinfra_root_path}/config/settings.toml"]),
+            (f"{local_pyinfra_root_path}/config", [f"{local_pyinfra_root_path}/config/settings.toml"]),
+        ],
+    )
+    def test_normalize_settings_files(self, settings_path, expected_file_paths):
+        files = normalize_to_settings_files(settings_path, local_pyinfra_root_path)
+        print(files)
+
+        assert len(files) == len(expected_file_paths)
+
+        for path, expected in zip(files, expected_file_paths):
+            assert path == Path(expected).absolute()

tests/unit_test/knutils_logger_test.py

@@ -4,7 +4,7 @@ from kn_utils.logging import logger
 
 def test_necessary_log_levels_are_supported_by_kn_utils():
     logger.setLevel("TRACE")
-    
+
     logger.trace("trace")
     logger.debug("debug")
     logger.info("info")
@@ -13,6 +13,7 @@ def test_necessary_log_levels_are_supported_by_kn_utils():
     logger.exception("exception", exc_info="this is an exception")
     logger.error("error", exc_info="this is an error")
 
+
 def test_setlevel_warn():
     logger.setLevel("WARN")
     logger.warning("warn")

tests/unit_test/utils_download_test.py

@@ -0,0 +1,83 @@
+import json
+
+import pytest
+from unittest.mock import patch
+from pyinfra.storage.utils import (
+    download_data_bytes_as_specified_in_message,
+    upload_data_as_specified_in_message,
+    DownloadedData,
+)
+from pyinfra.storage.storages.storage import Storage
+
+
+@pytest.fixture
+def mock_storage():
+    with patch("pyinfra.storage.utils.Storage") as MockStorage:
+        yield MockStorage()
+
+
+@pytest.fixture(
+    params=[
+        {
+            "raw_payload": {
+                "tenantId": "tenant1",
+                "dossierId": "dossier1",
+                "fileId": "file1",
+                "targetFileExtension": "txt",
+                "responseFileExtension": "json",
+            },
+            "expected_result": {
+                "data": b'{"key": "value"}',
+                "file_path": "tenant1/dossier1/file1.txt"
+            }
+        },
+        {
+            "raw_payload": {
+                "targetFilePath": "some/path/to/file.txt.gz",
+                "responseFilePath": "some/path/to/file.json"
+            },
+            "expected_result": {
+                "data": b'{"key": "value"}',
+                "file_path": "some/path/to/file.txt.gz"
+            }
+        },
+        {
+            "raw_payload": {
+                "targetFilePath": {
+                    "file1": "some/path/to/file1.txt.gz",
+                    "file2": "some/path/to/file2.txt.gz"
+                },
+                "responseFilePath": "some/path/to/file.json"
+            },
+            "expected_result": {
+                "file1": {
+                    "data": b'{"key": "value"}',
+                    "file_path": "some/path/to/file1.txt.gz"
+                },
+                "file2": {
+                    "data": b'{"key": "value"}',
+                    "file_path": "some/path/to/file2.txt.gz"
+                }
+            }
+        },
+    ]
+)
+def payload_and_expected_result(request):
+    return request.param
+
+def test_download_data_bytes_as_specified_in_message(mock_storage, payload_and_expected_result):
+    raw_payload = payload_and_expected_result["raw_payload"]
+    expected_result = payload_and_expected_result["expected_result"]
+    mock_storage.get_object.return_value = b'{"key": "value"}'
+
+    result = download_data_bytes_as_specified_in_message(mock_storage, raw_payload)
+
+    assert isinstance(result, dict)
+    assert result == expected_result
+    mock_storage.get_object.assert_called()
+
+def test_upload_data_as_specified_in_message(mock_storage, payload_and_expected_result):
+    raw_payload = payload_and_expected_result["raw_payload"]
+    data = {"key": "value"}
+    upload_data_as_specified_in_message(mock_storage, raw_payload, data)
+    mock_storage.put_object.assert_called_once()