added parser that turns log line strings into dictionaries

src/alogamous/log_line_parser.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+
+class LogLineParser:
+    def __init__(self, header_line: str, expected_fields: list[str], seperator: str):
+        self.header_line = header_line
+        self.expected_fields = expected_fields
+        self.separator = seperator
+        self.separator_count = len(self.expected_fields) - 1
+
+    def parse(self, line):
+        if line == self.header_line:
+            return {"type": "header line", "line": line}
+        if line.count(self.separator) == self.separator_count:
+            parsed_line = {"type": "log line"}
+            separated_line = line.split(self.separator)
+            for index in range(len(self.expected_fields)):
+                parsed_line[self.expected_fields[index]] = separated_line[index]
+            return parsed_line
+        return {"type": "unstructured line", "line": line}

tests/log_line_parser_test.py

@@ -0,0 +1,45 @@
+from alogamous import log_line_parser
+
+
+def test_parse_header_line():
+    parser = log_line_parser.LogLineParser(
+        "====================================================", ["datetime", "source", "level", "message"], " - "
+    )
+    line = "===================================================="
+    assert parser.parse(line) == {"type": "header line", "line": "===================================================="}
+
+
+def test_parse_log_line():
+    parser = log_line_parser.LogLineParser(
+        "====================================================", ["datetime", "source", "level", "message"], " - "
+    )
+    line = (
+        "2024-06-20 11:00:18,172 - aiokafka.consumer.subscription_state - INFO - Updating subscribed topics to: "
+        "frozenset({'internal'})"
+    )
+    assert parser.parse(line) == {
+        "type": "log line",
+        "datetime": "2024-06-20 11:00:18,172",
+        "source": "aiokafka.consumer.subscription_state",
+        "level": "INFO",
+        "message": "Updating subscribed topics to: frozenset({'internal'})",
+    }
+
+
+def test_parse_deviant_log_line():
+    parser = log_line_parser.LogLineParser(
+        "====================================================", ["datetime", "source", "level", "message"], " - "
+    )
+    line = "Hello I am a bad log line"
+    assert parser.parse(line) == {"type": "unstructured line", "line": "Hello I am a bad log line"}
+
+
+def test_parse_start_header_content():
+    parser = log_line_parser.LogLineParser(
+        "====================================================", ["datetime", "source", "level", "message"], " - "
+    )
+    line = "    Start time: 2024-06-20 09:00:00.001550+00:00"
+    assert parser.parse(line) == {
+        "type": "unstructured line",
+        "line": "    Start time: 2024-06-20 09:00:00.001550+00:00",
+    }