PostgreSQL connection caching and reuse (#131)

Author: tmikula-dev

.github/copilot-instructions.md

@@ -16,6 +16,7 @@ Structure
 Python style
 - Python 3.13
 - Type hints for public functions and classes
+- Use type aliases for complex types
 - Use built-in generics for type hints
 - Use `logging.getLogger(__name__)`, not print
 - Lazy % formatting in logging: `logger.info("msg %s", var)`
@@ -31,6 +32,7 @@ Python style
 Patterns
 - `__init__` methods must not raise exceptions; defer validation and connection to first use (lazy init)
 - Writers: inherit from `Writer(ABC)`, implement `write(topic, message) -> (bool, str|None)` and `check_health() -> (bool, str)`
+- PostgreSQL: `WriterPostgres` and `ReaderPostgres` cache a single connection per instance
 - Route dispatch via `ROUTE_MAP` dict mapping routes to handler functions in `event_gate_lambda.py` and `event_stats_lambda.py`
 - Separate business logic from environment access (env vars, file I/O, network calls)
 - No duplicate validation; centralize parsing in one layer where practical
@@ -50,3 +52,6 @@ Testing
 Quality gates (run after changes, fix only if below threshold)
 - Run all quality gates at once: `make qa`
 - Once a quality gate passes, do not re-run it in different scenarios
+
+Git workflow
+- Do NOT create git commits; committing is the developer's responsibility

.github/dependabot.yml

@@ -14,6 +14,10 @@ updates:
     commit-message:
       prefix: "chore"
       include: "scope"
+    groups:
+      github-actions:
+        patterns:
+          - "*"
 
   - package-ecosystem: "pip"
     directory: "/"
@@ -31,3 +35,7 @@ updates:
       include: "scope"
     allow:
       - dependency-type: "direct"
+    groups:
+      python-dependencies:
+        patterns:
+          - "*"

.pylintrc

@@ -93,7 +93,7 @@ prefer-stubs=no
 
 # Minimum Python version to use for version dependent checks. Will default to
 # the version used to run pylint.
-py-version=3.11
+py-version=3.13
 
 # Discover python modules and packages in the file system subtree.
 recursive=no
@@ -440,7 +440,8 @@ disable=raw-checker-failed,
         deprecated-pragma,
         use-symbolic-message-instead,
         use-implicit-booleaness-not-comparison-to-string,
-        use-implicit-booleaness-not-comparison-to-zero
+        use-implicit-booleaness-not-comparison-to-zero,
+        useless-return
 
 # Enable the message, report, category or checker with the given id(s). You can
 # either give multiple identifier separated by comma (,) or put this option

api.yaml

@@ -56,6 +56,14 @@ paths:
                   uptime_seconds:
                     type: integer
                     example: 12345
+                  dependencies:
+                    type: object
+                    additionalProperties:
+                      type: string
+                    example:
+                      kafka: ok
+                      eventbridge: not configured
+                      postgres: ok
         '503':
           description: Service is degraded
           content:
@@ -74,6 +82,14 @@ paths:
                       eventbridge: client not initialized
                       kafka: producer not initialized
                       postgres: host not configured
+                  dependencies:
+                    type: object
+                    additionalProperties:
+                      type: string
+                    example:
+                      kafka: ok
+                      eventbridge: client not initialized
+                      postgres: host not configured
 
   /topics:
     get:

requirements.txt

@@ -11,6 +11,8 @@ jsonschema==4.26.0
 PyJWT==2.12.1
 requests==2.33.1
 boto3==1.43.2
+aiosql==15.0
+botocore==1.43.2
 confluent-kafka==2.14.0
 moto[s3,secretsmanager,events]==5.2.0
 testcontainers==4.14.2

src/event_gate_lambda.py

@@ -16,7 +16,6 @@
 
 """AWS Lambda entry point for the EventGate service."""
 
-import logging
 import os
 import sys
 from typing import Any
@@ -31,20 +30,14 @@
 from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV
 from src.utils.config_loader import load_config
 from src.utils.constants import SSL_CA_BUNDLE_KEY
+from src.utils.logging_levels import init_root_logger
 from src.utils.utils import dispatch_request
 from src.writers.writer_eventbridge import WriterEventBridge
 from src.writers.writer_kafka import WriterKafka
 from src.writers.writer_postgres import WriterPostgres
 
-# Initialize logger
-root_logger = logging.getLogger()
-if not root_logger.handlers:
-    root_logger.addHandler(logging.StreamHandler())
-
-log_level = os.environ.get("LOG_LEVEL", "INFO")
-root_logger.setLevel(log_level)
-logger = logging.getLogger(__name__)
-logger.debug("Initialized logger with level %s.", log_level)
+logger = init_root_logger(__name__)
+logger.debug("Initialized logger with level %s.", os.environ.get("LOG_LEVEL", "INFO"))
 
 # Load main configuration
 logger.debug("Using CONF_DIR=%s.", CONF_DIR)

src/event_stats_lambda.py

@@ -16,7 +16,6 @@
 
 """AWS Lambda entry point for the EventStats service."""
 
-import logging
 import os
 from typing import Any
 
@@ -25,17 +24,11 @@
 from src.readers.reader_postgres import ReaderPostgres
 from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV
 from src.utils.config_loader import load_topic_names
+from src.utils.logging_levels import init_root_logger
 from src.utils.utils import dispatch_request
 
-# Initialize logger
-root_logger = logging.getLogger()
-if not root_logger.handlers:
-    root_logger.addHandler(logging.StreamHandler())
-
-log_level = os.environ.get("LOG_LEVEL", "INFO")
-root_logger.setLevel(log_level)
-logger = logging.getLogger(__name__)
-logger.debug("Initialized EventStats logger with level %s.", log_level)
+logger = init_root_logger(__name__)
+logger.debug("Initialized EventStats logger with level %s.", os.environ.get("LOG_LEVEL", "INFO"))
 
 # Load main configuration
 logger.debug("Using CONF_DIR=%s.", CONF_DIR)

src/handlers/handler_health.py

@@ -22,16 +22,20 @@
 from collections.abc import Mapping
 from typing import Any, Protocol
 
+from src.writers.writer import HealthCheckError
+
 logger = logging.getLogger(__name__)
 
 
 class HealthCheckable(Protocol):
     """Protocol for dependencies that support health checks."""
 
-    def check_health(self) -> tuple[bool, str]:
+    def check_health(self) -> str | None:
         """Check dependency health.
         Returns:
-            Tuple of (is_healthy, message).
+            `None` when healthy, a descriptive string when intentionally disabled.
+        Raises:
+            HealthCheckError: If the dependency is unhealthy.
         """
 
 
@@ -51,11 +55,15 @@ def get_health(self) -> dict[str, Any]:
         logger.debug("Handling GET Health.")
 
         failures: dict[str, str] = {}
+        statuses: dict[str, str] = {}
 
         for name, dependency in self.dependencies.items():
-            healthy, msg = dependency.check_health()
-            if not healthy:
-                failures[name] = msg
+            try:
+                result = dependency.check_health()
+                statuses[name] = result if result else "ok"
+            except HealthCheckError as exc:
+                failures[name] = str(exc)
+                statuses[name] = str(exc)
 
         uptime_seconds = int((datetime.now(timezone.utc) - self.start_time).total_seconds())
 
@@ -64,12 +72,12 @@ def get_health(self) -> dict[str, Any]:
             return {
                 "statusCode": 200,
                 "headers": {"Content-Type": "application/json"},
-                "body": json.dumps({"status": "ok", "uptime_seconds": uptime_seconds}),
+                "body": json.dumps({"status": "ok", "uptime_seconds": uptime_seconds, "dependencies": statuses}),
             }
 
         logger.debug("Health check degraded: %s.", failures)
         return {
             "statusCode": 503,
             "headers": {"Content-Type": "application/json"},
-            "body": json.dumps({"status": "degraded", "failures": failures}),
+            "body": json.dumps({"status": "degraded", "failures": failures, "dependencies": statuses}),
         }

src/handlers/handler_stats.py

@@ -21,7 +21,7 @@
 from typing import Any
 
 from src.readers.reader_postgres import ReaderPostgres
-from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_TOPICS
+from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_STATS_TOPICS
 from src.utils.utils import build_error_response
 
 logger = logging.getLogger(__name__)
@@ -54,9 +54,9 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
         if topic_name not in self.topics:
             return build_error_response(404, "topic", f"Topic '{topic_name}' not found.")
 
-        if topic_name not in SUPPORTED_TOPICS:
+        if topic_name not in SUPPORTED_STATS_TOPICS:
             return build_error_response(
-                400, "validation", f"Stats are only supported for topics '{', '.join(SUPPORTED_TOPICS)}'."
+                400, "validation", f"Stats are only supported for topics '{', '.join(SUPPORTED_STATS_TOPICS)}'."
             )
 
         # Parse request body
@@ -90,7 +90,7 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
                 cursor=cursor,
                 limit=limit,
             )
