feat(utils): finalize shared utility modules and tests

- Added and validated all utility modules: retry_request, request_with_timeout, track_polling_metrics, track_request_metrics, validate_data, validate_environment_variables, vault_client, setup_logger, and types
- Corrected and standardized all docstrings and type annotations
- Added full test coverage for metrics, retry, and validation utilities
- Synced and verified config_shared.py across repositories
- Fixed Pyright error in test_track_polling_metrics by using cast for Literal checks
- Confirmed no missing functions or unresolved references remain

Author: mobious999

src/app/config_shared.py

@@ -6,7 +6,7 @@
 """
 
 import json
-from typing import Any, cast
+from typing import Any
 
 from tenacity import retry, stop_after_attempt, wait_exponential
 
@@ -19,35 +19,41 @@
 
 
 def send_to_output(data: list[dict[str, Any]]) -> None:
-    """Route processed output to the configured destination.
+    """
+    Route processed output to one or more configured destinations.
 
-    Validates and dispatches messages to the configured output mode:
-    'log', 'stdout', 'queue', 'rest', 's3', or 'database'.
+    Validates and dispatches messages to each enabled output mode.
 
     Args:
         data (list[dict[str, Any]]): A list of enriched messages to route.
-
     """
     try:
         validate_list_of_dicts(data, required_keys=["text"])
 
-        mode: OutputMode = cast(OutputMode, config_shared.get_output_mode())
-
-        if mode == OutputMode.LOG:
-            _output_to_log(data)
-        elif mode == OutputMode.STDOUT:
-            _output_to_stdout(data)
-        elif mode == OutputMode.QUEUE:
-            _output_to_queue(data)
-        elif mode == OutputMode.REST:
-            _output_to_rest(data)
-        elif mode == OutputMode.S3:
-            _output_to_s3(data)
-        elif mode == OutputMode.DATABASE:
-            _output_to_database(data)
-        else:
-            logger.warning("⚠️ Unknown output mode: %s — defaulting to log.", mode)
-            _output_to_log(data)
+        # Use OUTPUT_MODES if defined, fallback to single OUTPUT_MODE
+        modes = config_shared.get_output_modes()
+        for mode_str in modes:
+            try:
+                mode = OutputMode(mode_str)
+            except ValueError:
+                logger.warning("⚠️ Unknown output mode: %s — skipping.", mode_str)
+                continue
+
+            if mode == OutputMode.LOG:
+                _output_to_log(data)
+            elif mode == OutputMode.STDOUT:
+                _output_to_stdout(data)
+            elif mode == OutputMode.QUEUE:
+                _output_to_queue(data)
+            elif mode == OutputMode.REST:
+                _output_to_rest(data)
+            elif mode == OutputMode.S3:
+                _output_to_s3(data)
+            elif mode == OutputMode.DATABASE:
+                _output_to_database(data)
+            else:
+                logger.warning("⚠️ Unhandled output mode: %s", mode)
+
     except Exception as e:
         logger.error("❌ Failed to send output: %s", e)
 

src/app/output_handler.py

@@ -20,7 +20,9 @@
 
 logger = setup_logger(__name__)
 
-REDACT_SENSITIVE_LOGS = config_shared.get_config_value("REDACT_SENSITIVE_LOGS", "true").lower() == "true"
+REDACT_SENSITIVE_LOGS = (
+    config_shared.get_config_value("REDACT_SENSITIVE_LOGS", "true").lower() == "true"
+)
 
 
 def safe_log_message(data: dict[str, Any]) -> str:

src/app/queue_sender.py

@@ -1,13 +1,13 @@
-"""The module initializes the utilities package for the application.
+"""Initialize the `utils` package for shared application utilities.
 
