SQL-driven named query discovery via aiosql

.github/copilot-instructions.md

@@ -7,9 +7,15 @@ Structure
 - Main lambda (event ingestion): `src/event_gate_lambda.py`
 - Stats lambda (read-only queries): `src/event_stats_lambda.py`
 - Shared config loading: `src/utils/config_loader.py`
-- Handlers: `src/handlers/` (HandlerApi, HandlerToken, HandlerTopic, HandlerHealth, HandlerStats)
+- Handlers: `src/handlers/` (HandlerApi, HandlerToken, HandlerTopic, HandlerHealth, HandlerStats, HandlerNamedQuery)
 - Writers: `src/writers/` (inherit from `Writer` base class)
 - Readers: `src/readers/` (read-only database access for stats)
+- Named queries: `POST /stats/{topic}/query/{query_name}`; public names are discovered from `src/readers/sql/*.sql` by `src/readers/query_catalog.py`, no Python registry
+  - `jobs_detail_queries.sql`: `<name>` + `<name>_with_cursor` (paginated)
+  - `aggregation_queries.sql`: `<name>_summary` + `<name>_timeline_hourly` + `<name>_timeline_daily` (no `cursor`/`limit`)
+  - Both families accept optional `source_app` and `tenant_id` filters (exact match, `null` = no filter)
+  - Public names must not clash across families; `stats.sql` queries stay private
+  - aiosql methods are dynamic: access with `getattr(queries, "name").sql`
 - Config: `conf/config.json`, `conf/access.json`, `conf/topic_schemas/*.json`
 
 Python style
@@ -52,6 +58,8 @@ Testing
 - 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
+- Keep tests minimal: only the most direct cases per behavior; no exhaustive parametrization, no tests of dataclass defaults, or duplicates of cases covered at another layer
+- Do not repeat a unit-covered case in integration tests
 
 Quality gates (run after changes, fix only if below threshold)
 - Run all quality gates at once: `make qa`

src/handlers/handler_named_query.py

@@ -21,7 +21,6 @@
 from dataclasses import asdict, dataclass
 from typing import Any
 
-from src.readers.named_query_registry import SUPPORTED_AGGREGATION_QUERIES, SUPPORTED_JOBS_DETAIL_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_aggregations_response, build_error_response, build_success_response
@@ -92,13 +91,25 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
         if error_response := self._validate_event_path_params(topic_name, query_name):
             return error_response
 
-        if query_name in SUPPORTED_AGGREGATION_QUERIES:
+        try:
+            family = self.reader_postgres.get_query_family(query_name)
+        except RuntimeError:
+            logger.exception("Internal configuration error.", extra={"query_name": query_name})
+            return build_error_response(
+                500, "configuration", "Internal configuration error. Please contact the service owners."
+            )
+
+        if family is None:
+            logger.warning("Request rejected: query is not supported.", extra={"query_name": query_name})
+            return build_error_response(400, "validation", f"Query '{query_name}' is not supported.")
+
+        if family == "aggregation":
             return self._handle_aggregation_query(event.get("body"), query_name, topic_name)
 
         return self._handle_jobs_detail_query(event.get("body"), query_name, topic_name)
 
     def _handle_jobs_detail_query(self, body: str | None, query_name: str, topic_name: str) -> dict[str, Any]:
-        """Handle a jobs-detail named query (`SUPPORTED_JOBS_DETAIL_QUERIES`).
+        """Handle a discovered jobs-detail named query.
         Args:
             body: The raw request body (JSON string) from the API Gateway event.
             query_name: The lower-cased, already-validated `query_name` path parameter.
