implement demo service in message queue

.env.variables.yml

@@ -0,0 +1,12 @@
+variables:
+  # image created by this repository
+  IMAGE_NAME: ec/bprm
+  # image used in .gitlab-ci.yml
+  IMAGE_CI: alpine:3.21.3
+  #IMAGE_CI: python:3.13.2
+  # https://github.com/GoogleContainerTools/kaniko/releases
+  IMAGE_KANIKO: gcr.io/kaniko-project/executor:v1.24.0-debug
+  # image used by Dockerfile
+  # build arguments still have issues in the latest io.fabric8:docker-maven-plugin:0.44,
+  # so currently the FROM image is still in Dockerfile
+  IMAGE_FROM: python:3.13.2

.gitlab-ci.yml

@@ -0,0 +1,57 @@
+include: .env.variables.yml
+
+image: ${IMAGE_CI}
+
+.deploy:helper:
+  script:
+  # anchor definition to determine VERSION_NUMBER
+  - &version_number |
+    # using yaml anchor to determine VERSION_NUMBER
+    export GIT_BRANCH_REPLACED=$(echo $CI_COMMIT_REF_NAME | tr '[:upper:]' '[:lower:]' | sed 's/\//-/g')
+    export GIT_COMMIT_DATETIME=$(TZ=UTC git show -s --format=%cd $CI_COMMIT_SHA --date=iso-local)
+    export GIT_COMMIT_DATE_ONLY=$(echo $GIT_COMMIT_DATETIME | sed 's/[-:]//g' | cut -f1 -d' ')
+    export GIT_COMMIT_TIME_ONLY=$(echo $GIT_COMMIT_DATETIME | sed 's/[-:]//g' | cut -f2 -d' ')
+    export GIT_COMMIT_TIME=${GIT_COMMIT_DATE_ONLY}-${GIT_COMMIT_TIME_ONLY}
+    export GIT_COMMIT_ID_DESCRIBE_REPLACED=$(git describe --abbrev=8 --always --dirty=-MODIFIED-SNAPSHOT | tr '[:upper:]' '[:lower:]' | sed 's/\//-/g')
+    export VERSION_NUMBER=${GIT_BRANCH_REPLACED}-${GIT_COMMIT_TIME}.${GIT_COMMIT_ID_DESCRIBE_REPLACED}
+
+# https://docs.gitlab.com/ci/docker/authenticate_registry/
+
+stages:
+- prepare
+- build
+
+prepare:git-info:
+  # build file .env.ci to be used in later steps
+  stage: prepare
+  script:
+  - apk add git
+  - *version_number
+  - echo "VERSION_NUMBER=${VERSION_NUMBER}" > ./.env.ci
+  artifacts:
+    paths:
+    - .env.ci
+    expire_in: 1 day
+
+# https://docs.gitlab.com/ci/docker/using_kaniko/
+build:kaniko:
+  # kaniko builds and pushes in one step
+  needs: ["prepare:git-info"]
+  stage: build
+  image:
+    name: ${IMAGE_KANIKO}
+    entrypoint: [""]
+  script:
+  - ./kaniko-build-and-push.sh
+  when: manual
+
+# handle-action:
+#   stage: action
+#   before_script:
+#     - echo "$REPO_PASS" | docker login ${IMAGE_HOST} --username $REPO_USER --password-stdin
+#   script:
+#     - ./cicontainer-install-dependencies.sh
+#     # https://docs.ansible.com/ansible/devel/reference_appendices/config.html#cfg-in-world-writable-dir
+#     - chmod -R o-w $(pwd)
+#     - ./handle-action.sh ${CI_TARGET}
+#   when: manual

.variables.parse.sh

