diff --git a/Dockerfile b/Dockerfile index 5dcb254..b2d295a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -50,16 +50,18 @@ RUN \ echo "#################" && \ echo "### librdkafka ###" && \ echo "#################" && \ + # Build librdkafka from the version pinned for confluent-kafka in requirements.txt. \ + CONFLUENT_KAFKA_VERSION=$(grep ^confluent-kafka== ${LAMBDA_TASK_ROOT}/requirements.txt | cut -d "=" -f3) && \ mkdir -p /tmp/env-install-workdir/librdkafka && \ cd /tmp/env-install-workdir/librdkafka && \ - wget --ca-certificate=/etc/pki/tls/certs/ca-bundle.crt https://github.com/confluentinc/librdkafka/archive/v2.15.0.tar.gz && \ - tar -xf v2.15.0.tar.gz && \ - cd /tmp/env-install-workdir/librdkafka/librdkafka-2.15.0 && \ + wget --ca-certificate=/etc/pki/tls/certs/ca-bundle.crt https://github.com/confluentinc/librdkafka/archive/v${CONFLUENT_KAFKA_VERSION}.tar.gz && \ + tar -xf v${CONFLUENT_KAFKA_VERSION}.tar.gz && \ + cd /tmp/env-install-workdir/librdkafka/librdkafka-${CONFLUENT_KAFKA_VERSION} && \ ./configure && make && make install && \ echo "###################" && \ echo "### pip installs ###" && \ echo "###################" && \ - # requirements.txt pins the version of confluent-kafka. + # requirements.txt pins the version of confluent-kafka. \ # --no-binary confluent-kafka forces source compilation against the system librdkafka # built above, which includes Kerberos/GSSAPI support. The PyPI compiles without GSSAPI. pip install -r ${LAMBDA_TASK_ROOT}/requirements.txt --no-binary confluent-kafka && \ diff --git a/api.yaml b/api.yaml index 644dc5a..c42984b 100644 --- a/api.yaml +++ b/api.yaml @@ -332,8 +332,8 @@ paths: post: summary: Execute a predefined named query description: > - Runs a curated, server-side named query by name and returns its - paginated result set. + Runs a curated, server-side named query by name. `runs_jobs_detail` returns paginated + results. `runs_jobs_detail_aggregation` returns window totals and a bucketed timeline. security: [] parameters: - name: topic_name @@ -347,100 +347,30 @@ paths: required: true schema: type: string - enum: [runs_jobs_detail] + enum: [runs_jobs_detail, runs_jobs_detail_aggregation] description: Identifier of the named query to run requestBody: - description: Query parameters including time window, pagination cursor, and limit + description: > + Query parameters. Shape depends on `query_name`: pagination fields for + `runs_jobs_detail`, time window only for `runs_jobs_detail_aggregation`. 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 + anyOf: + - $ref: '#/components/schemas/RunsJobsDetailQueryBody' + - $ref: '#/components/schemas/RunsJobsDetailAggregationQueryBody' responses: '200': - description: Paginated named-query result set with computed fields + description: > + Named-query result. `runs_jobs_detail` returns a paginated result set; + `runs_jobs_detail_aggregation` returns window-wide aggregations instead. 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 + anyOf: + - $ref: '#/components/schemas/RunsJobsDetailResponse' + - $ref: '#/components/schemas/RunsJobsDetailAggregationResponse' '400': description: Invalid parameters, unsupported topic, or unknown query_name content: @@ -483,6 +413,182 @@ paths: components: schemas: + RunsJobsDetailQueryBody: + type: object + description: Request body for the paginated `runs_jobs_detail` named query. + 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 + RunsJobsDetailAggregationQueryBody: + type: object + description: > + Request body for the window-wide `runs_jobs_detail_aggregation` named query. + `cursor`/`limit` are not accepted here (400 if present) since this query is not paginated. + 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 + RunsJobsDetailResponse: + 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 + RunsJobsDetailAggregationResponse: + type: object + properties: + success: + type: boolean + example: true + statusCode: + type: integer + example: 200 + aggregations: + type: object + description: > + Window-wide totals, success rate, and a bucketed timeline for `runs_jobs_detail`(grain + job rows). Always fully populated (zeroed counts, empty timeline entries). + properties: + window: + type: object + properties: + timestamp_start: + type: integer + example: 1704067200000 + timestamp_end: + type: integer + example: 1706745600000 + total_runs: + type: integer + description: Count of job rows in the window (same grain as runs_jobs_detail) + example: 1840 + total_tenants: + type: integer + description: Number of distinct tenants in the window + example: 12 + total_catalogs: + type: integer + description: Number of distinct catalogs in the window + example: 44 + succeeded: + type: integer + example: 1600 + failed: + type: integer + example: 210 + success_rate: + type: number + description: succeeded / total_runs * 100, rounded to 1 decimal, 0.0 when empty + example: 86.9 + avg_duration_ms: + type: integer + nullable: true + description: Average job duration in ms over rows with both timestamps; null when none + example: 41000 + timeline: + type: object + properties: + bucket: + type: string + enum: [hour, day] + description: Bucket width - hour for windows <= 7 days, day otherwise + entries: + type: array + description: Non-empty buckets only, ordered oldest-first + items: + type: object + properties: + bucket_start: + type: integer + description: Start of the bucket in epoch milliseconds + example: 1704067200000 + total: + type: integer + example: 12 + succeeded: + type: integer + example: 10 + failed: + type: integer + example: 2 ErrorResponse: type: object properties: diff --git a/src/handlers/handler_named_query.py b/src/handlers/handler_named_query.py index 4891fe1..029500f 100644 --- a/src/handlers/handler_named_query.py +++ b/src/handlers/handler_named_query.py @@ -18,13 +18,13 @@ import json import logging -from dataclasses import dataclass +from dataclasses import asdict, dataclass from typing import Any -from src.readers.named_query_registry import SUPPORTED_QUERIES +from src.readers.named_query_registry import SUPPORTED_AGGREGATION_QUERIES, 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 +from src.utils.utils import build_aggregations_response, build_error_response, build_success_response logger = logging.getLogger(__name__) @@ -46,6 +46,19 @@ class NamedQueryParams: limit: int +@dataclass(frozen=True) +class AggregationQueryParams: + """Validated query parameters ready to pass to `read_aggregations`. + + Attributes: + timestamp_start: Start of time window in epoch milliseconds. + timestamp_end: End of time window in epoch milliseconds. + """ + + timestamp_start: int | None + timestamp_end: int | None + + class HandlerNamedQuery: """Handle predefined named queries for a specific topic.""" @@ -71,8 +84,22 @@ 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 - body_params = self._validate_event_body(event.get("body")) - if isinstance(body_params, dict): + if query_name in SUPPORTED_AGGREGATION_QUERIES: + return self._handle_aggregation_query(event.get("body"), query_name, topic_name) + + return self._handle_paginated_query(event.get("body"), query_name, topic_name) + + def _handle_paginated_query(self, body: str | None, query_name: str, topic_name: str) -> dict[str, Any]: + """Handle a paginated named query (`SUPPORTED_QUERIES`). + Args: + body: The raw request body (JSON string) from the API Gateway event. + query_name: The lower-cased, already-validated `query_name` path parameter. + topic_name: The lower-cased, already-validated `topic_name` path parameter. + Returns: + API Gateway response dict. + """ + body_params = self._validate_event_body(body) + if not isinstance(body_params, NamedQueryParams): return body_params try: @@ -89,6 +116,31 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: 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`). + Args: + body: The raw request body (JSON string) from the API Gateway event. + query_name: The lower-cased, already-validated `query_name` path parameter. + topic_name: The lower-cased, already-validated `topic_name` path parameter. + Returns: + API Gateway response dict. + """ + body_params = self._validate_aggregation_event_body(body) + if not isinstance(body_params, AggregationQueryParams): + return body_params + + try: + aggregations = self.reader_postgres.read_aggregations( + query_name=query_name, + timestamp_start=body_params.timestamp_start, + timestamp_end=body_params.timestamp_end, + ) + except RuntimeError: + logger.exception("Aggregation query %s failed for topic %s.", query_name, topic_name) + return build_error_response(500, "database", "Aggregation query failed.") + + return build_aggregations_response(asdict(aggregations)) + 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: @@ -109,8 +161,8 @@ 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_QUERIES: - return build_error_response(400, "validation", f"Query '{query_name}' is not supported. ") + if query_name not in SUPPORTED_QUERIES and query_name not in SUPPORTED_AGGREGATION_QUERIES: + return build_error_response(400, "validation", f"Query '{query_name}' is not supported.") return None @@ -150,3 +202,40 @@ def _validate_event_body(body: str | None) -> NamedQueryParams | dict[str, Any]: return NamedQueryParams( timestamp_start=timestamp_start, timestamp_end=timestamp_end, cursor=cursor, limit=limit ) + + @staticmethod + def _validate_aggregation_event_body(body: str | None) -> AggregationQueryParams | dict[str, Any]: + """Parse and validate the request body for an aggregation named query. + Args: + body: The raw request body (JSON string) from the API Gateway event. + Returns: + The parsed_body `AggregationQueryParams`, 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.") + + unsupported_fields = [field_name for field_name in ("cursor", "limit") if field_name in parsed_body] + if unsupported_fields: + return build_error_response( + 400, + "validation", + f"Field(s) {', '.join(unsupported_fields)} are not supported for aggregation queries.", + ) + + timestamp_start = parsed_body.get("timestamp_start") + timestamp_end = parsed_body.get("timestamp_end") + + int_fields = ( + (timestamp_start, "timestamp_start"), + (timestamp_end, "timestamp_end"), + ) + 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.") + + return AggregationQueryParams(timestamp_start=timestamp_start, timestamp_end=timestamp_end) diff --git a/src/readers/aggregation_models.py b/src/readers/aggregation_models.py new file mode 100644 index 0000000..b9e7a8b --- /dev/null +++ b/src/readers/aggregation_models.py @@ -0,0 +1,97 @@ +# +# 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. +# + +"""Typed result shapes for window-wide named-query aggregations. + +These dataclasses are built by `ReaderPostgres.read_aggregations` and converted to a plain +dict (via `dataclasses.asdict`) only at the handler boundary, so the reader's internal +aggregation logic stays fully typed end to end. +""" + +from dataclasses import dataclass, field + +from src.utils.constants import TIMELINE_BUCKET_HOUR + + +@dataclass(frozen=True) +class AggregationsWindow: + """The time window an aggregation result covers. + + Attributes: + timestamp_start: Start of the window in epoch milliseconds. + timestamp_end: End of the window in epoch milliseconds. + """ + + timestamp_start: int + timestamp_end: int + + +@dataclass(frozen=True) +class TimelineEntry: + """A single non-empty timeline bucket. + + Attributes: + bucket_start: Start of the bucket in epoch milliseconds. + total: Number of rows in the bucket. + succeeded: Number of rows with raw status `succeeded`. + failed: Number of rows with raw status `failed`. + """ + + bucket_start: int + total: int + succeeded: int + failed: int + + +@dataclass(frozen=True) +class Timeline: + """Bucketed heatmap data for a window, oldest first, empty buckets skipped. + + Attributes: + bucket: Bucket width, `"hour"` or `"day"`. + entries: Non-empty buckets ordered oldest-first. + """ + + bucket: str + entries: list[TimelineEntry] + + +@dataclass(frozen=True) +class RunsJobsAggregations: + """Window-wide summary and timeline aggregations for `runs_jobs_detail_aggregation`. + + Attributes: + window: The time window the aggregations cover. + total_runs: Count of job rows in the window (same grain as `runs_jobs_detail`). + total_tenants: Number of distinct tenants in the window. + total_catalogs: Number of distinct catalogs in the window. + succeeded: Count of rows with raw status `succeeded`. + failed: Count of rows with raw status `failed`. + success_rate: `succeeded / total_runs * 100`, rounded to 1 decimal, `0.0` when empty. + avg_duration_ms: Average job duration in milliseconds, or `None` when no row has + both `timestamp_start` and `timestamp_end`. + timeline: Heatmap-ready bucketed data for the window. + """ + + window: AggregationsWindow + total_runs: int = 0 + total_tenants: int = 0 + total_catalogs: int = 0 + succeeded: int = 0 + failed: int = 0 + success_rate: float = 0.0 + avg_duration_ms: int | None = None + timeline: Timeline = field(default_factory=lambda: Timeline(bucket=TIMELINE_BUCKET_HOUR, entries=[])) diff --git a/src/readers/named_query_registry.py b/src/readers/named_query_registry.py index 62ed004..2ca35b5 100644 --- a/src/readers/named_query_registry.py +++ b/src/readers/named_query_registry.py @@ -16,20 +16,21 @@ """Registry of predefined, server-executed named queries. -Each named query maps a stable `query_name` to the aiosql query keys used to -run it (with and without a keyset-pagination cursor). Unknown query names are -rejected at the handler layer with a `400` response. New aggregation queries are -added here without introducing new routes. +`SUPPORTED_QUERIES` maps each paginated `query_name` to its aiosql query keys, with and +without a keyset-pagination cursor. `SUPPORTED_AGGREGATION_QUERIES` holds window-wide +aggregation queries instead, which have no pagination and a different response shape. +Names absent from both are rejected as a `400` at the handler layer. New named queries +are added here, not as new routes. """ from dataclasses import dataclass -from src.utils.constants import QUERY_RUNS_JOBS_DETAIL +from src.utils.constants import QUERY_RUNS_JOBS_DETAIL, QUERY_RUNS_JOBS_DETAIL_AGGREGATION @dataclass(frozen=True) class NamedQuery: - """Definition of a single named query. + """Definition of a single paginated named query. Attributes: name: Stable identifier used in the request path. @@ -49,3 +50,30 @@ class NamedQuery: sql_key_with_cursor="get_runs_jobs_detail_with_cursor", ), } + + +@dataclass(frozen=True) +class AggregationQuery: + """Definition of a single window-wide aggregation named query. + + Attributes: + name: Stable identifier used in the request path. + summary_sql_key: aiosql query name for the single-row summary aggregates. + timeline_hourly_sql_key: aiosql query name for hourly timeline buckets. + timeline_daily_sql_key: aiosql query name for daily timeline buckets. + """ + + name: str + summary_sql_key: str + timeline_hourly_sql_key: str + timeline_daily_sql_key: str + + +SUPPORTED_AGGREGATION_QUERIES: dict[str, AggregationQuery] = { + QUERY_RUNS_JOBS_DETAIL_AGGREGATION: AggregationQuery( + name=QUERY_RUNS_JOBS_DETAIL_AGGREGATION, + summary_sql_key="get_runs_jobs_aggregations", + timeline_hourly_sql_key="get_runs_jobs_timeline_hourly", + timeline_daily_sql_key="get_runs_jobs_timeline_daily", + ), +} diff --git a/src/readers/reader_postgres.py b/src/readers/reader_postgres.py index 9823909..60ea68f 100644 --- a/src/readers/reader_postgres.py +++ b/src/readers/reader_postgres.py @@ -28,15 +28,19 @@ import psycopg2.extensions from botocore.exceptions import BotoCoreError, ClientError -from src.readers.named_query_registry import SUPPORTED_QUERIES +from src.readers.aggregation_models import AggregationsWindow, RunsJobsAggregations, Timeline, TimelineEntry +from src.readers.named_query_registry import SUPPORTED_AGGREGATION_QUERIES, SUPPORTED_QUERIES from src.utils.constants import ( + AGGREGATION_BUCKET_THRESHOLD_MS, POSTGRES_DEFAULT_LIMIT, POSTGRES_DEFAULT_WINDOW_MS, POSTGRES_MAX_LIMIT, POSTGRES_STATEMENT_TIMEOUT_MS, REQUIRED_CONNECTION_FIELDS, + TIMELINE_BUCKET_DAY, + TIMELINE_BUCKET_HOUR, ) -from src.utils.postgres_base import PsycopgError, PostgresBase +from src.utils.postgres_base import PostgresConfig, PsycopgError, PostgresBase from src.utils.utils import Pagination, QueryRow from src.writers.writer import HealthCheckError @@ -46,6 +50,7 @@ type DbRow = tuple[Any, ...] type RawQueryResult = tuple[list[str], list[DbRow]] +type AggregationQueryResult = tuple[dict[str, Any], list[dict[str, Any]]] @dataclass(frozen=True) @@ -56,6 +61,9 @@ class ReaderQueries: 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): @@ -82,8 +90,32 @@ def _queries(self) -> ReaderQueries: 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 _require_connection_config(self) -> PostgresConfig: + """Load and validate the PostgreSQL connection config, or raise. + Returns: + The validated connection config dict. + Raises: + RuntimeError: On config retrieval errors or missing required fields. + """ + try: + config = self._pg_config + except (BotoCoreError, ClientError, ValueError, KeyError) as exc: + raise RuntimeError(f"PostgreSQL configuration error: {exc}") from exc + + if not config.get("database"): + raise RuntimeError("PostgreSQL config missing: database.") + + missing = [field for field in REQUIRED_CONNECTION_FIELDS if not config.get(field)] + if missing: + raise RuntimeError(f"PostgreSQL config missing: {', '.join(missing)}.") + + return config + def read_stats( self, timestamp_start: int | None = None, @@ -105,17 +137,7 @@ def read_stats( Raises: RuntimeError: On database connectivity or query errors. """ - try: - config = self._pg_config - except (BotoCoreError, ClientError, ValueError, KeyError) as exc: - raise RuntimeError(f"PostgreSQL configuration error: {exc}") from exc - - if not config.get("database"): - raise RuntimeError("PostgreSQL config missing: database.") - - missing = [field for field in REQUIRED_CONNECTION_FIELDS if not config.get(field)] - if missing: - raise RuntimeError(f"PostgreSQL config missing: {', '.join(missing)}.") + self._require_connection_config() limit = max(1, min(limit, POSTGRES_MAX_LIMIT)) now_ms = int(time.time() * 1000) @@ -223,17 +245,7 @@ def read_named_query( if query is None: raise RuntimeError(f"Unknown named query: {query_name}.") - try: - config = self._pg_config - except (BotoCoreError, ClientError, ValueError, KeyError) as exc: - raise RuntimeError(f"PostgreSQL configuration error: {exc}") from exc - - if not config.get("database"): - raise RuntimeError("PostgreSQL config missing: database.") - - missing = [field for field in REQUIRED_CONNECTION_FIELDS if not config.get(field)] - if missing: - raise RuntimeError(f"PostgreSQL config missing: {', '.join(missing)}.") + self._require_connection_config() limit = max(1, min(limit, POSTGRES_MAX_LIMIT)) now_ms = int(time.time() * 1000) @@ -309,6 +321,126 @@ def _run_named_query( self._close_connection() return col_names, raw_rows + def read_aggregations( + self, + query_name: str, + timestamp_start: int | None = None, + timestamp_end: int | None = None, + ) -> RunsJobsAggregations: + """Execute a predefined window-wide aggregation named query. + Args: + query_name: Registered aggregation query identifier. + 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*). + Returns: + `RunsJobsAggregations` with window-wide totals, success rate, and a bucketed timeline. + Raises: + RuntimeError: On unknown query name or database connectivity/query errors. + """ + query = SUPPORTED_AGGREGATION_QUERIES.get(query_name) + if query is None: + raise RuntimeError(f"Unknown aggregation query: {query_name}.") + + self._require_connection_config() + + now_ms = int(time.time() * 1000) + 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 + + 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) + + try: + summary_row, timeline_rows = self._execute_with_retry( + lambda conn: self._run_aggregation_queries(conn, summary_sql, timeline_sql, ts_start, ts_end) + ) + except PsycopgError as exc: + self._close_connection() + raise RuntimeError(f"Database aggregation query error: {exc}") from exc + + aggregations = self._build_aggregations(ts_start, ts_end, bucket, summary_row, timeline_rows) + + logger.debug("Aggregation query %s returned %d timeline buckets.", query_name, len(timeline_rows)) + return aggregations + + def _run_aggregation_queries( + self, + connection: psycopg2.extensions.connection, + summary_sql: str, + timeline_sql: str, + ts_start: int, + ts_end: int, + ) -> AggregationQueryResult: + """Execute the summary and timeline aggregation queries on the same connection.""" + try: + with connection.cursor() as db_cursor: + # Summary: single aggregate row for the whole window. + db_cursor.execute(summary_sql, {"ts_start": ts_start, "ts_end": ts_end}) + if db_cursor.description is None: + raise RuntimeError("Aggregation query returned no result description.") + summary_cols = [desc[0] for desc in db_cursor.description] + summary_values = db_cursor.fetchone() + summary_row = dict(zip(summary_cols, summary_values, strict=True)) if summary_values else {} + + # Timeline: bucketed rows for the heatmap. + db_cursor.execute(timeline_sql, {"ts_start": ts_start, "ts_end": ts_end}) + if db_cursor.description is None: + raise RuntimeError("Timeline query returned no result description.") + timeline_cols = [desc[0] for desc in db_cursor.description] + timeline_rows = [dict(zip(timeline_cols, row, strict=True)) for row in db_cursor.fetchall()] + finally: + # PostgreSQL (psycopg2) wraps every statement, including SELECT, in an implicit transaction. + # Without an explicit commit or rollback, the connection stays "idle in transaction". + try: + connection.rollback() + except PsycopgError: + logger.debug("Failed to close the implicit transaction. Closing cached connection.", exc_info=True) + self._close_connection() + return summary_row, timeline_rows + + @staticmethod + def _build_aggregations( + ts_start: int, + ts_end: int, + bucket: str, + summary_row: dict[str, Any], + timeline_rows: list[dict[str, Any]], + ) -> RunsJobsAggregations: + """Shape raw summary/timeline rows into a typed `RunsJobsAggregations`.""" + total_runs = summary_row.get("total_runs") or 0 + succeeded = summary_row.get("succeeded") or 0 + failed = summary_row.get("failed") or 0 + avg_duration_ms = summary_row.get("avg_duration_ms") + + return RunsJobsAggregations( + window=AggregationsWindow(timestamp_start=ts_start, timestamp_end=ts_end), + total_runs=total_runs, + total_tenants=summary_row.get("total_tenants") or 0, + total_catalogs=summary_row.get("total_catalogs") or 0, + succeeded=succeeded, + failed=failed, + success_rate=round((succeeded / total_runs) * 100, 1) if total_runs else 0.0, + avg_duration_ms=round(avg_duration_ms) if avg_duration_ms is not None else None, + timeline=Timeline( + bucket=bucket, + entries=[ + TimelineEntry( + bucket_start=int(row["bucket_start"]), + total=row["total"], + succeeded=row["succeeded"], + failed=row["failed"], + ) + for row in timeline_rows + ], + ), + ) + @staticmethod def _format_runs_jobs_detail_row(row: dict[str, Any]) -> dict[str, Any]: """Add computed columns to a `runs_jobs_detail` result row. diff --git a/src/readers/sql/named_queries.sql b/src/readers/sql/named_queries.sql index e69c0a0..6ce86ea 100644 --- a/src/readers/sql/named_queries.sql +++ b/src/readers/sql/named_queries.sql @@ -24,3 +24,42 @@ SELECT r.event_id, r.job_ref, r.tenant_id, r.source_app, AND j.internal_id < :cursor_id ORDER BY j.internal_id DESC LIMIT :lim; + +-- name: get_runs_jobs_aggregations(ts_start, ts_end) +-- Window-wide summary aggregates for runs_jobs_detail (job-row grain). +SELECT COUNT(*) AS total_runs, + COUNT(DISTINCT r.tenant_id) AS total_tenants, + COUNT(DISTINCT j.catalog_id) AS total_catalogs, + COUNT(*) FILTER (WHERE j.status = 'succeeded') AS succeeded, + COUNT(*) FILTER (WHERE j.status = 'failed') AS failed, + AVG(j.timestamp_end - j.timestamp_start) + FILTER (WHERE j.timestamp_start IS NOT NULL AND j.timestamp_end IS NOT NULL) AS avg_duration_ms + FROM public_cps_za_runs_jobs j + INNER JOIN public_cps_za_runs r ON j.event_id = r.event_id + WHERE r.timestamp_start >= :ts_start AND r.timestamp_start <= :ts_end; + +-- name: get_runs_jobs_timeline_hourly(ts_start, ts_end) +-- Hourly timeline buckets (non-empty buckets only) for runs_jobs_detail, oldest first. +SELECT (EXTRACT(EPOCH FROM date_trunc('hour', to_timestamp(r.timestamp_start / 1000.0))) * 1000)::bigint + AS bucket_start, + COUNT(*) AS total, + COUNT(*) FILTER (WHERE j.status = 'succeeded') AS succeeded, + COUNT(*) FILTER (WHERE j.status = 'failed') AS failed + FROM public_cps_za_runs_jobs j + INNER JOIN public_cps_za_runs r ON j.event_id = r.event_id + WHERE r.timestamp_start >= :ts_start AND r.timestamp_start <= :ts_end + GROUP BY bucket_start + ORDER BY bucket_start ASC; + +-- name: get_runs_jobs_timeline_daily(ts_start, ts_end) +-- Daily timeline buckets (non-empty buckets only) for runs_jobs_detail, oldest first. +SELECT (EXTRACT(EPOCH FROM date_trunc('day', to_timestamp(r.timestamp_start / 1000.0))) * 1000)::bigint + AS bucket_start, + COUNT(*) AS total, + COUNT(*) FILTER (WHERE j.status = 'succeeded') AS succeeded, + COUNT(*) FILTER (WHERE j.status = 'failed') AS failed + FROM public_cps_za_runs_jobs j + INNER JOIN public_cps_za_runs r ON j.event_id = r.event_id + WHERE r.timestamp_start >= :ts_start AND r.timestamp_start <= :ts_end + GROUP BY bucket_start + ORDER BY bucket_start ASC; diff --git a/src/utils/constants.py b/src/utils/constants.py index ec1ac01..ea140b6 100644 --- a/src/utils/constants.py +++ b/src/utils/constants.py @@ -44,3 +44,9 @@ # Named query constants QUERY_RUNS_JOBS_DETAIL = "runs_jobs_detail" +QUERY_RUNS_JOBS_DETAIL_AGGREGATION = "runs_jobs_detail_aggregation" + +# Named query aggregations +AGGREGATION_BUCKET_THRESHOLD_MS = 7 * 24 * 60 * 60 * 1000 # 7 days in milliseconds +TIMELINE_BUCKET_HOUR = "hour" +TIMELINE_BUCKET_DAY = "day" diff --git a/src/utils/utils.py b/src/utils/utils.py index a779015..27cc37b 100644 --- a/src/utils/utils.py +++ b/src/utils/utils.py @@ -39,28 +39,40 @@ class Pagination(TypedDict): limit: int -def build_error_response(status: int, err_type: str, message: str) -> dict[str, Any]: - """Build a standardized JSON error response body. +def _build_json_response(status: int, payload: dict[str, Any]) -> dict[str, Any]: + """Build a standardized API Gateway Lambda Proxy response envelope. Args: status: HTTP status code. - err_type: A short error classifier (e.g. `auth`, `validation`). - message: Human readable error description. + payload: The JSON-serializable response body. Returns: A dictionary compatible with API Gateway Lambda Proxy integration. """ return { "statusCode": status, "headers": {"Content-Type": "application/json"}, - "body": json.dumps( - { - "success": False, - "statusCode": status, - "errors": [{"type": err_type, "message": message}], - } - ), + "body": json.dumps(payload, default=str), } +def build_error_response(status: int, err_type: str, message: str) -> dict[str, Any]: + """Build a standardized JSON error response body. + Args: + status: HTTP status code. + err_type: A short error classifier (e.g. `auth`, `validation`). + message: Human readable error description. + Returns: + A dictionary compatible with API Gateway Lambda Proxy integration. + """ + return _build_json_response( + status, + { + "success": False, + "statusCode": status, + "errors": [{"type": err_type, "message": message}], + }, + ) + + def build_success_response(data: list[QueryRow], pagination: Pagination) -> dict[str, Any]: """Build a standardized JSON success response body for a paginated query result. Args: @@ -69,19 +81,32 @@ def build_success_response(data: list[QueryRow], pagination: Pagination) -> dict Returns: A dictionary compatible with API Gateway Lambda Proxy integration. """ - return { - "statusCode": 200, - "headers": {"Content-Type": "application/json"}, - "body": json.dumps( - { - "success": True, - "statusCode": 200, - "data": data, - "pagination": pagination, - }, - default=str, - ), - } + return _build_json_response( + 200, + { + "success": True, + "statusCode": 200, + "data": data, + "pagination": pagination, + }, + ) + + +def build_aggregations_response(aggregations: dict[str, Any]) -> dict[str, Any]: + """Build a standardized JSON success response body for a window-wide aggregation result. + Args: + aggregations: The aggregation result, already converted to a plain dict. + Returns: + A dictionary compatible with API Gateway Lambda Proxy integration. + """ + return _build_json_response( + 200, + { + "success": True, + "statusCode": 200, + "aggregations": aggregations, + }, + ) def resolve_request_topic(event: dict[str, Any]) -> tuple[str, dict[str, Any] | None]: diff --git a/tests/integration/test_named_query_endpoint.py b/tests/integration/test_named_query_endpoint.py index 48730de..51dd7a7 100644 --- a/tests/integration/test_named_query_endpoint.py +++ b/tests/integration/test_named_query_endpoint.py @@ -128,6 +128,46 @@ def test_message_buckets_derived(self, stats_client: EventStatsTestClient) -> No assert {"timeout", "no data received", "no data produced"}.issubset(statuses) +class TestNamedQueryAggregation: + """End-to-end for the dedicated runs_jobs_detail_aggregation named query.""" + + _AGG_QUERY = "runs_jobs_detail_aggregation" + + @pytest.fixture(scope="class", autouse=True) + def seed_events(self, eventgate_client: EventGateTestClient, valid_token: str) -> None: + """Seed a mix of succeeded/failed jobs to aggregate over.""" + _post_job(eventgate_client, valid_token, "succeeded", None, "cq-aggregation-test") + _post_job(eventgate_client, valid_token, "succeeded", None, "cq-aggregation-test") + _post_job(eventgate_client, valid_token, "failed", "boom", "cq-aggregation-test") + + def test_returns_populated_aggregations(self, stats_client: EventStatsTestClient) -> None: + """Test that totals/success_rate/timeline reflect the seeded jobs, with no data/pagination keys.""" + response = stats_client.post_named_query("public.cps.za.runs", self._AGG_QUERY, {}) + + assert 200 == response["statusCode"] + body = json.loads(response["body"]) + assert True is body["success"] + assert "aggregations" in body + assert "data" not in body + assert "pagination" not in body + + aggregations = body["aggregations"] + assert aggregations["total_runs"] >= 3 + assert aggregations["succeeded"] >= 2 + assert aggregations["failed"] >= 1 + timeline = aggregations["timeline"] + assert timeline["bucket"] in ("hour", "day") + assert sum(entry["total"] for entry in timeline["entries"]) == aggregations["total_runs"] + + def test_cursor_or_limit_returns_400(self, stats_client: EventStatsTestClient) -> None: + """Test that cursor/limit are rejected for the aggregation query (not silently ignored).""" + response = stats_client.post_named_query("public.cps.za.runs", self._AGG_QUERY, {"limit": 10}) + + assert 400 == response["statusCode"] + body = json.loads(response["body"]) + assert "limit" in body["errors"][0]["message"] + + class TestNamedQueryPagination: """Keyset pagination for runs_jobs_detail.""" diff --git a/tests/unit/handlers/test_handler_named_query.py b/tests/unit/handlers/test_handler_named_query.py index 5952404..9e16cf1 100644 --- a/tests/unit/handlers/test_handler_named_query.py +++ b/tests/unit/handlers/test_handler_named_query.py @@ -21,7 +21,8 @@ import pytest -from src.handlers.handler_named_query import HandlerNamedQuery, NamedQueryParams +from src.handlers.handler_named_query import AggregationQueryParams, HandlerNamedQuery, NamedQueryParams +from src.readers.aggregation_models import AggregationsWindow, RunsJobsAggregations @pytest.fixture @@ -32,6 +33,9 @@ def mock_reader() -> MagicMock: [{"event_id": "ev1", "internal_id": 1, "status": "succeeded", "run_status": "succeeded"}], {"cursor": None, "has_more": False, "limit": 50}, ) + mock.read_aggregations.return_value = RunsJobsAggregations( + window=AggregationsWindow(timestamp_start=1000, timestamp_end=2000), total_runs=1, succeeded=1 + ) return mock @@ -195,6 +199,74 @@ def test_database_error_returns_500(self, handler: HandlerNamedQuery, mock_reade assert "database" == body["errors"][0]["type"] +class TestHandlerAggregationQuery: + """Tests for the window-wide aggregation branch (`runs_jobs_detail_aggregation`).""" + + def test_returns_200_with_aggregations(self, handler: HandlerNamedQuery, mock_reader: MagicMock) -> None: + """Test that a supported aggregation query returns 200 with an `aggregations` key, not `data`/`pagination`.""" + response = handler.handle_request(_make_event(query_name="runs_jobs_detail_aggregation")) + + assert 200 == response["statusCode"] + body = json.loads(response["body"]) + assert True is body["success"] + assert "aggregations" in body + assert "data" not in body + assert "pagination" not in body + assert 1 == body["aggregations"]["total_runs"] + mock_reader.read_aggregations.assert_called_once_with( + query_name="runs_jobs_detail_aggregation", timestamp_start=None, timestamp_end=None + ) + + def test_cursor_or_limit_in_body_returns_400(self, handler: HandlerNamedQuery) -> None: + """Test that `cursor`/`limit` are rejected (400) instead of silently ignored for aggregation queries.""" + response = handler.handle_request( + _make_event(query_name="runs_jobs_detail_aggregation", body={"cursor": 1, "limit": 10}) + ) + + assert 400 == response["statusCode"] + body = json.loads(response["body"]) + assert "cursor" in body["errors"][0]["message"] + assert "limit" in body["errors"][0]["message"] + + def test_database_error_returns_500(self, handler: HandlerNamedQuery, mock_reader: MagicMock) -> None: + """Test that a RuntimeError from `read_aggregations` returns 500.""" + mock_reader.read_aggregations.side_effect = RuntimeError("boom") + + response = handler.handle_request(_make_event(query_name="runs_jobs_detail_aggregation")) + + assert 500 == response["statusCode"] + body = json.loads(response["body"]) + assert "database" == body["errors"][0]["type"] + + +class TestValidateAggregationEventBody: + """Tests for `HandlerNamedQuery._validate_aggregation_event_body`.""" + + @pytest.mark.parametrize( + "body, expected_message_fragment", + [ + pytest.param("not json", "valid JSON", id="invalid_json"), + pytest.param("[1, 2, 3]", "JSON object", id="non_dict_json"), + pytest.param(json.dumps({"timestamp_start": "bad"}), "timestamp_start", id="invalid_timestamp_start"), + ], + ) + def test_returns_error_response_for_invalid_body(self, body: str, expected_message_fragment: str) -> None: + """Test that each invalid body returns a 400 with the expected message.""" + result = HandlerNamedQuery._validate_aggregation_event_body(body) + + assert isinstance(result, dict) + assert 400 == result["statusCode"] + assert expected_message_fragment in json.loads(result["body"])["errors"][0]["message"] + + def test_returns_parsed_params_for_valid_body(self) -> None: + """Test that a valid body with only `timestamp_start`/`timestamp_end` parses correctly.""" + result = HandlerNamedQuery._validate_aggregation_event_body( + json.dumps({"timestamp_start": 1000, "timestamp_end": 2000}) + ) + + assert AggregationQueryParams(1000, 2000) == result + + class TestValidateEventPathParams: """Parametrized tests for `HandlerNamedQuery._validate_event_path_params`.""" diff --git a/tests/unit/readers/test_aggregation_models.py b/tests/unit/readers/test_aggregation_models.py new file mode 100644 index 0000000..8c6aafa --- /dev/null +++ b/tests/unit/readers/test_aggregation_models.py @@ -0,0 +1,41 @@ +# +# 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. +# + + +from src.readers.aggregation_models import AggregationsWindow, RunsJobsAggregations + + +def test_defaults_are_zeroed_not_missing() -> None: + aggregations = RunsJobsAggregations(window=AggregationsWindow(timestamp_start=1000, timestamp_end=2000)) + + assert 0 == aggregations.total_runs + assert 0 == aggregations.total_tenants + assert 0 == aggregations.total_catalogs + assert 0 == aggregations.succeeded + assert 0 == aggregations.failed + assert 0.0 == aggregations.success_rate + assert aggregations.avg_duration_ms is None + assert "hour" == aggregations.timeline.bucket + assert [] == aggregations.timeline.entries + + +def test_default_timeline_instances_are_independent() -> None: + """Each default `RunsJobsAggregations` gets its own `Timeline`/`entries` list, not a shared mutable default.""" + first = RunsJobsAggregations(window=AggregationsWindow(timestamp_start=1000, timestamp_end=2000)) + second = RunsJobsAggregations(window=AggregationsWindow(timestamp_start=3000, timestamp_end=4000)) + + assert first.timeline is not second.timeline + assert first.timeline.entries is not second.timeline.entries diff --git a/tests/unit/readers/test_reader_named_query.py b/tests/unit/readers/test_reader_named_query.py index ddb62ed..5c266c1 100644 --- a/tests/unit/readers/test_reader_named_query.py +++ b/tests/unit/readers/test_reader_named_query.py @@ -251,6 +251,154 @@ def test_unknown_query_name_raises_runtime_error(self, reader: ReaderPostgres, p reader.read_named_query(query_name="does_not_exist") +class TestReadAggregations: + """Tests for read_aggregations execution.""" + + def test_build_aggregations_shapes_populated_rows(self) -> None: + """Test that _build_aggregations shapes summary/timeline rows into a typed result.""" + summary_row = { + "total_runs": 2, + "total_tenants": 1, + "total_catalogs": 1, + "succeeded": 1, + "failed": 1, + "avg_duration_ms": 1800000.0, + } + timeline_rows = [{"bucket_start": 1704067200000, "total": 2, "succeeded": 1, "failed": 1}] + + aggregations = ReaderPostgres._build_aggregations( + ts_start=1704067200000, + ts_end=1704070800000, + bucket="hour", + summary_row=summary_row, + timeline_rows=timeline_rows, + ) + + assert 1704067200000 == aggregations.window.timestamp_start + assert 1704070800000 == aggregations.window.timestamp_end + assert 2 == aggregations.total_runs + assert 1 == aggregations.total_tenants + assert 1 == aggregations.total_catalogs + assert 1 == aggregations.succeeded + assert 1 == aggregations.failed + assert 50.0 == aggregations.success_rate + assert 1800000 == aggregations.avg_duration_ms + assert "hour" == aggregations.timeline.bucket + assert 1 == len(aggregations.timeline.entries) + assert 1704067200000 == aggregations.timeline.entries[0].bucket_start + assert 2 == aggregations.timeline.entries[0].total + + def test_build_aggregations_zeroes_empty_window(self) -> None: + """Test that an empty summary/timeline still produces a fully-populated, zeroed result.""" + summary_row = { + "total_runs": 0, + "total_tenants": 0, + "total_catalogs": 0, + "succeeded": 0, + "failed": 0, + "avg_duration_ms": None, + } + + aggregations = ReaderPostgres._build_aggregations( + ts_start=1704067200000, ts_end=1704070800000, bucket="hour", summary_row=summary_row, timeline_rows=[] + ) + + assert 0 == aggregations.total_runs + assert 0.0 == aggregations.success_rate + assert aggregations.avg_duration_ms is None + assert [] == aggregations.timeline.entries + + @pytest.mark.parametrize( + "window_ms, expected_bucket, expected_fragment", + [ + pytest.param(24 * 60 * 60 * 1000, "hour", "date_trunc('hour'", id="7_day_or_less_is_hourly"), + pytest.param(30 * 24 * 60 * 60 * 1000, "day", "date_trunc('day'", id="over_7_days_is_daily"), + ], + ) + def test_window_length_selects_timeline_bucket( + self, + reader: ReaderPostgres, + pg_secret: dict[str, Any], + window_ms: int, + expected_bucket: str, + expected_fragment: str, + ) -> None: + """Test that the window length picks the hourly vs. daily timeline SQL variant.""" + mock_conn = _make_mock_multi_query_connection( + [ + ( + [ + ("total_runs",), + ("total_tenants",), + ("total_catalogs",), + ("succeeded",), + ("failed",), + ("avg_duration_ms",), + ], + [(0, 0, 0, 0, 0, None)], + ), + ([("bucket_start",), ("total",), ("succeeded",), ("failed",)], []), + ] + ) + + with ( + patch("boto3.Session") as mock_session, + patch.object(pb, "psycopg2") as mock_pg, + ): + mock_client = MagicMock() + mock_client.get_secret_value.return_value = {"SecretString": json.dumps(pg_secret)} + mock_session.return_value.client.return_value = mock_client + mock_pg.connect.return_value = mock_conn + + aggregations = reader.read_aggregations( + query_name="runs_jobs_detail_aggregation", timestamp_start=0, timestamp_end=window_ms + ) + + executed_sql = mock_conn.cursor.return_value.__enter__.return_value.execute.call_args_list[1][0][0] + + assert expected_bucket == aggregations.timeline.bucket + assert expected_fragment in executed_sql + + def test_unknown_query_name_raises_runtime_error(self, reader: ReaderPostgres, pg_secret: dict[str, Any]) -> None: + """Test that an unknown aggregation query_name raises RuntimeError.""" + with ( + patch("boto3.Session") as mock_session, + pytest.raises(RuntimeError, match="Unknown aggregation query"), + ): + mock_client = MagicMock() + mock_client.get_secret_value.return_value = {"SecretString": json.dumps(pg_secret)} + mock_session.return_value.client.return_value = mock_client + reader.read_aggregations(query_name="does_not_exist") + + +def _make_mock_multi_query_connection( + query_results: list[tuple[list[tuple[str, ...]], list[tuple[Any, ...]]]], +) -> MagicMock: + """Build a mock psycopg2 connection whose cursor returns a different result per `execute` call.""" + mock_cursor = MagicMock() + descriptions = [description for description, _ in query_results] + rows = [query_rows for _, query_rows in query_results] + mock_cursor.description = descriptions[0] + + call_state = {"index": -1} + + def _execute(*_args: Any, **_kwargs: Any) -> None: + call_state["index"] += 1 + mock_cursor.description = descriptions[call_state["index"]] + mock_cursor.fetchall.return_value = rows[call_state["index"]] + mock_cursor.fetchone.return_value = rows[call_state["index"]][0] if rows[call_state["index"]] else None + + mock_cursor.execute.side_effect = _execute + + mock_conn = MagicMock() + mock_conn.closed = 0 + mock_conn.__enter__ = MagicMock(return_value=mock_conn) + mock_conn.__exit__ = MagicMock(return_value=False) + mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor) + mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False) + return mock_conn + + def _base_row( status: str, message: str | None, diff --git a/tests/unit/utils/test_utils.py b/tests/unit/utils/test_utils.py index 5e8d7d9..bdf9756 100644 --- a/tests/unit/utils/test_utils.py +++ b/tests/unit/utils/test_utils.py @@ -21,7 +21,7 @@ from src.utils.observability import CORRELATION_ID_RESPONSE_HEADER, bind_request_context from src.utils.observability import logger as powertools_logger -from src.utils.utils import build_error_response, dispatch_request, resolve_request_topic +from src.utils.utils import build_aggregations_response, build_error_response, dispatch_request, resolve_request_topic logger = logging.getLogger(__name__) @@ -59,6 +59,18 @@ def test_build_error_response_structure(): assert "Topic not found" == body["errors"][0]["message"] +## build_aggregations_response() +def test_build_aggregations_response_structure(): + """Test that build_aggregations_response returns the distinct envelope (no data/pagination keys).""" + resp = build_aggregations_response({"total_runs": 10}) + + body = json.loads(resp["body"]) + assert body["success"] is True + assert 200 == body["statusCode"] + assert {"total_runs": 10} == body["aggregations"] + assert "data" not in body + + ## resolve_request_topic() def test_resolve_request_topic_normalizes_and_binds_the_topic(caplog): caplog.set_level(logging.INFO)