make service publish-only | create Testing page | refactoring

src/pages/Eventportal.py

@@ -1,12 +1,10 @@
-import os
 from datetime import datetime
+from itertools import batched
 
 from bson import ObjectId
 from streamlit import streamlit as st
 
-from dotenv import load_dotenv
-from pymongo import MongoClient
-from itertools import batched
+from services.db_service import init_connection
 
 st.set_page_config(layout="wide")
 
@@ -26,23 +24,16 @@ st.markdown(
 review_options = [":green[:material/done:]", ":red[:material/highlight_off:]"]
 review_filter_options = ["Abgeschlossene Reviews", "Offene Reviews"]
 review_fields = [
-        ("Titel", "title_valid"),
-        ("Kategorie", "category_valid"),
-        ("Subkategorie", "format_valid"),
-        ("Startdatum", "startdate_valid"),
-        ("Startzeit", "starttime_valid"),
-        ("Enddatum", "enddate_valid"),
-        ("Endzeit", "endtime_valid"),
-        ("Adresse", "address_valid"),
-        ("Link", "url_valid")
-    ]
-
-@st.cache_resource
-def init_connection():
-    load_dotenv()
-    uri = f"mongodb+srv://{os.getenv('MONGO_USERNAME')}:{os.getenv('MONGO_PASSWORD')}@{os.getenv('MONGO_HOST')}/?retryWrites=true&w=majority&appName=Cluster0"
-    client = MongoClient(uri)
-    return client.event_data
+    ("Titel", "title_valid"),
+    ("Kategorie", "category_valid"),
+    ("Subkategorie", "format_valid"),
+    ("Startdatum", "startdate_valid"),
+    ("Startzeit", "starttime_valid"),
+    ("Enddatum", "enddate_valid"),
+    ("Endzeit", "endtime_valid"),
+    ("Adresse", "address_valid"),
+    ("Link", "url_valid")
+]
 
 
 @st.cache_resource
@@ -183,9 +174,9 @@ def display_event(event_data):
 
         if comments:
             st.write("**Kommentare**")
-            for comment_id,comment in enumerate(comments):
+            for comment_id, comment in enumerate(comments):
                 st.write(f"💬 {comment}")