@@ -0,0 +1,52 @@
+#!/bin/sh
+
+# source: https://stackoverflow.com/questions/5014632/how-can-i-parse-a-yaml-file-from-a-linux-shell-script
+parse_ci_variables_yaml()
+{
+   filename=${1:-.env.variables.yml}
+   prefix=$2
+   s='[[:space:]]*' w='[a-zA-Z0-9_]*' fs=$(echo @|tr @ '\034')
+   # skipped indents at the start (for our case defaults to 1, since we always have "variables:" present)
+   skipped=${3:-1}
+   sed -ne "s|^\($s\):|\1|" \
+        -e "s|^\($s\)\($w\)$s:$s[\"']\(.*\)[\"']$s\$|\1$fs\2$fs\3|p" \
+        -e "s|^\($s\)\($w\)$s:$s\(.*\)$s\$|\1$fs\2$fs\3|p" $filename |
+   awk -F$fs -vskipped=$skipped '{
+      indent = length($1)/2;
+      vname[indent] = $2;
+      for (i in vname) {if (i > indent) {delete vname[i]}}
+      if (length($3) > 0) {
+         vn=""; for (i=skipped; i<indent; i++) {vn=(vn)(vname[i])("_")}
+         printf("%s%s%s=\"%s\"\n", "'$prefix'",vn, $2, $3);
+      }
+   }'
+}
+
+eval $(parse_ci_variables_yaml)
+
+if [ -e "./.env.ci" ]; then
+   # sourcing environment variables from other steps
+   . ./.env.ci
+fi
+
+if [ -e ".env" ]; then
+  . ./.env
+fi
+
+CONTAINER_IMAGE=${IMAGE_NAME}
+if [ -n "$IMAGE_HOST" ]; then
+   CONTAINER_IMAGE=$IMAGE_HOST/$CONTAINER_IMAGE
+fi
+export CONTAINER_LATEST=latest
+if [ -z "$1" ]; then
+  CONTAINER_TAG=
+else
+  CONTAINER_TAG=$1
+fi
+if [ -n "$VERSION_NUMBER" ]; then
+   # use version number (if set) as tag
+   CONTAINER_TAG=${VERSION_NUMBER}
+fi
+
+export CONTAINER_IMAGE=${CONTAINER_IMAGE}
+export CONTAINER_TAG=${CONTAINER_TAG}

README.md

