diff --git a/app/main.py b/app/main.py index 560f055..42bff85 100644 --- a/app/main.py +++ b/app/main.py @@ -11,6 +11,7 @@ jobs_status, unit_jobs, health, + statistics, tiles, upscale_tasks, sync_jobs, @@ -47,3 +48,4 @@ app.include_router(upscale_tasks.router) app.include_router(health.router) app.include_router(parameters.router) +app.include_router(statistics.router) diff --git a/app/routers/statistics.py b/app/routers/statistics.py new file mode 100644 index 0000000..c5adb70 --- /dev/null +++ b/app/routers/statistics.py @@ -0,0 +1,72 @@ +from datetime import date + +from fastapi import APIRouter, Depends, HTTPException, Query +from loguru import logger +from sqlalchemy.orm import Session + +from app.database.db import get_db +from app.error import DispatcherException, ErrorResponse, InternalException +from app.middleware.error_handling import get_dispatcher_error_response +from app.schemas.statistics import PublicStatisticsResponse +from app.services.statistics import get_public_statistics + +router = APIRouter() + + +@router.get( + "/statistics", + tags=["Statistics"], + summary="Get public statistics for executed processing jobs and upscaling tasks", + responses={ + InternalException.http_status: { + "description": "Internal server error", + "model": ErrorResponse, + "content": { + "application/json": { + "example": get_dispatcher_error_response( + InternalException(), "request-id" + ) + } + }, + }, + }, +) +async def get_statistics( + db: Session = Depends(get_db), + date_from: date | None = Query( + default=None, + description=( + "Start date (inclusive) for filtering records by creation date, " + "format YYYY-MM-DD" + ), + ), + date_to: date | None = Query( + default=None, + description=( + "End date (inclusive) for filtering records by creation date, " + "format YYYY-MM-DD" + ), + ), +) -> PublicStatisticsResponse: + try: + if date_from and date_to and date_from > date_to: + raise HTTPException( + status_code=400, + detail="date_from must be before or equal to date_to", + ) + + return get_public_statistics( + database=db, + date_from=date_from, + date_to=date_to, + ) + except HTTPException as he: + raise he + except DispatcherException as de: + raise de + except Exception as e: + logger.error(f"Error retrieving public statistics: {e}") + raise InternalException( + message="An error occurred while retrieving public statistics.", + details={"error": str(e)}, + ) diff --git a/app/schemas/statistics.py b/app/schemas/statistics.py new file mode 100644 index 0000000..c12f60c --- /dev/null +++ b/app/schemas/statistics.py @@ -0,0 +1,32 @@ +from datetime import datetime +from typing import Dict + +from pydantic import BaseModel, Field + + +class EntityStatistics(BaseModel): + total: int = Field(..., description="Total amount of records") + by_status: Dict[str, int] = Field(..., description="Totals grouped by status") + by_platform: Dict[str, int] = Field( + ..., description="Totals grouped by platform label" + ) + by_service: Dict[str, int] = Field( + ..., + description="Totals grouped by executed service process ID", + ) + + +class UpscalingStatistics(EntityStatistics): + average_processing_jobs_per_upscaling_task: float = Field( + ..., description="Average number of processing jobs linked to each upscaling task" + ) + + +class PublicStatisticsResponse(BaseModel): + generated_at: datetime = Field(..., description="Timestamp when stats were generated") + processing_jobs: EntityStatistics = Field( + ..., description="Statistics for executed processing jobs" + ) + upscaling_tasks: UpscalingStatistics = Field( + ..., description="Statistics for executed upscaling tasks" + ) diff --git a/app/services/statistics.py b/app/services/statistics.py new file mode 100644 index 0000000..ade38bc --- /dev/null +++ b/app/services/statistics.py @@ -0,0 +1,222 @@ +from datetime import date, datetime, time +import json +from threading import Lock +from typing import Any, Dict, Iterable, Sequence, Tuple, cast +from urllib.parse import unquote, urlparse + +from sqlalchemy import func +from sqlalchemy.orm import Session + +from app.database.models.processing_job import ProcessingJobRecord +from app.database.models.upscaling_task import UpscalingTaskRecord +from app.schemas.statistics import ( + EntityStatistics, + PublicStatisticsResponse, + UpscalingStatistics, +) + +STATISTICS_CACHE_TTL_SECONDS = 60 * 15 +_STATISTICS_CACHE: Dict[tuple[str, str], tuple[datetime, dict]] = {} +_STATISTICS_CACHE_LOCK = Lock() + + +def _normalize_grouped_counts(rows: Iterable[Any]) -> Dict[str, int]: + counts: Dict[str, int] = {} + for row in rows: + key, count = cast(Sequence[Any], row) + normalized_key = key.value if hasattr(key, "value") else str(key) + counts[normalized_key] = int(count) + return counts + + +def _extract_service_key(raw_service: str | None) -> str: + if not raw_service: + return "unknown" + try: + service = json.loads(raw_service) + except (TypeError, ValueError): + return "unknown" + + application = service.get("application") + if not isinstance(application, str) or not application.strip(): + return "unknown" + + parsed = urlparse(application) + process_id = parsed.path.strip("/").split("/")[-1] if parsed.path else "" + if not process_id: + process_id = application.strip() + + process_id = unquote(process_id) + if process_id.endswith(".json"): + process_id = process_id[:-5] + + return process_id or "unknown" + + +def _count_services(service_rows: Iterable[Tuple[str | None]]) -> Dict[str, int]: + counts: Dict[str, int] = {} + for (raw_service,) in service_rows: + key = _extract_service_key(raw_service) + counts[key] = counts.get(key, 0) + 1 + return counts + + +def _get_date_bounds( + date_from: date | None, + date_to: date | None, +) -> tuple[datetime | None, datetime | None]: + start_datetime = datetime.combine(date_from, time.min) if date_from else None + end_datetime = datetime.combine(date_to, time.max) if date_to else None + return start_datetime, end_datetime + + +def _make_cache_key( + date_from: date | None, + date_to: date | None, +) -> tuple[str, str]: + return ( + date_from.isoformat() if date_from else "", + date_to.isoformat() if date_to else "", + ) + + +def _get_cached_statistics( + cache_key: tuple[str, str], now: datetime +) -> PublicStatisticsResponse | None: + with _STATISTICS_CACHE_LOCK: + entry = _STATISTICS_CACHE.get(cache_key) + if not entry: + return None + + expiry, payload = entry + if expiry <= now: + _STATISTICS_CACHE.pop(cache_key, None) + return None + + return PublicStatisticsResponse.model_validate(payload) + + +def _set_cached_statistics( + cache_key: tuple[str, str], + now: datetime, + statistics: PublicStatisticsResponse, +) -> None: + expires_at = now.replace(microsecond=0) + expires_at = datetime.fromtimestamp( + expires_at.timestamp() + STATISTICS_CACHE_TTL_SECONDS + ) + + with _STATISTICS_CACHE_LOCK: + _STATISTICS_CACHE[cache_key] = ( + expires_at, + statistics.model_dump(mode="json"), + ) + + +def get_public_statistics( + database: Session, + date_from: date | None = None, + date_to: date | None = None, +) -> PublicStatisticsResponse: + now = datetime.utcnow() + cache_key = _make_cache_key(date_from, date_to) + cached_statistics = _get_cached_statistics(cache_key=cache_key, now=now) + if cached_statistics is not None: + return cached_statistics + + start_datetime, end_datetime = _get_date_bounds(date_from, date_to) + + processing_jobs_query = database.query(ProcessingJobRecord) + upscaling_tasks_query = database.query(UpscalingTaskRecord) + + if start_datetime is not None: + processing_jobs_query = processing_jobs_query.filter( + ProcessingJobRecord.created >= start_datetime + ) + upscaling_tasks_query = upscaling_tasks_query.filter( + UpscalingTaskRecord.created >= start_datetime + ) + + if end_datetime is not None: + processing_jobs_query = processing_jobs_query.filter( + ProcessingJobRecord.created <= end_datetime + ) + upscaling_tasks_query = upscaling_tasks_query.filter( + UpscalingTaskRecord.created <= end_datetime + ) + + processing_jobs_total = ( + processing_jobs_query.with_entities(func.count(ProcessingJobRecord.id)).scalar() + or 0 + ) + upscaling_tasks_total = ( + upscaling_tasks_query.with_entities(func.count(UpscalingTaskRecord.id)).scalar() + or 0 + ) + + processing_jobs_by_status = _normalize_grouped_counts( + processing_jobs_query.with_entities( + ProcessingJobRecord.status, func.count(ProcessingJobRecord.id) + ) + .group_by(ProcessingJobRecord.status) + .all() + ) + processing_jobs_by_platform = _normalize_grouped_counts( + processing_jobs_query.with_entities( + ProcessingJobRecord.label, func.count(ProcessingJobRecord.id) + ) + .group_by(ProcessingJobRecord.label) + .all() + ) + processing_jobs_by_service = _count_services( + processing_jobs_query.with_entities(ProcessingJobRecord.service).all() + ) + upscaling_tasks_by_status = _normalize_grouped_counts( + upscaling_tasks_query.with_entities( + UpscalingTaskRecord.status, func.count(UpscalingTaskRecord.id) + ) + .group_by(UpscalingTaskRecord.status) + .all() + ) + upscaling_tasks_by_platform = _normalize_grouped_counts( + upscaling_tasks_query.with_entities( + UpscalingTaskRecord.label, func.count(UpscalingTaskRecord.id) + ) + .group_by(UpscalingTaskRecord.label) + .all() + ) + upscaling_tasks_by_service = _count_services( + upscaling_tasks_query.with_entities(UpscalingTaskRecord.service).all() + ) + + processing_jobs_linked_to_upscaling_tasks = ( + processing_jobs_query.with_entities(func.count(ProcessingJobRecord.id)) + .filter(ProcessingJobRecord.upscaling_task_id.isnot(None)) + .scalar() + or 0 + ) + + average_jobs_per_upscaling_task = ( + round(processing_jobs_linked_to_upscaling_tasks / upscaling_tasks_total, 2) + if upscaling_tasks_total + else 0.0 + ) + + statistics = PublicStatisticsResponse( + generated_at=now, + processing_jobs=EntityStatistics( + total=processing_jobs_total, + by_status=processing_jobs_by_status, + by_platform=processing_jobs_by_platform, + by_service=processing_jobs_by_service, + ), + upscaling_tasks=UpscalingStatistics( + total=upscaling_tasks_total, + by_status=upscaling_tasks_by_status, + by_platform=upscaling_tasks_by_platform, + by_service=upscaling_tasks_by_service, + average_processing_jobs_per_upscaling_task=average_jobs_per_upscaling_task, + ), + ) + _set_cached_statistics(cache_key=cache_key, now=now, statistics=statistics) + return statistics diff --git a/requirements.txt b/requirements.txt index afd66e0..f76f501 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,6 +12,7 @@ mkdocs-material mkdocs-material[diagrams] mypy mypy_extensions +typing_extensions>=4.12.2 openeo psycopg2-binary pydantic diff --git a/tests/routers/test_statistics.py b/tests/routers/test_statistics.py new file mode 100644 index 0000000..9f4dc70 --- /dev/null +++ b/tests/routers/test_statistics.py @@ -0,0 +1,92 @@ +from datetime import datetime +from unittest.mock import patch + +from fastapi import status + +from app.schemas.statistics import ( + EntityStatistics, + PublicStatisticsResponse, + UpscalingStatistics, +) + + +@patch("app.routers.statistics.get_public_statistics") +def test_statistics_get_200(mock_get_public_statistics, client): + mock_get_public_statistics.return_value = PublicStatisticsResponse( + generated_at=datetime(2026, 1, 1, 12, 0, 0), + processing_jobs=EntityStatistics( + total=42, + by_status={"finished": 30, "failed": 12}, + by_platform={"openeo": 40, "ogc_api_process": 2}, + by_service={ + "variabilitymap": 40, + "land-cover": 2, + }, + ), + upscaling_tasks=UpscalingStatistics( + total=10, + by_status={"finished": 9, "failed": 1}, + by_platform={"openeo": 10}, + by_service={"variabilitymap": 10}, + average_processing_jobs_per_upscaling_task=4.2, + ), + ) + + response = client.get("/statistics") + + assert response.status_code == status.HTTP_200_OK + payload = response.json() + assert payload["processing_jobs"]["total"] == 42 + assert payload["upscaling_tasks"]["total"] == 10 + assert payload["upscaling_tasks"]["average_processing_jobs_per_upscaling_task"] == 4.2 + assert payload["processing_jobs"]["by_status"]["finished"] == 30 + assert payload["processing_jobs"]["by_service"]["variabilitymap"] == 40 + + +@patch("app.routers.statistics.get_public_statistics") +def test_statistics_get_500(mock_get_public_statistics, client): + mock_get_public_statistics.side_effect = RuntimeError("Database timeout") + + response = client.get("/statistics") + + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + assert "An error occurred while retrieving public statistics." in response.json().get( + "message", "" + ) + + +@patch("app.routers.statistics.get_public_statistics") +def test_statistics_get_200_with_date_filter(mock_get_public_statistics, client): + mock_get_public_statistics.return_value = PublicStatisticsResponse( + generated_at=datetime(2026, 1, 1, 12, 0, 0), + processing_jobs=EntityStatistics( + total=0, + by_status={}, + by_platform={}, + by_service={}, + ), + upscaling_tasks=UpscalingStatistics( + total=0, + by_status={}, + by_platform={}, + by_service={}, + average_processing_jobs_per_upscaling_task=0.0, + ), + ) + + response = client.get("/statistics?date_from=2026-01-01&date_to=2026-01-31") + + assert response.status_code == status.HTTP_200_OK + mock_get_public_statistics.assert_called_once() + _, kwargs = mock_get_public_statistics.call_args + assert kwargs["date_from"].isoformat() == "2026-01-01" + assert kwargs["date_to"].isoformat() == "2026-01-31" + + +def test_statistics_get_400_when_date_range_invalid(client): + response = client.get("/statistics?date_from=2026-02-01&date_to=2026-01-01") + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "date_from must be before or equal to date_to" in response.json().get( + "detail", "" + )