-        st.chat_input("Neuer Kommentar" , key="comment_input" + str(event_data["_id"]),
+        st.chat_input("Neuer Kommentar", key="comment_input" + str(event_data["_id"]),
                       on_submit=update_review_comment,
                       args=(str(event_data["_id"]),))
 
@@ -193,8 +184,6 @@ def display_event(event_data):
 def render_review_controls(event_data):
     """Erzeugt dynamisch alle Review-Segmented-Controls für ein Event."""
 
-
-
     # Dynamisch Spalten erzeugen
     cols = st.columns(len(review_fields))
 
@@ -335,13 +324,13 @@ batches = st.session_state.batched_events
 
 # Page Content
 
-col_rev_toggle, col_control1 = st.columns([1,4])
+col_rev_toggle, col_control1 = st.columns([1, 4])
 with col_rev_toggle:
     st.toggle("Review Mode", value=False, on_change=toggle_review_mode)
 if st.session_state.review_mode:
     with col_control1:
-        st.pills("Filter Reviews", review_filter_options, key="filter_reviews_input", label_visibility="collapsed", on_change=update_filter)
-
+        st.pills("Filter Reviews", review_filter_options, key="filter_reviews_input", label_visibility="collapsed",
+                 on_change=update_filter)
 
 st.title("Eventportal")
 

src/pages/Pipeline_Demo.py

@@ -1,43 +1,15 @@
-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
-
-logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
-logger = logging.getLogger(__name__)
+from project import Config
+from services.mq_service import create_out_adapter
 
 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()
+out_adapter = create_out_adapter(config, config.queue_dest_pipeline_trigger)
 
 st.title("Pipeline Demo")
 st.markdown("Sende eine Url an die Message Queue um die Pipeline zu starten:")
@@ -47,11 +19,9 @@ user_input = st.text_input("URL eingeben:")
 if st.button("Senden"):
     if user_input:
         try:
-
-            publish_thread._DemoThread__queue_dest.send_message(user_input.strip())
+            out_adapter.send_message(user_input.strip())
             st.info("URL wurde gesendet")
-            logger.info(f"Message sent to MQ: {user_input}")
         except Exception as e:
-            logger.exception("Fehler beim MQ-Senden")
+            st.error(f"Fehler beim senden in die Message Queue: {e}")
     else:
         st.warning("Bitte eine Nachricht eingeben, bevor du sendest.")

src/pages/Testing.py

@@ -0,0 +1,63 @@
+import base64
+import json
+
+import streamlit as st
+from dotenv import load_dotenv
+
+from project import Config
+from services.db_service import init_connection
+from services.mq_service import create_out_adapter
+
+load_dotenv()
+config = Config()
+config.display()
+out_adapter = create_out_adapter(config, config.queue_dest_test_trigger)
+
+db = init_connection()
+
+st.title("Testing")
+st.write(f"Es sind {db.testdata_1.count_documents({})} Testdatensätze vorhanden.")
+
+with st.form("Testkonfiguration"):
+    batch_id = st.selectbox("Testbatch wählen", [0, 1, 2, 3])
+
+    start_test = st.form_submit_button("Starte Test")
+
+if start_test:
+    testbatch = list(db.testdata_1.find({"batch_id": batch_id}))
+    batch_size = len(testbatch)
+    doc = db.test_evaluation.find_one({}, sort=[("test_number", -1)])
+    test_number = doc.get("test_number") + 1 if doc else 0
+
+    st.write(f"Starte Test mit {len(testbatch)} Testdaten, Test-ID: {test_number}")
+    insert_result = db.test_evaluation.insert_one(
+        {"test_number": test_number, "batch_size": batch_size, "batch_id": batch_id, "status": "Pending"})
+    test_id = insert_result.inserted_id if insert_result else None
+    if test_id:
+        for testdata in testbatch:
+            id = str(testdata["_id"])
+            url = testdata.get("url")
+            html = testdata.get("html").decode("utf-8")
+            data = testdata.get("data")
+            st.write(id)
+            st.write(url)
+            st.write("___")
+
+            try:
+                message = {
+                    "url": url,
+                    "body": base64.b64encode(html.encode("utf-8")).decode("utf-8"),
+                    "meta": [{
+                        "service_name": "demo-service",
+                        "queue_dest": "storehtml-out",
+                        "test_data": {"test_id": str(test_id), "test_number": test_number, "batch_size": batch_size,
+                                      "batch_id": batch_id, "data_id": id},
+                    }]
+                }
+                encoded = json.dumps(message).encode("utf-8")
+                out_adapter.send_message(encoded)
+
+            except Exception as e:
+                st.error(f"Fehler beim senden in die Message Queue: {e}")
+    else:
+        st.error("Fehler beim anlegen des Tests")

src/project/__init__.py

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

src/project/adapters/__init__.py

@@ -1,4 +1,2 @@
-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

@@ -1,12 +0,0 @@
-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

@@ -1,19 +0,0 @@
-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/config.py

@@ -21,7 +21,7 @@ class Config:
         self.mq_password = os.environ.get('MQ_PASSWORD')
 
         # different modes:
-        self.queue_mode = os.environ.get('QUEUE_MODE', MODE_CONSUME).lower()
+        self.queue_mode = os.environ.get('QUEUE_MODE', MODE_PUBLISH).lower()
         print(self.queue_mode)
         # consume: take elements from a queue (alias pull, alias store)
         if self.queue_mode in [MODE_CONSUME, 'pull', 'store', ]:
@@ -33,8 +33,8 @@ class Config:
         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.queue_dest_test_trigger= os.environ.get('QUEUE_DEST_TEST_TRIGGER')
+        self.queue_dest_pipeline_trigger = os.environ.get('QUEUE_DEST_PIPELINE_TRIGGER')
         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')
@@ -43,9 +43,9 @@ class Config:
         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")
+            self.__assert_variable_is_defined(self.queue_dest_test_trigger, "Destination queue is not specified in QUEUE_DEST_TEST_TRIGGER")
+            self.__assert_variable_is_defined(self.queue_dest_pipeline_trigger, "Destination queue is not specified in QUEUE_DEST_PIPELINE_TRIGGER")
+
 
     def display(self) -> None:
         for name in sorted(dir(self)):
@@ -58,8 +58,6 @@ class Config:
                 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

@@ -1,48 +0,0 @@
-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 review_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, url: bytes) -> None:
-        print(f"CONSUME: Received element {properties}: {len(url)}")
-
-        try:
-
-            self.__queue_dest.send_message(bytes(str(url), 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/services/db_service.py

@@ -0,0 +1,13 @@
+import os
+
+import streamlit as st
+from dotenv import load_dotenv
+from pymongo import MongoClient
+
+
+@st.cache_resource
+def init_connection():
+    load_dotenv()
+    uri = f"mongodb+srv://{os.getenv('MONGO_USERNAME')}:{os.getenv('MONGO_PASSWORD')}@{os.getenv('MONGO_HOST')}/?retryWrites=true&w=majority&appName=Cluster0"
+    client = MongoClient(uri)
+    return client.event_data

src/services/mq_service.py

@@ -0,0 +1,17 @@
+import streamlit as st
+
+from project import Config
+from project.adapters import MessageQueueOutAdapterImpl, OutAdapter
+
+
+class PrintingOutAdapter(OutAdapter):
+    def send_message(self, message):
+        print(f"Publishing message {len(message)}")
+
+
+@st.cache_resource
+def create_out_adapter(_config: Config, queue_dest: str) -> OutAdapter:
+    if _config.mq_user == "local":
+        return PrintingOutAdapter()
+    else:
+        return MessageQueueOutAdapterImpl(_config, queue_dest)

src/wrapper/__init__.py

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

src/wrapper/health_check.py

@@ -1,73 +0,0 @@
-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

@@ -42,6 +42,7 @@ class MessageQueue:
                 properties=pika.BasicProperties(
                     delivery_mode=pika.DeliveryMode.Persistent
                 ))
+            print(f"[PUBLISH] Queued message in {self.__queue_name}")
 
         # try repeating once, if connection was lost
         retry = True
@@ -60,21 +61,13 @@ class MessageQueue:
         # connection.close()
 
     def __ack_and_call_callback(
-            self,
-            ch: BlockingChannel,
-            method: spec.Basic.Deliver,
-            properties: spec.BasicProperties,
-            body: bytes) -> None:
+        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()