create minimal working Logger with Getinfo Method handling

RequestLogger.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+import os
+import re
+import sys
+from json import JSONDecodeError
+from types import SimpleNamespace
+
+class Request:
+    def __init__(self, id: int, method: str, params: dict = None) -> None:
+        self.id = id
+        self.method = method
+        self.params = params
+
+
+class Response:
+    def __init__(self, id: int, error: str = None, result: object = None) -> None:
+        self.id = id
+        self.error = error
+        self.result = result
+
+    def __str__(self) -> str:
+        try:
+            return json.dumps(
+                Response.__remove_empty_properties(self.__dict__.copy()),
+                default=lambda o: o.__dict__,
+                sort_keys=True,
+                indent=None
+            )
+        except BaseException as _:
+            return self.__repr__()
+
+    @staticmethod
+    def __remove_empty_properties(dictionary) -> dict:
+        for k, v in list(dictionary.items()):
+            if v is None:
+                del dictionary[k]
+            elif isinstance(v, dict):
+                Response.__remove_empty_properties(v)
+        return dictionary
+
+class FSCLogger:
+
+    def __init__(self: FSCLogger) -> None:
+        self.token = None
+        self.chat_id = None
+        self.base_url = None
+        self.log("starting")
+
+    @staticmethod
+    def log(message: str) -> None:
+        sys.stderr.write("[FSCLogger] {message}{linesep}".format(message=message, linesep=os.linesep))
+        sys.stdout.flush()
+
+    @staticmethod
+    def return_and_continue(response: Response) -> None:
+        if response.error:
+            # removing result in case of an error
+            response.result = None
+            FSCLogger.log("returning an error: {}".format(response.error))
+
+        FSCLogger.log("returning response: {response}{linesep}".format(response=response, linesep=os.linesep))
+        sys.stdout.write("{response}{linesep}".format(response=response, linesep=os.linesep))
+        sys.stdout.flush()
+
+    @staticmethod
+    def return_and_exit(response: Response) -> None:
+        FSCLogger.return_and_continue(response)
+
+        if response.error:
+            exit(1)
+        else:
+            exit(0)
+
+    def run(self) -> None:
+        self.log("waiting for requests")
+        for line in sys.stdin:
+            parsed_id = None
+            raw = None
+
+            # parse request identifier
+            id_match = re.match(r'.*"id":(\d+)', line)
+            if id_match:
+                parsed_id = id_match.group(1)
+
+            # parse request
+            try:
+                raw = json.loads(line.strip())
+            except JSONDecodeError as err:
+                self.return_and_exit(Response(id=parsed_id, error="failed decoding input: {}".format(err.msg)))
+
+            # map request
+            if raw:
+                request = Request(**raw)
+                self.log("received request: {}".format(vars(request)))
+                self.handle_request(request)
+            else:
+                self.log("dropped input as it couldn't be mapped to a request")
+
+        self.log("exiting")
+
+    def handle_request(self, request: Request) -> None:
+        match request.method:
+            case 'GetInfo':
+                # return information to Icinga Notifications about this channel
+                info = SimpleNamespace()
+                info.name = 'FSCLogger'
+                info.version = '0.1.0'
+                info.author = 'FSC'
+                info.config_attrs = [
+                    {
+                        "name": "logfile",
+                        "type": "string",
+                        "label": {
+                            "de_DE": "Logfile Pfad",
+                            "en_US": "Logfile path"
+                        },
+                        "help": {
+                            "de_DE": "Pfad zum Logfile",
+                            "en_US": "Path to logfile."
+                        },
+                        "required": True,
+                        "min": None,
+                        "max": None
+                    },
+                ]
+
+                self.return_and_exit(Response(id=request.id, result=info))
+
+
+# run entrypoint
+if __name__ == '__main__':
+    app = FSCLogger()
+    app.run()