-        except RuntimeError as exc:
+        except RuntimeError:
             logger.exception("Stats query failed for topic %s.", topic_name)
             return build_error_response(500, "database", "Stats query failed.")
 

src/handlers/handler_topic.py

@@ -29,8 +29,9 @@
 from src.handlers.handler_token import HandlerToken
 from src.utils.conf_path import CONF_DIR
 from src.utils.config_loader import TopicAccessMap, load_access_config
+from src.utils.constants import TOPIC_DLCHANGE, TOPIC_RUNS, TOPIC_TEST
 from src.utils.utils import build_error_response
-from src.writers.writer import Writer
+from src.writers.writer import WriteError, Writer
 
 logger = logging.getLogger(__name__)
 
@@ -69,11 +70,11 @@ def with_load_topic_schemas(self) -> "HandlerTopic":
         logger.debug("Loading topic schemas from %s.", topic_schemas_dir)
 
         with open(os.path.join(topic_schemas_dir, "runs.json"), "r", encoding="utf-8") as file:
-            self.topics["public.cps.za.runs"] = json.load(file)
+            self.topics[TOPIC_RUNS] = json.load(file)
         with open(os.path.join(topic_schemas_dir, "dlchange.json"), "r", encoding="utf-8") as file:
-            self.topics["public.cps.za.dlchange"] = json.load(file)
+            self.topics[TOPIC_DLCHANGE] = json.load(file)
         with open(os.path.join(topic_schemas_dir, "test.json"), "r", encoding="utf-8") as file:
-            self.topics["public.cps.za.test"] = json.load(file)
+            self.topics[TOPIC_TEST] = json.load(file)
 
         logger.debug("Loaded topic schemas successfully.")
         return self
@@ -170,9 +171,10 @@ def _post_topic_message(self, topic_name: str, topic_message: dict[str, Any], to
 
         errors = []
         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})
+            try:
+                writer.write(topic_name, topic_message)
+            except WriteError as exc:
+                errors.append({"type": writer_name, "message": str(exc)})
 
         if errors:
             return {