-Utilities included:
-- retry_request: Function for retrying operations with exponential backoff.
-- validate_data: Validates the structure and content of data.
-- track_polling_metrics: Tracks polling metrics for success and failure.
-- track_request_metrics: Tracks metrics for individual API requests.
-- request_with_timeout: Makes HTTP requests with a timeout.
-- validate_environment_variables: Validates required environment variables.
-- setup_logger: Configures logging for the application.
+Included Utilities:
+- setup_logger: Configures logging with structured output.
+- retry_request: Retries a function with optional delay on failure.
+- request_with_timeout: Makes HTTP GET requests with timeout and validation.
+- validate_data: Validates schema and batch structure of data.
+- validate_environment_variables: Ensures required environment variables are set.
+- track_polling_metrics: Logs success/failure of polling operations.
+- track_request_metrics: Logs request-level metrics (rate limits, success, etc.).
 """
 
 from .request_with_timeout import request_with_timeout
@@ -19,14 +19,14 @@
 from .validate_environment_variables import validate_environment_variables
 
 __all__ = [
+    "setup_logger",
     "retry_request",
+    "request_with_timeout",
     "validate_data",
+    "validate_environment_variables",
     "track_polling_metrics",
     "track_request_metrics",
-    "request_with_timeout",
-    "validate_environment_variables",
-    "setup_logger",
 ]
 
-# Initialize package-level logger
+# Initialize package-level logger for utilities
 logger = setup_logger(name="utils")

src/app/utils/__init__.py

@@ -0,0 +1,111 @@
+"""Healthcheck utility module for readiness and liveness probes.
+
+Provides application status flags and an optional HTTP server for use with
+container orchestrators like Kubernetes or Docker.
+"""
+
+import logging
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+from app import config_shared
+
+logger: logging.Logger = logging.getLogger(__name__)
+
+# Service status flags
+_readiness_flag: bool = False
+_health_flag: bool = True
+
+
+def is_ready() -> bool:
+    """
+    Check if the service is ready to handle requests.
+
+    Returns:
+        bool: True if the service has completed startup and is ready.
+    """
+    return _readiness_flag
+
+
+def is_healthy() -> bool:
+    """
+    Check if the service is currently healthy.
+
+    Returns:
+        bool: True if the service is healthy and not in a failure state.
+    """
+    return _health_flag
+
+
+def set_ready() -> None:
+    """
+    Mark the service as ready to handle traffic.
+    """
+    global _readiness_flag
+    _readiness_flag = True
+    logger.info("✅ Service marked as ready")
+
+
+def set_unhealthy() -> None:
+    """
+    Mark the service as unhealthy (e.g., during shutdown or failure).
+    """
+    global _health_flag
+    _health_flag = False
+    logger.warning("❌ Service marked as unhealthy")
+
+
+class HealthHandler(BaseHTTPRequestHandler):
+    """
+    HTTP request handler for /health and /ready endpoints.
+    """
+
+    def do_GET(self) -> None:
+        """
+        Handle GET requests for readiness and liveness checks.
+        """
+        if self.path == "/health":
+            status: int = 200 if is_healthy() else 500
+            self.send_response(status)
+            self.end_headers()
+            self.wfile.write(b"healthy" if status == 200 else b"unhealthy")
+
+        elif self.path == "/ready":
+            status: int = 200 if is_ready() else 503
+            self.send_response(status)
+            self.end_headers()
+            self.wfile.write(b"ready" if status == 200 else b"not ready")
+
+        else:
+            self.send_response(404)
+            self.end_headers()
+            self.wfile.write(b"not found")
+
+    def log_message(self, format: str, *args: object) -> None:
+        """
+        Suppress default access log output from BaseHTTPRequestHandler.
+        """
+        pass
+
+
+def start_health_server(port: int = 8081) -> None:
+    """
+    Start an HTTP server exposing /health and /ready endpoints.
+
+    This server runs in a background thread and is intended for use with
+    readiness and liveness probes in orchestration environments.
+
+    Args:
+        port (int): Port to bind the health server to. Defaults to 8081.
+    """
+    if not config_shared.get_healthcheck_enabled():
+        logger.info("⚠️ Healthcheck server is disabled by configuration.")
+        return
+
+    def serve() -> None:
+        with HTTPServer(("0.0.0.0", port), HealthHandler) as httpd:
+            logger.info("📡 Healthcheck server running on port %d", port)
+            httpd.serve_forever()
+
+    thread: threading.Thread = threading.Thread(target=serve, daemon=True)
+    thread.start()

src/app/utils/healthcheck.py

@@ -0,0 +1,75 @@
+"""Shared Prometheus metric definitions for all pollers and processors.
+
+Exports counters and histograms for:
+- Polling operations
+- HTTP requests
+- Output handling
+- Message processing
+"""
+
+from prometheus_client import Counter, Histogram
+
+# Output Metrics
+output_counter = Counter(
+    "output_messages_total",
+    "Total number of messages successfully sent to each output mode.",
+    ["mode"],
+)
+
+output_failures = Counter(
+    "output_failures_total", "Total number of output failures by mode.", ["mode"]
+)
+
+output_duration = Histogram(
+    "output_duration_seconds", "Time taken to send output messages by mode.", ["mode"]
+)
+
+# Polling Metrics
+poll_counter = Counter("poll_cycles_total", "Total number of polling cycles by poller.", ["poller"])
+
+poll_errors = Counter("poll_errors_total", "Total number of poller errors by poller.", ["poller"])
+
+poll_duration = Histogram(
+    "poll_duration_seconds", "Duration of polling cycles by poller.", ["poller"]
+)
+
+# HTTP Request Metrics
+http_request_counter = Counter(
+    "http_requests_total",
+    "Total number of HTTP requests by service and method.",
+    ["service", "method", "status"],
+)
+
+http_request_duration = Histogram(
+    "http_request_duration_seconds",
+    "Duration of HTTP requests by service and method.",
+    ["service", "method"],
+)
+
+# Message Processing Metrics
+process_success = Counter(
+    "message_processing_success_total", "Number of messages processed successfully.", ["processor"]
+)
+
+process_failure = Counter(
+    "message_processing_failure_total",
+    "Number of failed message processing attempts.",
+    ["processor"],
+)
+
+process_duration = Histogram(
+    "message_processing_duration_seconds", "Time taken to process messages.", ["processor"]
+)
+
+# Validation Failures
+validation_failures = Counter(
+    "message_validation_failures_total",
+    "Number of failed message validation attempts.",
+    ["processor"],
+)
+
+validation_duration = Histogram(
+    "message_validation_duration_seconds",
+    "Duration of validation checks per message.",
+    ["processor"],
+)