From 1b66da2c7214d2a27d73a37dbe2ea9db68153cec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=A1s=20Guti=C3=A9rrez?= Date: Sun, 13 Sep 2026 16:39:57 -0300 Subject: [PATCH] Apikey SQL aggregation --- backend/src/modules/api_keys/service.py | 103 +++++++----- backend/src/modules/api_keys/utils.py | 149 ------------------ .../unit/modules/api_keys/test_service.py | 125 +++++++++++++++ 3 files changed, 190 insertions(+), 187 deletions(-) delete mode 100644 backend/src/modules/api_keys/utils.py diff --git a/backend/src/modules/api_keys/service.py b/backend/src/modules/api_keys/service.py index cc4b2493..faab4994 100644 --- a/backend/src/modules/api_keys/service.py +++ b/backend/src/modules/api_keys/service.py @@ -9,14 +9,14 @@ from typing import Any from fastcrud.types import GetMultiResponseDict -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from ...infrastructure.logging import get_logger from ..common.exceptions import PermissionDeniedError, ResourceNotFoundError from .crud import crud_api_keys, crud_key_permissions, crud_key_usage from .enums import KeyPermissionAction, KeyPermissionResource -from .models import APIKey +from .models import APIKey, KeyUsage from .schemas import ( APIKeyCreate, APIKeyCreateInternal, @@ -26,14 +26,6 @@ KeyUsageCreate, KeyUsageRead, ) -from .utils import ( - calculate_basic_metrics, - calculate_daily_usage, - calculate_endpoint_usage, - calculate_error_breakdown, - calculate_response_time_metrics, - parse_usage_records, -) logger = get_logger() @@ -458,31 +450,62 @@ async def get_usage_analytics( since_date = datetime.now(UTC) - timedelta(days=days) - result = await crud_key_usage.get_multi( - db=db, - api_key_id=key_id, - created_at__gte=since_date, - schema_to_select=KeyUsageRead, + # Aggregated in SQL rather than fetched and looped in Python: get_multi caps at + # FastCRUD's default page size (100 rows), which silently truncated every metric. + in_window = (KeyUsage.api_key_id == key_id, KeyUsage.created_at >= since_date) + successful = func.count().filter(KeyUsage.status_code.between(200, 299)) + tokens = func.coalesce(func.sum(KeyUsage.tokens_used), 0) + cost = func.coalesce(func.sum(KeyUsage.cost_microcents), 0) + + totals = ( + await db.execute( + select(func.count(), successful, tokens, cost, func.avg(KeyUsage.response_time_ms)).where(*in_window) + ) + ).one() + total_requests, successful_requests, total_tokens, total_cost, avg_response_time = totals + + endpoint_count = func.count().label("count") + endpoint_rows = await db.execute( + select(KeyUsage.endpoint, endpoint_count) + .where(*in_window) + .group_by(KeyUsage.endpoint) + .order_by(endpoint_count.desc(), KeyUsage.endpoint) + .limit(10) ) - usage_records = parse_usage_records(result) - basic_metrics = calculate_basic_metrics(usage_records) - avg_response_time = calculate_response_time_metrics(usage_records) - most_used_endpoints = calculate_endpoint_usage(usage_records) - error_breakdown = calculate_error_breakdown(usage_records) - usage_by_day = calculate_daily_usage(usage_records) + error_rows = await db.execute( + select(KeyUsage.status_code, func.count()) + .where(*in_window, KeyUsage.status_code >= 400) + .group_by(KeyUsage.status_code) + .order_by(KeyUsage.status_code) + ) + + day = func.date(func.timezone("UTC", KeyUsage.created_at)).label("day") + daily_rows = await db.execute( + select(day, func.count(), successful, tokens, cost).where(*in_window).group_by(day).order_by(day) + ) return { "api_key_id": key_id, - "total_requests": basic_metrics["total_requests"], - "successful_requests": basic_metrics["successful_requests"], - "failed_requests": basic_metrics["failed_requests"], - "total_tokens": basic_metrics["total_tokens"], - "total_cost_microcents": basic_metrics["total_cost"], - "average_response_time_ms": avg_response_time, - "most_used_endpoints": most_used_endpoints, - "error_breakdown": error_breakdown, - "usage_by_day": usage_by_day, + "total_requests": total_requests, + "successful_requests": successful_requests, + "failed_requests": total_requests - successful_requests, + "total_tokens": int(total_tokens), + "total_cost_microcents": int(total_cost), + "average_response_time_ms": float(avg_response_time) if avg_response_time is not None else None, + "most_used_endpoints": [{"endpoint": endpoint, "count": count} for endpoint, count in endpoint_rows], + "error_breakdown": {str(status_code): count for status_code, count in error_rows}, + "usage_by_day": [ + { + "date": row_day.isoformat(), + "requests": requests, + "successful_requests": day_successful, + "failed_requests": requests - day_successful, + "tokens": int(day_tokens), + "cost_microcents": int(day_cost), + } + for row_day, requests, day_successful, day_tokens, day_cost in daily_rows + ], } async def get_user_summary( @@ -505,14 +528,7 @@ async def get_user_summary( total_requests_result = await crud_key_usage.count(db=db, user_id=user_id) total_requests = total_requests_result if isinstance(total_requests_result, int) else 0 - usage_result = await crud_key_usage.get_multi(db=db, user_id=user_id, schema_to_select=KeyUsageRead) - total_cost = 0 - if isinstance(usage_result, dict) and usage_result.get("data"): - usage_data = usage_result["data"] - if isinstance(usage_data, list): - for u in usage_data: - if isinstance(u, dict) and u.get("cost_microcents"): - total_cost += u["cost_microcents"] + total_cost = await self.sum_user_usage_cost(user_id=user_id, db=db) return { "user_id": user_id, @@ -523,6 +539,17 @@ async def get_user_summary( "keys": keys_data, } + async def sum_user_usage_cost(self, user_id: int, db: AsyncSession) -> int: + """Return total ``cost_microcents`` across all of a user's key-usage records. + + Replaces a fetch-all + Python-loop sum that silently capped at FastCRUD's + default page size (100 rows), under-reporting cost for any active user. + Single query, single round-trip. coalesce(..., 0) covers the no-usage + case (SUM returns NULL otherwise). + """ + stmt = select(func.coalesce(func.sum(KeyUsage.cost_microcents), 0)).where(KeyUsage.user_id == user_id) + return int((await db.execute(stmt)).scalar_one() or 0) + async def _check_permission( self, api_key_id: int, diff --git a/backend/src/modules/api_keys/utils.py b/backend/src/modules/api_keys/utils.py deleted file mode 100644 index 59eb1c07..00000000 --- a/backend/src/modules/api_keys/utils.py +++ /dev/null @@ -1,149 +0,0 @@ -"""Utility functions for API key analytics and data processing.""" - -from datetime import datetime -from typing import Any - - -def calculate_basic_metrics(usage_records: list[dict[str, Any]]) -> dict[str, Any]: - """Calculate basic usage metrics from usage records. - - Args: - usage_records: List of usage record dictionaries - - Returns: - Dictionary containing basic metrics - """ - total_requests = len(usage_records) - successful_requests = len([u for u in usage_records if isinstance(u, dict) and 200 <= u.get("status_code", 0) < 300]) - failed_requests = total_requests - successful_requests - - total_tokens = sum(u.get("tokens_used", 0) or 0 for u in usage_records if isinstance(u, dict)) - - total_cost = sum(u.get("cost_microcents", 0) or 0 for u in usage_records if isinstance(u, dict)) - - return { - "total_requests": total_requests, - "successful_requests": successful_requests, - "failed_requests": failed_requests, - "total_tokens": total_tokens, - "total_cost": total_cost, - } - - -def calculate_response_time_metrics(usage_records: list[dict[str, Any]]) -> float | None: - """Calculate average response time from usage records. - - Args: - usage_records: List of usage record dictionaries - - Returns: - Average response time in milliseconds or None if no data - """ - response_times = [] - for u in usage_records: - if isinstance(u, dict) and u.get("response_time_ms") is not None: - response_times.append(u["response_time_ms"]) - - return sum(response_times) / len(response_times) if response_times else None - - -def calculate_endpoint_usage(usage_records: list[dict[str, Any]], limit: int = 10) -> list[dict[str, Any]]: - """Calculate most used endpoints from usage records. - - Args: - usage_records: List of usage record dictionaries - limit: Maximum number of endpoints to return - - Returns: - List of endpoint usage dictionaries sorted by count - """ - endpoint_counts: dict[str, int] = {} - for record in usage_records: - if isinstance(record, dict): - endpoint = record.get("endpoint", "") - endpoint_counts[endpoint] = endpoint_counts.get(endpoint, 0) + 1 - - return [ - {"endpoint": endpoint, "count": count} - for endpoint, count in sorted(endpoint_counts.items(), key=lambda x: x[1], reverse=True)[:limit] - ] - - -def calculate_error_breakdown(usage_records: list[dict[str, Any]]) -> dict[str, int]: - """Calculate error status code breakdown from usage records. - - Args: - usage_records: List of usage record dictionaries - - Returns: - Dictionary mapping status codes to counts - """ - error_counts: dict[str, int] = {} - for record in usage_records: - if isinstance(record, dict) and record.get("status_code", 0) >= 400: - status = record.get("status_code", 0) - error_counts[str(status)] = error_counts.get(str(status), 0) + 1 - - return error_counts - - -def calculate_daily_usage(usage_records: list[dict[str, Any]]) -> list[dict[str, Any]]: - """Calculate daily usage breakdown from usage records. - - Args: - usage_records: List of usage record dictionaries - - Returns: - List of daily usage dictionaries sorted by date - """ - daily_usage: dict[str, dict[str, Any]] = {} - - for record in usage_records: - if not isinstance(record, dict) or not record.get("created_at"): - continue - - created_at = record["created_at"] - if isinstance(created_at, str): - try: - created_at = datetime.fromisoformat(created_at.replace("Z", "+00:00")) - except (ValueError, AttributeError): - continue - - day_key = created_at.strftime("%Y-%m-%d") - if day_key not in daily_usage: - daily_usage[day_key] = { - "date": day_key, - "requests": 0, - "successful_requests": 0, - "failed_requests": 0, - "tokens": 0, - "cost_microcents": 0, - } - - daily_usage[day_key]["requests"] += 1 - if 200 <= record.get("status_code", 0) < 300: - daily_usage[day_key]["successful_requests"] += 1 - else: - daily_usage[day_key]["failed_requests"] += 1 - daily_usage[day_key]["tokens"] += record.get("tokens_used", 0) or 0 - daily_usage[day_key]["cost_microcents"] += record.get("cost_microcents", 0) or 0 - - return sorted(daily_usage.values(), key=lambda x: x["date"]) - - -def parse_usage_records(result: Any) -> list[dict[str, Any]]: - """Parse usage records from database result. - - Args: - result: Database query result - - Returns: - List of usage record dictionaries - """ - usage_records: list[dict[str, Any]] = [] - if isinstance(result, dict) and result.get("data"): - data = result["data"] - if isinstance(data, list): - usage_records = data - - return usage_records diff --git a/backend/tests/unit/modules/api_keys/test_service.py b/backend/tests/unit/modules/api_keys/test_service.py index 3371b7a6..342bf2e6 100644 --- a/backend/tests/unit/modules/api_keys/test_service.py +++ b/backend/tests/unit/modules/api_keys/test_service.py @@ -8,6 +8,7 @@ from src.modules.api_keys.crud import crud_api_keys, crud_key_permissions from src.modules.api_keys.enums import KeyPermissionAction, KeyPermissionResource +from src.modules.api_keys.models import KeyUsage from src.modules.api_keys.schemas import ( APIKeyCreate, APIKeyCreateInternal, @@ -370,6 +371,130 @@ async def test_get_user_summary(api_key_service, db_session: AsyncSession, test_ assert len(summary["keys"]) >= 1 +@pytest.mark.asyncio +async def test_sum_user_usage_cost_not_page_capped(api_key_service, db_session: AsyncSession, test_user: dict, test_api_key): + """Usage cost is summed in SQL, so it is not capped at a page size. + + Regression lock for the old fetch-all + Python-loop sum, which silently + capped at FastCRUD's default 100 rows and under-reported cost for any user + with many usage records. We seed 150 usage rows of 1_000 microcents each; + the sum must report all 150_000, not the 100-row cap. + """ + db_session.add_all( + [ + KeyUsage( + api_key_id=test_api_key["id"], + user_id=test_user["id"], + endpoint="/api/v1/test", + method="GET", + status_code=200, + cost_microcents=1_000, + ) + for _ in range(150) + ] + ) + await db_session.commit() + + total = await api_key_service.sum_user_usage_cost(user_id=test_user["id"], db=db_session) + summary = await api_key_service.get_user_summary(user_id=test_user["id"], db=db_session) + + assert total == 150_000 # 150 * 1_000, proving no 100-row cap + assert summary["total_requests"] == 150 + assert summary["total_cost_microcents"] == 150_000 + + +@pytest.mark.asyncio +async def test_sum_user_usage_cost_scoped_to_user( + api_key_service, db_session: AsyncSession, test_user: dict, test_user_2: dict, test_api_key +): + """The sum is scoped to the given user and never bleeds another user's cost. + + Locks the ``WHERE user_id == user_id`` filter: with rows for two users, each + user's sum reflects only their own usage. + """ + db_session.add_all( + [ + KeyUsage( + api_key_id=test_api_key["id"], + user_id=test_user["id"], + endpoint="/api/v1/test", + method="GET", + status_code=200, + cost_microcents=1_000, + ), + KeyUsage( + api_key_id=test_api_key["id"], + user_id=test_user_2["id"], + endpoint="/api/v1/test", + method="GET", + status_code=200, + cost_microcents=9_999, + ), + ] + ) + await db_session.commit() + + assert await api_key_service.sum_user_usage_cost(user_id=test_user["id"], db=db_session) == 1_000 + assert await api_key_service.sum_user_usage_cost(user_id=test_user_2["id"], db=db_session) == 9_999 + + +@pytest.mark.asyncio +async def test_get_usage_analytics_not_page_capped(api_key_service, db_session: AsyncSession, test_user: dict, test_api_key): + """Analytics are aggregated in SQL, so no metric is truncated at 100 rows. + + Seeds 150 in-window rows (every third one a 500 on /api/v1/a, the rest 200s on + /api/v1/b) plus one row older than the window, which must be excluded. + """ + rows = [ + KeyUsage( + api_key_id=test_api_key["id"], + user_id=test_user["id"], + endpoint="/api/v1/a" if i % 3 == 0 else "/api/v1/b", + method="GET", + status_code=500 if i % 3 == 0 else 200, + tokens_used=2, + cost_microcents=1_000, + response_time_ms=100, + ) + for i in range(150) + ] + stale = KeyUsage( + api_key_id=test_api_key["id"], + user_id=test_user["id"], + endpoint="/api/v1/stale", + method="GET", + status_code=404, + tokens_used=1_000, + cost_microcents=1_000_000, + response_time_ms=9_999, + ) + stale.created_at = datetime.now(UTC) - timedelta(days=40) + db_session.add_all([*rows, stale]) + await db_session.commit() + + analytics = await api_key_service.get_usage_analytics(key_id=test_api_key["id"], user_id=test_user["id"], db=db_session) + + assert analytics["total_requests"] == 150 + assert analytics["successful_requests"] == 100 + assert analytics["failed_requests"] == 50 + assert analytics["total_tokens"] == 300 + assert analytics["total_cost_microcents"] == 150_000 + assert analytics["average_response_time_ms"] == 100.0 + assert analytics["most_used_endpoints"] == [ + {"endpoint": "/api/v1/b", "count": 100}, + {"endpoint": "/api/v1/a", "count": 50}, + ] + assert analytics["error_breakdown"] == {"500": 50} + + usage_by_day = analytics["usage_by_day"] + assert sum(d["requests"] for d in usage_by_day) == 150 + assert sum(d["successful_requests"] for d in usage_by_day) == 100 + assert sum(d["failed_requests"] for d in usage_by_day) == 50 + assert sum(d["tokens"] for d in usage_by_day) == 300 + assert sum(d["cost_microcents"] for d in usage_by_day) == 150_000 + assert all(len(d["date"]) == 10 for d in usage_by_day) # YYYY-MM-DD + + @pytest.mark.asyncio async def test_api_key_hash_roundtrip(api_key_service): """Hashing produces a fresh salt each call; verifying must still succeed."""