@@ -18,3 +18,10 @@ Edit `/src/streamlit_app.py` to customize this app to your heart's desire. :hear
 
 If you have any questions, checkout our [documentation](https://docs.streamlit.io) and [community
 forums](https://discuss.streamlit.io).
+
+## Development
+Build Image:
+```
+docker build -t ec/demo .
+
+```

build.sh

@@ -0,0 +1,14 @@
+#!/bin/sh
+# parse and export .env.variables.yml
+. ./.variables.parse.sh
+
+echo "building $CONTAINER_IMAGE:$CONTAINER_LATEST"
+[ -z "$CONTAINER_TAG" ] || echo "and      $CONTAINER_IMAGE:$CONTAINER_TAG"
+
+docker\
+  build\
+  --pull\
+  --build-arg IMAGE_FROM=${IMAGE_FROM}\
+  -t $CONTAINER_IMAGE:$CONTAINER_LATEST\
+  `[ -z "$CONTAINER_TAG" ] || echo "-t $CONTAINER_IMAGE:$CONTAINER_TAG"`\
+  .

from-container-install-dependencies.sh

@@ -0,0 +1,9 @@
+#!/bin/sh
+#apk add py3-kubernetes\
+#        py3-jmespath
+apt update &&
+        apt install -y python3-pika curl &&
+        rm -rf /var/lib/apt/lists/*
+
+cd /workdir
+pip install -r requirements.txt --user

kaniko-build-and-push.sh

@@ -0,0 +1,105 @@
+#!/bin/sh
+# parse and export .env.variables.yml
+if [ -n "${CI_PROJECT_DIR}" ]; then
+  # CI/CD environment
+  cd ${CI_PROJECT_DIR}
+fi
+
+. ./.variables.parse.sh
+
+echo "building $CONTAINER_IMAGE:$CONTAINER_LATEST"
+[ -z "$CONTAINER_TAG" ] || echo "and      $CONTAINER_IMAGE:$CONTAINER_TAG"
+
+# build variants
+
+can_create_kaniko_config()
+{
+  if [ -z "${IMAGE_KANIKO}" -o -z "${IMAGE_HOST}" -o -z "${REPO_USER}" -o -z "${REPO_PASSWORD}" ]; then
+    echo "1"
+  else
+    echo "0"
+  fi
+}
+
+create_kaniko_config()
+{
+  TARGET_FILENAME="${1:-/kaniko/.docker/config.json}"
+  echo "creating temporary docker.config for kaniko (${TARGET_FILENAME})"
+  if [ -z "${IMAGE_HOST}" -o -z "${REPO_USER}" -o -z "${REPO_PASSWORD}" -o -z "${TARGET_FILENAME}" ]; then
+    echo "kaniko docker config cannot be created. are variables set? IMAGE_HOST, REPO_USER, REPO_PASSWORD"
+    exit 1
+  fi
+  # content for /kaniko/.docker/config.json
+  echo "{
+    \"auths\": {
+      \"${IMAGE_HOST}\": {
+        \"auth\": \"$(printf "%s:%s" "${REPO_USER}" "${REPO_PASSWORD}" | base64 | tr -d '\n')\"
+      }
+    }
+  }" > "${TARGET_FILENAME}"
+}
+
+build_and_push_kaniko()
+{
+  create_kaniko_config
+  /kaniko/executor \
+    --context . \
+    --dockerfile Dockerfile \
+    --destination "${CONTAINER_IMAGE}:${CONTAINER_TAG}"
+}
+
+build_and_push_docker_kaniko()
+{
+  TMP_KANIKO_CONFIG="kaniko.docker.config.json.tmp"
+  create_kaniko_config "${TMP_KANIKO_CONFIG}"
+
+  docker \
+    run --rm -i \
+    -mount type=bind,source="$(pwd)/${TMP_KANIKO_CONFIG}",target=/kaniko/.docker/config.json \
+    -mount type=bind,source="$(pwd)",target=/workspace \
+    -e "IMAGE_HOST=${IMAGE_HOST}" \
+    -e "REPO_USER=${REPO_USER}" \
+    -e "REPO_PASSWORD=${REPO_PASSWORD}" \
+    ${IMAGE_KANIKO} \
+    /kaniko/executor \
+    --context . \
+    --dockerfile Dockerfile \
+    --destination "${CONTAINER_IMAGE}:${CONTAINER_TAG}"
+
+  rm "${TMP_KANIKO_CONFIG}"
+}
+
+build_docker_only()
+{
+  docker \
+    build \
+    --build-arg "IMAGE_FROM=${IMAGE_FROM}" \
+    -t $CONTAINER_IMAGE:$CONTAINER_LATEST \
+    $([ -z "$CONTAINER_TAG" ] && echo "" || echo "-t $CONTAINER_IMAGE:$CONTAINER_TAG") \
+    .
+}
+
+push_docker_only()
+{
+  echo docker \
+    push \
+    -t $CONTAINER_IMAGE:$CONTAINER_LATEST \
+    $([ -z "$CONTAINER_TAG" ] && echo "" || echo "-t $CONTAINER_IMAGE:$CONTAINER_TAG")
+}
+
+HAS_REQUIRED_PARAMETERS=$(can_create_kaniko_config)
+# kaniko not present, using docker to start kaniko build
+if [ "${HAS_REQUIRED_PARAMETERS}" = "0" ]; then
+  if [ -e "/kaniko/executor" ]; then
+    # kaniko present, building with kaniko
+    echo building with kaniko
+    build_and_push_kaniko
+  else
+    echo building with kaniko using docker
+    build_and_push_docker_kaniko
+  fi
+else
+  echo building with docker
+  build_docker_only
+  push_docker_only
+fi

local-docker-build.sh

@@ -0,0 +1,13 @@
+#!/bin/sh
+# parse and export .env.variables.yml
+. ./.variables.parse.sh
+
+echo "building $CONTAINER_IMAGE:$CONTAINER_LATEST"
+[ -z "$CONTAINER_TAG" ] || echo "and      $CONTAINER_IMAGE:$CONTAINER_TAG"
+
+docker\
+  build\
+  --build-arg IMAGE_FROM=${IMAGE_FROM}\
+  -t $CONTAINER_IMAGE:$CONTAINER_LATEST\
+  $([ -z "$CONTAINER_TAG" ] && echo "" || echo "-t $CONTAINER_IMAGE:$CONTAINER_TAG")\
+  .

push.sh

@@ -0,0 +1,13 @@
+#!/bin/sh
+# parse and export .env.variables.yml
+. ./.variables.parse.sh
+echo "pushing $CONTAINER_IMAGE:$CONTAINER_LATEST" &&\
+docker\
+  push\
+  $CONTAINER_IMAGE:$CONTAINER_LATEST
+
+[ -z "$CONTAINER_TAG" ] ||\
+echo "pushing $CONTAINER_IMAGE:$CONTAINER_TAG" &&\
+docker\
+  push\
+  $CONTAINER_IMAGE:$CONTAINER_TAG

run-ci-container.sh

@@ -0,0 +1,20 @@
+#!/bin/sh
+# parse and export .env.variables.yml
+. ./.variables.parse.sh
+
+export BUILDKIT_PROGRESS=plain
+
+# special case for windows
+DOCKER_PREFIX=$(which winpty)
+if [ -n "$DOCKER_PREFIX" ]; then
+  DOCKER_PREFIX="$DOCKER_PREFIX "
+fi
+
+# https://superuser.com/questions/344478/bash-execute-command-given-in-commandline-and-dont-exit
+${DOCKER_PREFIX}docker run \
+  --rm\
+  -ti\
+  --mount "type=bind,src=/var/run/docker.sock,target=/var/run/docker.sock"\
+  --mount "type=bind,src=$(pwd),target=/workdir"\
+  "$IMAGE_CI"\
+  sh -c "cd /workdir ; exec sh"

run-from-container.sh

@@ -0,0 +1,19 @@
+#!/bin/sh
+# parse and export .env.variables.yml
+. ./.variables.parse.sh
+
+export BUILDKIT_PROGRESS=plain
+
+# special case for windows
+DOCKER_PREFIX=$(which winpty)
+if [ -n "$DOCKER_PREFIX" ]; then
+  DOCKER_PREFIX="$DOCKER_PREFIX "
+fi
+
+# https://superuser.com/questions/344478/bash-execute-command-given-in-commandline-and-dont-exit
+${DOCKER_PREFIX}docker run \
+  --rm\
+  -ti\
+  --mount "type=bind,src=$(pwd),target=/workdir"\
+  "$IMAGE_FROM"\
+  bash -c "echo Entering FROM container && /workdir/from-container-install-dependencies.sh && cd /workdir; exec bash"

src/pages/Pipeline_Demo.py

@@ -0,0 +1,77 @@
+import os
+import logging
+import threading
+from typing import Callable
+
+import streamlit as st
+from dotenv import load_dotenv
+
+from project import Config, DemoThread
+from project.adapters import InAdapter, OutAdapter, MessageQueueOutAdapterImpl
+from wrapper import HealthCheck
+
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger(__name__)
+
+load_dotenv()
+
+
+@st.cache_resource
+def create_project_thread(_config: Config) -> DemoThread:
+    in_adapter_injector: Callable[[Config], InAdapter] = lambda c: InAdapter()
+
+    if config.mq_user == "local":
+        logger.info("Running with local OutAdapter (no MQ connection).")
+        out_adapter_injector: Callable[[Config], OutAdapter] = lambda c: OutAdapter()
+    else:
+        logger.info(f"Using real MessageQueueOutAdapter -> {config.queue_dest}")
+        out_adapter_injector: Callable[[Config], OutAdapter] = lambda c: MessageQueueOutAdapterImpl(c, c.queue_dest)
+
+    return DemoThread(config, None, in_adapter_injector, out_adapter_injector)
+
+
+config = Config()
+config.display()
+
+
+publish_thread = create_project_thread(config)
+
+if config.is_start_publish():
+    logger.info("Starting MQ consume thread...")
+    consume_thread = threading.Thread(target=publish_thread.run, daemon=True)
+    consume_thread.start()
+
+
+
+st.title("Pipeline Demo")
+st.markdown("Sende eine Url an die Message Queue um die Pipeline zu starten:")
+
+user_input = st.text_input("URL eingeben:")
+
+if st.button("Senden"):
+    if user_input:
+        try:
+            data = {"url": user_input}
+            message_bytes = bytes(str(data), "utf-8")
+
+            # Direkter Aufruf des OutAdapters zum Senden
+            publish_thread._DemoThread__queue_dest.send_message(message_bytes)
+
+            logger.info(f"Message sent to MQ: {data}")
+        except Exception as e:
+            logger.exception("Fehler beim MQ-Senden")
+    else:
+        st.warning("Bitte eine Nachricht eingeben, bevor du sendest.")
+
+
+# ------------------------------------------------------------
+# 🔍 Optional: Debug-Bereich
+# ------------------------------------------------------------
+with st.expander("Debug-Informationen"):
+    st.json({
+        "mq_host": config.mq_host,
+        "mq_user": config.mq_user,
+        "mq_queue_dest": config.queue_dest,
+        "is_start_publish": config.is_start_publish()
+    })

src/project/__init__.py

@@ -0,0 +1,2 @@
+from .config import Config
+from .demo_thread import DemoThread

src/project/adapters/__init__.py

@@ -0,0 +1,4 @@
+from .in_adapter import InAdapter
+from .message_queue_in_adapter_impl import MessageQueueInAdapterImpl
+from .message_queue_out_adapter_impl import MessageQueueOutAdapterImpl
+from .out_adapter import OutAdapter

src/project/adapters/in_adapter.py

@@ -0,0 +1,12 @@
+from abc import abstractmethod
+from typing import Callable
+
+from pika import spec
+from pika.adapters.blocking_connection import BlockingChannel
+
+
+class InAdapter:
+
+    @abstractmethod
+    def start_consuming(self, callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]) -> None:
+        pass

src/project/adapters/message_queue_in_adapter_impl.py

@@ -0,0 +1,19 @@
+from typing import Callable
+
+from pika import spec
+from pika.adapters.blocking_connection import BlockingChannel
+
+from project import Config
+from project.adapters.in_adapter import InAdapter
+from wrapper import MessageQueue
+
+
+class MessageQueueInAdapterImpl(InAdapter):
+
+    def __init__(self, config: Config, queue_name: str) -> None:
+        self.__config = config
+        # initialize queue
+        self.__queue = MessageQueue(config.mq_host, config.mq_user, config.mq_password, queue_name)
+
+    def start_consuming(self, callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]) -> None:
+        self.__queue.start_consuming(callback)

src/project/adapters/message_queue_out_adapter_impl.py

@@ -0,0 +1,14 @@
+from project import Config
+from project.adapters.out_adapter import OutAdapter
+from wrapper import MessageQueue
+
+
+class MessageQueueOutAdapterImpl(OutAdapter):
+
+    def __init__(self, config: Config, queue_name: str) -> None:
+        self.__config = config
+        # initialize queue
+        self.__queue = MessageQueue(config.mq_host, config.mq_user, config.mq_password, queue_name)
+
+    def send_message(self, message: bytes) -> None:
+        self.__queue.send_message(message)

src/project/adapters/out_adapter.py

@@ -0,0 +1,8 @@
+from abc import abstractmethod
+
+
+class OutAdapter:
+
+    @abstractmethod
+    def send_message(self, message: bytes) -> None:
+        pass

src/project/config.py

@@ -0,0 +1,65 @@
+import os
+import sys
+
+MODE_CONSUME = 'consume'
+MODE_CONSUME_AND_PUBLISH = 'consume-and-publish'
+MODE_PUBLISH = 'publish'
+
+
+class Config:
+    @staticmethod
+    def __assert_variable_is_defined(value: ..., message: str) -> None:
+        if value is None:
+            # https://www.geeksforgeeks.org/python-exit-commands-quit-exit-sys-exit-and-os-_exit/
+            print(message)
+            # https://docs.python.org/3/library/sys.html#sys.exit
+            sys.exit(1)
+
+    def __init__(self) -> None:
+        self.mq_host = os.environ.get('MQ_HOST', 'mq-service')
+        self.mq_user = os.environ.get('MQ_USER')
+        self.mq_password = os.environ.get('MQ_PASSWORD')
+
+        # different modes:
+        self.queue_mode = os.environ.get('QUEUE_MODE', MODE_CONSUME).lower()
+        print(self.queue_mode)
+        # consume: take elements from a queue (alias pull, alias store)
+        if self.queue_mode in [MODE_CONSUME, 'pull', 'store', ]:
+            self.queue_mode = MODE_CONSUME
+        # publish: emit elements to a queue (alias push, alias load)
+        if self.queue_mode in [MODE_PUBLISH, 'push', 'load', ]:
+            self.queue_mode = MODE_PUBLISH
+        # consume-and-publish: take elements from a queue, (store the last COUNT), and publish to another queue
+        if self.queue_mode in [MODE_CONSUME_AND_PUBLISH, 'pull-push', 'store-n-forward', ]:
+            self.queue_mode = MODE_CONSUME_AND_PUBLISH
+
+        self.queue_src = os.environ.get('QUEUE_SRC')
+        self.queue_dest = os.environ.get('QUEUE_DEST')
+        self.data_directory = os.environ.get('DATA_DIRECTORY', '/storage/data/')
+        # if mode is consume-and-publish the data_index file will be used for consume and will have a suffix ".published"
+        self.data_index = os.environ.get('DATA_INDEX', '/storage/index')
+
+        # asserts
+        self.__assert_variable_is_defined(self.mq_user, "User not specified in MQ_USER")
+        self.__assert_variable_is_defined(self.mq_password, "Password not specified in MQ_PASSWORD")
+        if self.is_start_publish():
+            self.__assert_variable_is_defined(self.queue_dest, "Destination queue is not specified in QUEUE_DEST")
+        if self.is_start_consume():
+            self.__assert_variable_is_defined(self.queue_src, "Source queue is not specified in QUEUE_SRC")
+
+    def display(self) -> None:
+        for name in sorted(dir(self)):
+            if not name.startswith('_') and name not in [
+                'display',
+                'is_start_consume',
+                'is_start_publish',
+                'mq_password',
+            ]:
+                print(f"{name.upper()}={getattr(self, name)}")
+        print()
+
+    def is_start_consume(self) -> bool:
+        return self.queue_mode in [MODE_CONSUME, MODE_CONSUME_AND_PUBLISH, ]
+
+    def is_start_publish(self) -> bool:
+        return self.queue_mode in [MODE_PUBLISH, MODE_CONSUME_AND_PUBLISH, ]

src/project/demo_thread.py

@@ -0,0 +1,49 @@
+import ast
+from typing import Callable, Union
+
+from project.adapters import InAdapter, OutAdapter
+from wrapper import HealthCheck
+from . import Config
+
+
+class DemoThread:
+    def __init__(
+            self,
+            config: Config,
+            health_check: Union[HealthCheck, None],
+            in_adapter_injector: Callable[[Config], InAdapter],
+            out_adapter_injector: Callable[[Config], OutAdapter]
+    ) -> None:
+        """
+        Create BoilerplateRemovalThread instance.
+        :param config: configuration options
+        :param health_check: health check
+        :param in_adapter_injector: input adapter, based on config
+        :param out_adapter_injector: output adapter, based on config
+        """
+        self.__config = config
+        self.__health_check = health_check
+        # initialize queues
+        self.__queue_src = in_adapter_injector(config)
+        self.__queue_dest = out_adapter_injector(config)
+
+    def handle_element(self, ch, method, properties, body: bytes) -> None:
+        print(f"CONSUME: Received element {properties}: {len(body)}")
+
+        try:
+            url = ast.literal_eval(body.decode('utf-8'))['url']
+            data = {"url": url}
+            self.__queue_dest.send_message(bytes(str(data), encoding="utf-8"))
+            print(f"CONSUME: Send element {properties}")
+
+        except Exception as e:
+            print("Error", e)
+
+    def run(self) -> None:
+        # consume elements
+        # TODO: are these callbacks called only one at a time or in multiple threads?
+        # self.__handle_element("ch", "method", "properties", bytes("localhost/health", "utf-8"))
+        # self.__handle_element("ch", "method", "properties", bytes("localhost/health", "utf-8"))
+        self.__queue_src.start_consuming(self.handle_element)
+        # print("CONSUME iteration")
+        # time.sleep(1)

src/streamlit_app.py

@@ -1 +1,3 @@
 import streamlit as st
+
+

src/wrapper/__init__.py

@@ -0,0 +1,2 @@
+from .health_check import HealthCheck
+from .message_queue import MessageQueue

src/wrapper/health_check.py

@@ -0,0 +1,73 @@
+import logging
+from typing import Callable, Dict
+
+from flask import Flask
+from flask.typing import ResponseValue
+
+
+# adapted from https://stackoverflow.com/questions/62000942/flask-block-specific-endpoints-from-logging
+class AjaxFilter(logging.Filter):
+    def filter(self, record):
+        return "/health" not in record.getMessage()
+
+
+class HealthCheck:
+    # adapted from https://thelinuxnotes.com/index.php/implementing-a-flask-health-check-and-kubernetes-liveness-probe-in-python-application/
+    def __do_check(self) -> tuple[ResponseValue, int]:
+        # https://docs.python.org/3.9/library/functions.html#all
+        # https://stackoverflow.com/questions/12229064/mapping-over-values-in-a-python-dictionary
+        # loop over checks
+        # execute check method for each key
+        results = {k: v() for k, v in self.__checks.items()}
+        overall_result = all(results.values())
+        json = {
+            "state": "OK" if overall_result else "ERROR",
+            "elements": results
+        }
+        status_code = 200 if overall_result else 500
+        return json, status_code
+
+    def __init__(self, app: Flask) -> None:
+        self.__app = app
+        self.__checks: Dict[str, Callable[[], bool]] = {}
+        self.__log = logging.getLogger('werkzeug')
+        self.disable_logging_filter = AjaxFilter()
+
+        # register route dynamically
+        # https://pytutorial.com/flask-appadd_url_rule-dynamic-url-registration-guide/
+        # @app.route('/health')
+        # @app.route('/health/')
+        self.__app.add_url_rule(
+            '/health',  # URL rule with variable
+            'health_check',  # Endpoint name
+            view_func=self.__do_check,  # View function
+            methods=['GET'],  # Allowed methods
+            strict_slashes=False  # URL trailing slash handling
+        )
+
+    def disable_logging(self) -> None:
+        print("health check logging disabled")
+        self.__log.addFilter(self.disable_logging_filter)
+
+    def enable_logging(self) -> None:
+        print("health check logging enabled")
+        self.__log.removeFilter(self.disable_logging_filter)
+
+    def add_check(self, name, method: Callable[[], bool]) -> None:
+        """method to add named check methods to health check
+        Parameters
+        ----------
+        name : str
+            named key of check method
+        method : function
+            method which returns a bool for the health check
+        """
+        self.__checks.update({f"{name}": method})
+
+    def wrap_exception_handler(self, name: str, method: Callable[[], None]) -> None:
+        """wrap exception handler around method and register simple health check for it"""
+        self.add_check(name, lambda: True)
+        try:
+            method()
+        finally:
+            self.add_check(name, lambda: False)

src/wrapper/message_queue.py

@@ -0,0 +1,80 @@
+from typing import Callable
+
+import pika
+from pika import BlockingConnection, spec
+from pika.adapters.blocking_connection import BlockingChannel
+from pika.exceptions import StreamLostError
+
+
+class MessageQueue:
+    __channel: BlockingChannel
+    __connection: BlockingConnection
+    __callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]
+
+    def __init__(self, host: str, user: str, password: str, queue_name: str) -> None:
+        self.__host = host
+        self.__user = user
+        self.__password = password
+        self.__queue_name = queue_name
+        self.connect()
+        # create queue if necessary
+        self.__channel.queue_declare(queue=self.__queue_name, durable=True)
+
+    def close(self) -> None:
+        self.__connection.close()
+
+    def connect(self) -> None:
+        # print(f"Connecting to {self.__queue_name}")
+        self.__connection = pika.BlockingConnection(
+            pika.ConnectionParameters(
+                credentials=pika.PlainCredentials(self.__user, self.__password),
+                host=self.__host
+            ))
+        self.__channel = self.__connection.channel()
+
+    def send_message(self, message: bytes) -> None:
+        # message = ' '.join(sys.argv[1:]) or "Hello World!"
+        def __send():
+            self.__channel.basic_publish(
+                exchange='',
+                routing_key=self.__queue_name,
+                body=message,
+                properties=pika.BasicProperties(
+                    delivery_mode=pika.DeliveryMode.Persistent
+                ))
+
+        # try repeating once, if connection was lost
+        retry = True
+        while retry:
+            try:
+                __send()
+                retry = False
+            except StreamLostError as e:
+                print("Reconnecting...")
+                if retry:
+                    self.connect()
+                    retry = False
+                else:
+                    raise e
+        # print(f" [x] Sent {message}")
+        # connection.close()
+
+    def __ack_and_call_callback(
+            self,
+            ch: BlockingChannel,
+            method: spec.Basic.Deliver,
+            properties: spec.BasicProperties,
+            body: bytes) -> None:
+        # acknowledge processing directly
+        ch.basic_ack(delivery_tag=method.delivery_tag)
+        # print(f" [x] Received {body.decode()}")
+        self.__callback(ch, method, properties, body)
+        # print(" [x] Done")
+
+    def start_consuming(self, callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]) -> None:
+        """callback method parameter is a method with the following signature:
+        def callback(ch, method, properties, body):"""
+        self.__callback = callback
+        self.__channel.basic_qos(prefetch_count=1)
+        self.__channel.basic_consume(queue=self.__queue_name, on_message_callback=self.__ack_and_call_callback)
+        self.__channel.start_consuming()