@@ -127,7 +138,7 @@ def _handle_jobs_detail_query(self, body: str | None, query_name: str, topic_nam
         return build_success_response(rows, pagination)
 
     def _handle_aggregation_query(self, body: str | None, query_name: str, topic_name: str) -> dict[str, Any]:
-        """Handle a window-wide aggregation named query (`SUPPORTED_AGGREGATION_QUERIES`).
+        """Handle a discovered window-wide aggregation named query.
         Args:
             body: The raw request body (JSON string) from the API Gateway event.
             query_name: The lower-cased, already-validated `query_name` path parameter.
@@ -173,9 +184,6 @@ def _validate_event_path_params(self, topic_name: str, query_name: str) -> dict[
         if not query_name:
             return build_error_response(400, "validation", "Missing path parameter 'query_name'.")
 
-        if query_name not in SUPPORTED_JOBS_DETAIL_QUERIES and query_name not in SUPPORTED_AGGREGATION_QUERIES:
-            return build_error_response(400, "validation", f"Query '{query_name}' is not supported.")
-
         return None
 
     @staticmethod

src/readers/query_catalog.py

@@ -0,0 +1,56 @@
+#
+# 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.
+#
+
+"""Discover public query groups from aiosql's family namespaces.
+
+Detail groups contain `<name>` and `<name>_with_cursor`; aggregation groups contain
+`<name>_summary`, `<name>_timeline_hourly`, and `<name>_timeline_daily`.
+Only complete groups are public. Root stats methods and generated cursor helpers are private.
+"""
+
+from typing import Literal
+
+from aiosql.queries import Queries
+
+type QueryFamily = Literal["jobs_detail", "aggregation"]
+
+_FAMILY_SUFFIXES: dict[QueryFamily, tuple[str, ...]] = {
+    "jobs_detail": ("_with_cursor", ""),
+    "aggregation": ("_summary", "_timeline_hourly", "_timeline_daily"),
+}
+
+
+def validate_query_groups(queries: Queries) -> None:
+    """Reject incomplete, unexpected, or conflicting public SQL groups."""
+    public_names: set[str] = set()
+    for family, suffixes in _FAMILY_SUFFIXES.items():
+        available = set(getattr(queries, family).available_queries)
+        names = {name.removesuffix(suffixes[0]) for name in available if name.endswith(suffixes[0])}
+        # Every authored SELECT also has an aiosql-generated `<statement>_cursor` method.
+        expected = {f"{name}{suffix}{helper}" for name in names for suffix in suffixes for helper in ("", "_cursor")}
+        if available != expected or any(not name or name.endswith("_cursor") or name != name.lower() for name in names):
+            raise ValueError(f"Invalid SQL query groups in {family}: {', '.join(sorted(available ^ expected))}.")
+        if conflicts := public_names & names:
+            raise ValueError(f"Conflicting public SQL query names: {', '.join(sorted(conflicts))}.")
+        public_names.update(names)
+
+
+def query_family(queries: Queries, query_name: str) -> QueryFamily | None:
+    """Classify a public name in a validated query tree without exposing helper methods."""
+    for family, suffixes in _FAMILY_SUFFIXES.items():
+        if f"{query_name}{suffixes[0]}" in getattr(queries, family).available_queries:
+            return family
+    return None

src/readers/reader_postgres.py

@@ -18,18 +18,19 @@
 
 import logging
 import time
-from dataclasses import dataclass
 from datetime import datetime, timezone
 from functools import cached_property
 from pathlib import Path
 from typing import Any
 
 import aiosql
 import psycopg2.extensions
+from aiosql.queries import Queries
+from aiosql.utils import SQLLoadException, SQLParseException
 from botocore.exceptions import BotoCoreError, ClientError
 
 from src.readers.aggregation_models import AggregationsWindow, RunsJobsAggregations, Timeline, TimelineEntry
-from src.readers.named_query_registry import SUPPORTED_AGGREGATION_QUERIES, SUPPORTED_JOBS_DETAIL_QUERIES
+from src.readers.query_catalog import QueryFamily, query_family, validate_query_groups
 from src.utils.constants import (
     AGGREGATION_BUCKET_THRESHOLD_MS,
     POSTGRES_DEFAULT_LIMIT,
@@ -53,19 +54,6 @@
 type AggregationQueryResult = tuple[dict[str, Any], list[dict[str, Any]]]
 
 
-@dataclass(frozen=True)
-class ReaderQueries:
-    """Typed holder for reader SQL query strings loaded via aiosql."""
-
-    get_stats: str
-    get_stats_with_cursor: str
-    get_runs_jobs_detail: str
-    get_runs_jobs_detail_with_cursor: str
-    get_runs_jobs_aggregations: str
-    get_runs_jobs_timeline_hourly: str
-    get_runs_jobs_timeline_daily: str
-
-
 class ReaderPostgres(PostgresBase):
     """Read-only Postgres accessor for run/job statistics."""
 
@@ -78,22 +66,20 @@ def _connect_options(self) -> str | None:
         return f"-c statement_timeout={POSTGRES_STATEMENT_TIMEOUT_MS} -c default_transaction_read_only=on"
 
     @cached_property
-    def _queries(self) -> ReaderQueries:
-        """Load SQL queries from the `sql/` directory via aiosql."""
-        queries = aiosql.from_path(_SQL_DIR, "psycopg2")
-        # aiosql loads SQL files and attaches query functions as dynamic attributes at runtime.
-        # pylint cannot resolve these statically.
-        return ReaderQueries(
-            get_stats=queries.get_stats.sql,  # pylint: disable=no-member
-            get_stats_with_cursor=queries.get_stats_with_cursor.sql,  # pylint: disable=no-member
-            get_runs_jobs_detail=queries.get_runs_jobs_detail.sql,  # pylint: disable=no-member
-            get_runs_jobs_detail_with_cursor=(
-                queries.get_runs_jobs_detail_with_cursor.sql  # pylint: disable=no-member
-            ),
-            get_runs_jobs_aggregations=queries.get_runs_jobs_aggregations.sql,  # pylint: disable=no-member
-            get_runs_jobs_timeline_hourly=queries.get_runs_jobs_timeline_hourly.sql,  # pylint: disable=no-member
-            get_runs_jobs_timeline_daily=queries.get_runs_jobs_timeline_daily.sql,  # pylint: disable=no-member
-        )
+    def _queries(self) -> Queries:
+        """Load native aiosql queries with separate public family namespaces."""
+        try:
+            queries = aiosql.from_path(_SQL_DIR / "stats.sql", "psycopg2")
+            queries.add_child_queries("jobs_detail", aiosql.from_path(_SQL_DIR / "jobs_detail_queries.sql", "psycopg2"))
+            queries.add_child_queries("aggregation", aiosql.from_path(_SQL_DIR / "aggregation_queries.sql", "psycopg2"))
+            validate_query_groups(queries)
+        except (OSError, ValueError, SQLLoadException, SQLParseException) as exc:
+            raise RuntimeError(f"SQL query configuration error: {exc}") from exc
+        return queries
+
+    def get_query_family(self, query_name: str) -> QueryFamily | None:
+        """Discover a public query's family without accessing database configuration."""
+        return query_family(self._queries, query_name)
 
     def _require_connection_config(self) -> PostgresConfig:
         """Load and validate the PostgreSQL connection config, or raise.
@@ -198,12 +184,12 @@ def _run_stats_query(
             with connection.cursor() as db_cursor:
                 if cursor is not None:
                     db_cursor.execute(
-                        self._queries.get_stats_with_cursor,
+                        getattr(self._queries, "get_stats_with_cursor").sql,
                         {"ts_start": ts_start, "ts_end": ts_end, "cursor_id": cursor, "lim": limit + 1},
                     )
                 else:
                     db_cursor.execute(
-                        self._queries.get_stats,
+                        getattr(self._queries, "get_stats").sql,
                         {"ts_start": ts_start, "ts_end": ts_end, "lim": limit + 1},
                     )
                 if db_cursor.description is None:
@@ -232,7 +218,7 @@ def read_named_query(
     ) -> tuple[list[QueryRow], Pagination]:
         """Execute a predefined named query with keyset pagination.
         Args:
-            query_name: Registered named query identifier (see `SUPPORTED_JOBS_DETAIL_QUERIES`).
+            query_name: Public query identifier discovered from the jobs-detail SQL file.
             timestamp_start: Start of time window in epoch milliseconds (default to *now - 7 days*).
             timestamp_end: End of time window in epoch milliseconds (default to *now*).
             cursor: Last `internal_id` from previous page (keyset pagination).
@@ -245,8 +231,7 @@ def read_named_query(
         Raises:
             RuntimeError: On unknown query name or database connectivity/query errors.
         """
-        query = SUPPORTED_JOBS_DETAIL_QUERIES.get(query_name)
-        if query is None:
+        if self.get_query_family(query_name) != "jobs_detail":
             raise RuntimeError(f"Unknown named query: {query_name}.")
 
         self._require_connection_config()
@@ -256,8 +241,9 @@ def read_named_query(
         ts_start = timestamp_start if timestamp_start is not None else (now_ms - POSTGRES_DEFAULT_WINDOW_MS)
         ts_end = timestamp_end if timestamp_end is not None else now_ms
 
-        sql = getattr(self._queries, query.sql_key)
-        sql_with_cursor = getattr(self._queries, query.sql_key_with_cursor)
+        queries = getattr(self._queries, "jobs_detail")
+        sql = getattr(queries, query_name).sql
+        sql_with_cursor = getattr(queries, f"{query_name}_with_cursor").sql
 
         try:
             col_names, raw_rows = self._execute_with_retry(
@@ -341,7 +327,7 @@ def read_aggregations(
     ) -> RunsJobsAggregations:
         """Execute a predefined window-wide aggregation named query.
         Args:
-            query_name: Registered aggregation query identifier.
+            query_name: Public query identifier discovered from the aggregation SQL file.
             timestamp_start: Start of time window in epoch milliseconds (default to *now - 7 days*).
             timestamp_end: End of time window in epoch milliseconds (default to *now*).
             source_app: Exact source application filter, or `None` for all applications.
@@ -351,8 +337,7 @@ def read_aggregations(
         Raises:
             RuntimeError: On unknown query name or database connectivity/query errors.
         """
-        query = SUPPORTED_AGGREGATION_QUERIES.get(query_name)
-        if query is None:
+        if self.get_query_family(query_name) != "aggregation":
             raise RuntimeError(f"Unknown aggregation query: {query_name}.")
 
         self._require_connection_config()
@@ -363,12 +348,10 @@ def read_aggregations(
 
         window_ms = ts_end - ts_start
         bucket = TIMELINE_BUCKET_HOUR if window_ms <= AGGREGATION_BUCKET_THRESHOLD_MS else TIMELINE_BUCKET_DAY
-        timeline_sql_key = (
-            query.timeline_hourly_sql_key if bucket == TIMELINE_BUCKET_HOUR else (query.timeline_daily_sql_key)
-        )
-
-        summary_sql = getattr(self._queries, query.summary_sql_key)
-        timeline_sql = getattr(self._queries, timeline_sql_key)
+        timeline_suffix = "timeline_hourly" if bucket == TIMELINE_BUCKET_HOUR else "timeline_daily"
+        queries = getattr(self._queries, "aggregation")
+        summary_sql = getattr(queries, f"{query_name}_summary").sql
+        timeline_sql = getattr(queries, f"{query_name}_{timeline_suffix}").sql
 
         try:
             summary_row, timeline_rows = self._execute_with_retry(