feat(event-stats): add named-query endpoint for runs/jobs detail

.github/copilot-instructions.md

@@ -51,6 +51,7 @@ Testing
 - No real API/DB calls in unit tests
 - Use `mocker.patch("module.dependency")` or `mocker.patch.object(Class, "method")`
 - Assert pattern: `assert expected == actual`
+- Pylint is disabled for `tests/`. Do not add any `# pylint: disable=...` comments in test files
 
 Quality gates (run after changes, fix only if below threshold)
 - Run all quality gates at once: `make qa`

.pylintrc

@@ -104,10 +104,6 @@ recursive=no
 # source root.
 source-roots=
 
-# When enabled, pylint would attempt to guess common misconfiguration and emit
-# user-friendly hints instead of false-positive error messages.
-suggestion-mode=yes
-
 # Allow loading of arbitrary C extensions. Extensions are imported into the
 # active Python interpreter and may run arbitrary code.
 unsafe-load-any-extension=no

DEVELOPER.md

@@ -27,6 +27,8 @@ EventGate ships two Lambda functions:
 ## Prerequisites
 - Python 3.13 (current required runtime)
 - Docker (for local integration tests using testcontainers)
+- Flyway CLI 13.x, Community edition (for local integration tests).
+  Requires a JDK 17+ (CI uses temurin 21). See [database/README.md](database/README.md).
 
 ## Set Up Python Environment
 ```shell

README.md

@@ -63,6 +63,7 @@ All responses are JSON unless otherwise noted. The POST endpoint requires a vali
 | GET | `/topics/{topicName}` | none | Returns JSON Schema for the topic |
 | POST | `/topics/{topicName}` | JWT | Validates + forwards message to configured sinks |
 | POST | `/stats/{topicName}` | none | Queries ingested events with filtering, sorting, and cursor pagination |
+| POST | `/stats/{topicName}/query/{queryName}` | none | Executes a predefined named query with cursor pagination |
 | POST | `/terminate` | (internal) | Forces Lambda process exit (used to trigger cold start & config reload) |
 
 Status codes:

api.yaml

@@ -328,6 +328,144 @@ paths:
               schema:
                 $ref: '#/components/schemas/ErrorResponse'
 
+  /stats/{topic_name}/query/{query_name}:
+    post:
+      summary: Execute a predefined named query
+      description: >
+        Runs a curated, server-side named query by name and returns its
+        paginated result set.
+      security: []
+      parameters:
+        - name: topic_name
+          in: path
+          required: true
+          schema:
+            type: string
+          description: Name of the topic to query
+        - name: query_name
+          in: path
+          required: true
+          schema:
+            type: string
+            enum: [runs_jobs_detail]
+          description: Identifier of the named query to run
+      requestBody:
+        description: Query parameters including time window, pagination cursor, and limit
+        required: false
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                timestamp_start:
+                  type: integer
+                  nullable: true
+                  description: Start of time window in epoch milliseconds (default now - 7 days)
+                  example: 1704067200000
+                timestamp_end:
+                  type: integer
+                  nullable: true
+                  description: End of time window in epoch milliseconds (default now)
+                  example: 1706745600000
+                cursor:
+                  type: integer
+                  nullable: true
+                  description: Last internal_id from previous page (keyset pagination)
+                limit:
+                  type: integer
+                  minimum: 1
+                  maximum: 1000
+                  default: 50
+                  description: Maximum number of records per page
+      responses:
+        '200':
+          description: Paginated named-query result set with computed fields
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  success:
+                    type: boolean
+                    example: true
+                  statusCode:
+                    type: integer
+                    example: 200
+                  data:
+                    type: array
+                    items:
+                      type: object
+                      properties:
+                        event_id:
+                          type: string
+                        job_ref:
+                          type: string
+                        tenant_id:
+                          type: string
+                        internal_id:
+                          type: integer
+                        catalog_id:
+                          type: string
+                        status:
+                          type: string
+                        run_date:
+                          type: string
+                          description: Date of the job-level timestamp_start (DD-MM-YYYY, UTC)
+                        run_status:
+                          type: string
+                          description: >
+                            Computed status bucket. Message reclassification is applied to
+                            every row (no data received, no data produced, timeout), else the
+                            raw status is preserved (succeeded, failed, killed, skipped).
+                        formatted_tenant:
+                          type: string
+                          description: Lowercase tenant_id
+                        elapsed_time:
+                          type: integer
+                          nullable: true
+                          description: Difference in milliseconds between job end and start
+                        start_time:
+                          type: string
+                          description: Formatted job start (YYYY-MM-DD HH:MM:SS UTC)
+                        end_time:
+                          type: string
+                          description: Formatted job end (YYYY-MM-DD HH:MM:SS UTC)
+                      additionalProperties: true
+                  pagination:
+                    type: object
+                    properties:
+                      cursor:
+                        type: integer
+                        nullable: true
+                      has_more:
+                        type: boolean
+                      limit:
+                        type: integer
+        '400':
+          description: Invalid parameters, unsupported topic, or unknown query_name
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        '403':
+          description: User not authorized for topic
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        '404':
+          description: Topic not found
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        '500':
+          description: Database query error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+
   /terminate:
     post:
       summary: Terminates lambda environment

src/event_gate_lambda.py

@@ -16,7 +16,6 @@
 
 """AWS Lambda entry point for the EventGate service."""
 
-import logging
 import sys
 import time
 from typing import Any

src/event_stats_lambda.py

@@ -16,10 +16,10 @@
 
 """AWS Lambda entry point for the EventStats service."""
 
-import logging
 import time
 from typing import Any
 
+from src.handlers.handler_named_query import HandlerNamedQuery
 from src.handlers.handler_health import HandlerHealth
 from src.handlers.handler_stats import HandlerStats
 from src.readers.reader_postgres import ReaderPostgres
@@ -45,6 +45,7 @@
 
 # Initialize EventStats handlers
 handler_stats = HandlerStats(topics, reader_postgres)
+handler_named_query = HandlerNamedQuery(topics, reader_postgres)
 handler_health = HandlerHealth({"postgres_reader": reader_postgres})
 
 logger.info(
@@ -58,6 +59,7 @@
 # Route to handler function mapping
 ROUTE_MAP: dict[str, Any] = {
     "/stats/{topic_name}": handler_stats.handle_request,
+    "/stats/{topic_name}/query/{query_name}": handler_named_query.handle_request,
     "/health": lambda _: handler_health.get_health(),
 }
 

src/handlers/handler_named_query.py

@@ -0,0 +1,152 @@
+#
+# 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.
+#
+
+"""Handler for the /stats/{topic_name}/query/{query_name} endpoint."""
+
+import json
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+from src.readers.named_query_registry import SUPPORTED_QUERIES
+from src.readers.reader_postgres import ReaderPostgres
+from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_STATS_TOPICS
+from src.utils.utils import build_error_response, build_success_response
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class NamedQueryParams:
+    """Validated query parameters ready to pass to `read_named_query`.
+
+    Attributes:
+        timestamp_start: Start of time window in epoch milliseconds, or `None` for the default.
+        timestamp_end: End of time window in epoch milliseconds, or `None` for the default.
+        cursor: Last `internal_id` from previous page, or `None` for the first page.
+        limit: Maximum number of rows per page.
+    """
+
+    timestamp_start: int | None
+    timestamp_end: int | None
+    cursor: int | None
+    limit: int
+
+
+class HandlerNamedQuery:
+    """Handle predefined named queries for a specific topic."""
+
+    def __init__(
+        self,
+        topics: dict[str, dict[str, Any]],
+        reader_postgres: ReaderPostgres,
+    ) -> None:
+        self.topics = topics
+        self.reader_postgres = reader_postgres
+
+    def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
+        """Handle POST /stats/{topic_name}/query/{query_name} requests.
+        Args:
+            event: API Gateway proxy event.
+        Returns:
+            API Gateway response dict.
+        """
+        path_params = event.get("pathParameters") or {}
+        topic_name = path_params.get("topic_name", "").lower()
+        query_name = path_params.get("query_name", "").lower()
+
+        if error_response := self._validate_event_path_params(topic_name, query_name):
+            return error_response
+
+        body_params = self._validate_event_body(event.get("body"))
+        if isinstance(body_params, dict):
+            return body_params
+
+        try:
+            rows, pagination = self.reader_postgres.read_named_query(
+                query_name=query_name,
+                timestamp_start=body_params.timestamp_start,
+                timestamp_end=body_params.timestamp_end,
+                cursor=body_params.cursor,
+                limit=body_params.limit,
+            )
+        except RuntimeError:
+            logger.exception("Named query %s failed for topic %s.", query_name, topic_name)
+            return build_error_response(500, "database", "Named query failed.")
+
+        return build_success_response(rows, pagination)
+
+    def _validate_event_path_params(self, topic_name: str, query_name: str) -> dict[str, Any] | None:
+        """Validate the `topic_name`/`query_name` path parameters.
+        Args:
+            topic_name: The lower-cased `topic_name` path parameter.
+            query_name: The lower-cased `query_name` path parameter.
+        Returns:
+            An `error_response` dict if validation fails, or `None` if valid.
+        """
+        if not topic_name:
+            return build_error_response(400, "validation", "Missing path parameter 'topic_name'.")
+
+        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_STATS_TOPICS:
+            return build_error_response(400, "validation", f"Topic '{topic_name}' is not supported.")
+
+        if not query_name:
+            return build_error_response(400, "validation", "Missing path parameter 'query_name'.")
+
+        if query_name not in SUPPORTED_QUERIES:
+            return build_error_response(400, "validation", f"Query '{query_name}' is not supported. ")
+
+        return None
+
+    @staticmethod
+    def _validate_event_body(body: str | None) -> NamedQueryParams | dict[str, Any]:
+        """Parse and validate the request body.
+        Args:
+            body: The raw request body (JSON string) from the API Gateway event, or `None`.
+        Returns:
+            The parsed_body `NamedQueryParams`, or an `error_response` dict if validation fails.
+        """
+        try:
+            parsed_body = json.loads(body or "{}")
+        except (json.JSONDecodeError, TypeError):
+            return build_error_response(400, "validation", "Request body must be valid JSON.")
+
+        if not isinstance(parsed_body, dict):
+            return build_error_response(400, "validation", "Request body must be a JSON object.")
+
+        timestamp_start = parsed_body.get("timestamp_start")
+        timestamp_end = parsed_body.get("timestamp_end")
+        cursor = parsed_body.get("cursor")
+        limit: int = parsed_body.get("limit", POSTGRES_DEFAULT_LIMIT)
+
+        int_fields = (
+            (timestamp_start, "timestamp_start"),
+            (timestamp_end, "timestamp_end"),
+            (cursor, "cursor"),
+        )
+        for value, field_name in int_fields:
+            if value is not None and (isinstance(value, bool) or not isinstance(value, int)):
+                return build_error_response(400, "validation", f"Field '{field_name}' must be an integer.")
+
+        if not isinstance(limit, int) or isinstance(limit, bool) or limit < 1:
+            return build_error_response(400, "validation", "Field 'limit' must be a positive integer.")
+
+        return NamedQueryParams(
+            timestamp_start=timestamp_start, timestamp_end=timestamp_end, cursor=cursor, limit=limit
+        )

src/handlers/handler_stats.py

@@ -24,7 +24,7 @@
 from src.readers.reader_postgres import ReaderPostgres
 from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_STATS_TOPICS
 from src.utils.observability import append_request_context
-from src.utils.utils import build_error_response, resolve_request_topic
+from src.utils.utils import build_error_response, build_success_response, resolve_request_topic
 
 logger = logging.getLogger(__name__)
 
@@ -130,16 +130,4 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
         )
         logger.debug("Stats query completed.")
 
-        return {
-            "statusCode": 200,
-            "headers": {"Content-Type": "application/json"},
-            "body": json.dumps(
-                {
-                    "success": True,
-                    "statusCode": 200,
-                    "data": rows,
-                    "pagination": pagination,
-                },
-                default=str,
-            ),
-        }
+        return build_success_response(rows, pagination)