OOP for project writers (#104)

Author: tmikula-dev

.github/workflows/check_python.yml

@@ -39,9 +39,9 @@ jobs:
           if [[ "${{ github.event_name }}" == "pull_request" ]]; then
             CHANGED_FILES=$(gh api \
               "repos/${{ github.repository }}/pulls/${{ github.event.pull_request.number }}/files" \
-              --jq '.[].filename | select(endswith(".py"))')
+              --jq '.[].filename | select(endswith(".py") or . == "requirements.txt")')
           else
-            CHANGED_FILES=$(git diff --name-only "${{ github.sha }}~1" "${{ github.sha }}" -- '*.py')
+            CHANGED_FILES=$(git diff --name-only "${{ github.sha }}~1" "${{ github.sha }}" -- '*.py' 'requirements.txt')
           fi
 
           if [[ -n "$CHANGED_FILES" ]]; then

DEVELOPER.md

@@ -1,12 +1,10 @@
-# Event Gate—for Developers
+# EventGate for Developers
 
 - [Get Started](#get-started)
 - [Set Up Python Environment](#set-up-python-environment)
 - [Run Pylint Tool Locally](#run-pylint-tool-locally)
 - [Run Black Tool Locally](#run-black-tool-locally)
 - [Run mypy Tool Locally](#run-mypy-tool-locally)
-- [Run TFLint Tool Locally](#run-tflint-tool-locally)
-- [Run Trivy Tool Locally](#run-trivy-tool-locally)
 - [Run Unit Test](#running-unit-test)
 - [Code Coverage](#code-coverage)
 
@@ -103,47 +101,8 @@ To run mypy on a specific file, follow the pattern `mypy <path_to_file>/<name_of
 Example:
 ```shell
 mypy src/writer_kafka.py
-``` 
-
-## Run TFLint Tool Locally
-
-This project uses the [TFLint](https://github.com/terraform-linters/tflint) tool for static analysis of Terraform code.
-We are forcing to eliminate **all** errors reported by TFLint. Any detected warnings and notices should be corrected as well as a best practice.
-
-- Find possible errors (like invalid instance types) for Major Cloud providers (AWS/Azure/GCP).
-- Warn about deprecated syntax, unused declarations. 
-- Enforce best practices, naming conventions.
-
-> For installation instructions, please refer to the [following link.](https://github.com/terraform-linters/tflint)
-
-### Run TFLint
-
-For running TFLint you need to be in the `terraform/` directory. From the root file run the following commands:
-```shell
-cd terraform
-tflint --init
-tflint
-cd ..
 ```
 
-## Run Trivy Tool Locally
-
-This project uses the [Trivy](https://trivy.dev/latest/) tool to scan changes for security issues and misconfigurations.
-It is an open‑source security scanner maintained by Aqua Security (AquaSec).
-
-> For installation instructions, please refer to the [following link.](https://trivy.dev/latest/getting-started/installation/)
-
-### Run Trivy
-
-For running Trivy tool locally you can execute one of following commands from the root file:
-```shell
-trivy fs . --scanners vuln,secret,misconfig,license > trivy_scan.txt # Scan the whole project with all available scans (all severities for the whole project)
-trivy fs --severity MEDIUM,HIGH,CRITICAL terraform/ > trivy_scan.txt # Show only selected severities for terraform files
-trivy config Dockerfile > trivy_scan.txt # Scan only Dockerfile
-```
-
-You can see the scan results in the `trivy_scan.txt` file located in the root directory.
-
 ## Running Unit Test
 
 Unit tests are written using pytest. To run the tests, use the following command:

src/event_gate_lambda.py

@@ -29,7 +29,9 @@
 from src.handlers.handler_health import HandlerHealth
 from src.utils.constants import SSL_CA_BUNDLE_KEY
 from src.utils.utils import build_error_response
-from src.writers import writer_eventbridge, writer_kafka, writer_postgres
+from src.writers.writer_eventbridge import WriterEventBridge
+from src.writers.writer_kafka import WriterKafka
+from src.writers.writer_postgres import WriterPostgres
 from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV
 
 PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
@@ -80,15 +82,17 @@
 handler_token = HandlerToken(config).load_public_keys()
 
 # Initialize EventGate writers
-writer_eventbridge.init(logger, config)
-writer_kafka.init(logger, config)
-writer_postgres.init(logger)
+writers = {
+    "kafka": WriterKafka(config),
+    "eventbridge": WriterEventBridge(config),
+    "postgres": WriterPostgres(config),
+}
 
 # Initialize topic handler and load topic schemas
-handler_topic = HandlerTopic(CONF_DIR, ACCESS, handler_token).load_topic_schemas()
+handler_topic = HandlerTopic(CONF_DIR, ACCESS, handler_token, writers).load_topic_schemas()
 
 # Initialize health handler
-handler_health = HandlerHealth()
+handler_health = HandlerHealth(writers)
 
 
 def get_api() -> Dict[str, Any]:

src/handlers/handler_health.py

@@ -23,7 +23,7 @@
 from datetime import datetime, timezone
 from typing import Dict, Any
 
-from src.writers import writer_eventbridge, writer_kafka, writer_postgres
+from src.writers.writer import Writer
 
 logger = logging.getLogger(__name__)
 log_level = os.environ.get("LOG_LEVEL", "INFO")
@@ -35,8 +35,9 @@ class HandlerHealth:
     HandlerHealth manages service health checks and dependency status monitoring.
     """
 
-    def __init__(self):
+    def __init__(self, writers: Dict[str, Writer]):
         self.start_time: datetime = datetime.now(timezone.utc)
+        self.writers = writers
 
     def get_health(self) -> Dict[str, Any]:
         """
@@ -51,28 +52,10 @@ def get_health(self) -> Dict[str, Any]:
 
         failures: Dict[str, str] = {}
 
-        # Check Kafka writer
-        if writer_kafka.STATE.get("producer") is None:
-            failures["kafka"] = "producer not initialized"
-
-        # Check EventBridge writer
-        eventbus_arn = writer_eventbridge.STATE.get("event_bus_arn")
-        eventbridge_client = writer_eventbridge.STATE.get("client")
-        if eventbus_arn:
-            if eventbridge_client is None:
-                failures["eventbridge"] = "client not initialized"
-
-        # Check PostgreSQL writer
-        postgres_config = writer_postgres.POSTGRES
-        if postgres_config.get("database"):
-            if not postgres_config.get("host"):
-                failures["postgres"] = "host not configured"
-            elif not postgres_config.get("user"):
-                failures["postgres"] = "user not configured"
-            elif not postgres_config.get("password"):
-                failures["postgres"] = "password not configured"
-            elif not postgres_config.get("port"):
-                failures["postgres"] = "port not configured"
+        for name, writer in self.writers.items():
+            healthy, msg = writer.check_health()
+            if not healthy:
+                failures[name] = msg
 
         uptime_seconds = int((datetime.now(timezone.utc) - self.start_time).total_seconds())
 

src/handlers/handler_topic.py

@@ -28,7 +28,7 @@
 
 from src.handlers.handler_token import HandlerToken
 from src.utils.utils import build_error_response
-from src.writers import writer_eventbridge, writer_kafka, writer_postgres
+from src.writers.writer import Writer
 
 logger = logging.getLogger(__name__)
 log_level = os.environ.get("LOG_LEVEL", "INFO")
@@ -40,10 +40,17 @@ class HandlerTopic:
     HandlerTopic manages topic schemas, access control, and message posting.
     """
 
-    def __init__(self, conf_dir: str, access_config: Dict[str, list[str]], handler_token: HandlerToken):
+    def __init__(
+        self,
+        conf_dir: str,
+        access_config: Dict[str, list[str]],
+        handler_token: HandlerToken,
+        writers: Dict[str, Writer],
+    ):
         self.conf_dir = conf_dir
         self.access_config = access_config
         self.handler_token = handler_token
+        self.writers = writers
         self.topics: Dict[str, Dict[str, Any]] = {}
 
     def load_topic_schemas(self) -> "HandlerTopic":
@@ -129,17 +136,11 @@ def post_topic_message(self, topic_name: str, topic_message: Dict[str, Any], tok
         except ValidationError as exc:
             return build_error_response(400, "validation", exc.message)
 
-        kafka_ok, kafka_err = writer_kafka.write(topic_name, topic_message)
-        eventbridge_ok, eventbridge_err = writer_eventbridge.write(topic_name, topic_message)
-        postgres_ok, postgres_err = writer_postgres.write(topic_name, topic_message)
-
         errors = []
-        if not kafka_ok:
-            errors.append({"type": "kafka", "message": kafka_err})
-        if not eventbridge_ok:
-            errors.append({"type": "eventbridge", "message": eventbridge_err})
-        if not postgres_ok:
-            errors.append({"type": "postgres", "message": postgres_err})
+        for writer_name, writer in self.writers.items():
+            ok, err = writer.write(topic_name, topic_message)
+            if not ok:
+                errors.append({"type": writer_name, "message": err})
 
         if errors:
             return {

src/writers/writer.py

@@ -0,0 +1,59 @@
+#
+# Copyright 2026 ABSA Group Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+This module provides abstract base class for all EventGate writers.
+"""
+
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Optional, Tuple
+
+
+class Writer(ABC):
+    """
+    Abstract base class for EventGate writers.
+    All writers inherit from this class and implement the write() method. Writers use lazy initialization.
+    """
+
+    def __init__(self, config: Dict[str, Any]) -> None:
+        self.config = config
+
+    @abstractmethod
+    def write(self, topic_name: str, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        """
+        Publish a message to the target system.
+
+        Args:
+            topic_name: Target writer topic (destination) name.
+            message: JSON-serializable payload to publish.
+
+        Returns:
+            Tuple of (success: bool, error_message: Optional[str]).
+            - (True, None) on success or when writer is disabled/skipped.
+            - (False, "error description") on failure.
+        """
+
+    @abstractmethod
+    def check_health(self) -> Tuple[bool, str]:
+        """
+        Check writer health and connectivity.
+
+        Returns:
+            Tuple of (is_healthy: bool, message: str).
+            - (True, "ok") - configured and working.
+            - (True, "not configured") - not configured, skipped.
+            - (False, "error message") - configured but failing.
+        """