From a141aa86e93af61f38e6023cee4e6a25128acd45 Mon Sep 17 00:00:00 2001 From: Majyles <123649730+Majyles@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:18:33 +0200 Subject: [PATCH 1/4] feat(alerts): count alerts per day over a period Add GET /alerts/all/fromdate/count-per-day so that a month calendar gets its daily alert counts in one request instead of one per day. - from_date / to_date (inclusive), at most 62 days: a month calendar shows at most 6 weeks, and this bounds the per-day risk lookups. - tz: IANA timezone used to split the days, UTC by default to match /alerts/all/fromdate. - Same alerts as /alerts/all/fromdate/count: organization scoping, and the risk filter of each day (fetched concurrently; days sharing the same scores are counted by a single query), or the risk_score override. - Returns [{"date": ..., "count": ...}] for the days with alerts only. Closes #719 Co-Authored-By: Claude Opus 5.5 --- src/app/api/api_v1/endpoints/alerts.py | 79 +++++++++++++++- src/app/schemas/alerts.py | 17 +++- src/tests/endpoints/test_alerts.py | 123 ++++++++++++++++++++++++- 3 files changed, 216 insertions(+), 3 deletions(-) diff --git a/src/app/api/api_v1/endpoints/alerts.py b/src/app/api/api_v1/endpoints/alerts.py index 855fdd4b..5e20646c 100644 --- a/src/app/api/api_v1/endpoints/alerts.py +++ b/src/app/api/api_v1/endpoints/alerts.py @@ -4,10 +4,12 @@ # See LICENSE or go to for full license details. +import asyncio import csv import io from datetime import date, datetime, time, timedelta from typing import Any, AsyncIterator, Dict, Iterable, Iterator, List, Tuple, Union, cast +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from fastapi import APIRouter, Depends, HTTPException, Path, Query, Security, status from fastapi.responses import StreamingResponse @@ -22,7 +24,7 @@ from app.crud import AlertCRUD, CameraCRUD, SequenceCRUD from app.db import get_session, session_factory from app.models import Alert, AlertSequence, AnnotationType, Camera, Sequence, UserRole -from app.schemas.alerts import AlertCount, AlertCreate, AlertReadWithSequences +from app.schemas.alerts import AlertCount, AlertCreate, AlertDayCount, AlertReadWithSequences from app.schemas.login import TokenPayload from app.schemas.sequences import SequenceRead from app.services.alerts import refresh_alert_state @@ -458,6 +460,81 @@ async def count_alerts_from_date( return AlertCount(count=int((await session.exec(count_stmt)).one())) +# A month calendar shows at most 6 weeks: 62 days cover it, and bound the per-day risk lookups. +MAX_COUNT_PER_DAY_PERIOD = 62 + + +@router.get( + "/all/fromdate/count-per-day", + status_code=status.HTTP_200_OK, + summary="Count the alerts per day over a period", +) +async def count_alerts_per_day( + from_date: date = Query(..., description="First day of the period (inclusive), in `tz`"), + to_date: date = Query(..., description="Last day of the period (inclusive), in `tz`"), + tz: str = Query("UTC", description="IANA timezone used to split the days, e.g. Europe/Paris"), + risk_score: Union[FwiClass, None] = Query( + None, + description="Override FWI class applied to every sequence; bypasses risk-api lookup. Ignored for admins.", + ), + session: AsyncSession = Depends(get_session), + token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]), +) -> List[AlertDayCount]: + telemetry_client.capture( + token_payload.sub, + event="alerts-count-per-day", + properties={"from_date": from_date.isoformat(), "to_date": to_date.isoformat(), "tz": tz}, + ) + if to_date < from_date: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="to_date must be on or after from_date", + ) + days = [from_date + timedelta(days=offset) for offset in range((to_date - from_date).days + 1)] + if len(days) > MAX_COUNT_PER_DAY_PERIOD: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"The period cannot exceed {MAX_COUNT_PER_DAY_PERIOD} days", + ) + try: + ZoneInfo(tz) + except (ZoneInfoNotFoundError, ValueError): + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Unknown timezone: {tz}") + + # Same alerts as /all/fromdate/count, each day with its own risk scores: days sharing the + # same scores are counted by a single query. + days_by_scores: Dict[Tuple[Tuple[int, Union[str, None]], ...], List[date]] = {} + if token_payload.is_admin: + days_by_scores[()] = days + elif risk_score is not None: + class_per_camera = await _resolve_fwi_class_per_camera( + session, token_payload.organization_id, override_class=risk_score + ) + days_by_scores[tuple(sorted(class_per_camera.items()))] = days + else: + scores_per_day = await asyncio.gather( + *(risk_service.get_scores_for_date(day, organization_id=token_payload.organization_id) for day in days) + ) + for day, scores in zip(days, scores_per_day, strict=True): + days_by_scores.setdefault(tuple(sorted(scores.items())), []).append(day) + + # started_at is a naive UTC datetime: convert it to the requested timezone before taking its day + local_day: Any = func.date(func.timezone(tz, func.timezone("UTC", Alert.started_at))) + counts: Dict[date, int] = {} + for scores_key, scores_days in days_by_scores.items(): + stmt: Any = select(local_day, func.count()).where(local_day.in_(scores_days)).group_by(local_day) + if not token_payload.is_admin: + stmt = stmt.where(Alert.organization_id == token_payload.organization_id) + seq_filter = max_conf_filter_clause(dict(scores_key)) + if seq_filter is not None: + seq_match: Any = select(AlertSequence.alert_id).join( + Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id) + ) + stmt = stmt.where(cast(Any, Alert.id).in_(seq_match.where(seq_filter))) + counts.update((day, count) for day, count in (await session.exec(stmt)).all()) + return [AlertDayCount(date=day, count=count) for day, count in sorted(counts.items())] + + @router.post( "/{alert_id}/sequences/{sequence_id}/unmatch", status_code=status.HTTP_200_OK, diff --git a/src/app/schemas/alerts.py b/src/app/schemas/alerts.py index 2be5c08e..36971e94 100644 --- a/src/app/schemas/alerts.py +++ b/src/app/schemas/alerts.py @@ -3,6 +3,7 @@ # This program is licensed under the Apache License 2.0. # See LICENSE or go to for full license details. +import datetime as dt from datetime import datetime from typing import List, Optional @@ -10,7 +11,15 @@ from app.schemas.sequences import SequenceRead -__all__ = ["AlertBase", "AlertCount", "AlertCreate", "AlertRead", "AlertReadWithSequences", "AlertUpdate"] +__all__ = [ + "AlertBase", + "AlertCount", + "AlertCreate", + "AlertDayCount", + "AlertRead", + "AlertReadWithSequences", + "AlertUpdate", +] class AlertBase(BaseModel): @@ -41,3 +50,9 @@ class AlertReadWithSequences(AlertRead): class AlertCount(BaseModel): count: int + + +class AlertDayCount(BaseModel): + # Module-qualified type: a field named `date` would shadow `datetime.date` + date: dt.date = Field(..., description="Day, in the requested timezone") + count: int = Field(..., ge=1, description="Number of alerts started that day") diff --git a/src/tests/endpoints/test_alerts.py b/src/tests/endpoints/test_alerts.py index 170c4e7c..d433f29a 100644 --- a/src/tests/endpoints/test_alerts.py +++ b/src/tests/endpoints/test_alerts.py @@ -6,7 +6,8 @@ import csv import io from datetime import datetime, timedelta -from typing import Any, Dict, List, Tuple, cast +from typing import Any, Dict, List, Tuple, Union, cast +from unittest.mock import AsyncMock, patch import pytest # type: ignore from httpx import AsyncClient @@ -322,6 +323,126 @@ async def test_alerts_count_unauthenticated(async_client: AsyncClient): assert resp.status_code == 401, resp.text +async def _create_alert_started_at( + session: AsyncSession, org_id: int, started_at: datetime, camera_id: Union[int, None] = None, max_conf: float = 0.9 +) -> Alert: + alert = Alert(organization_id=org_id, started_at=started_at, last_seen_at=started_at + timedelta(minutes=5)) + session.add(alert) + await session.commit() + await session.refresh(alert) + if camera_id is not None: + seq = Sequence( + camera_id=camera_id, + pose_id=None, + camera_azimuth=180.0, + sequence_azimuth=175.0, + cone_angle=5.0, + is_wildfire=None, + started_at=started_at, + last_seen_at=started_at + timedelta(minutes=5), + max_conf=max_conf, + ) + session.add(seq) + await session.commit() + await session.refresh(seq) + session.add(AlertSequence(alert_id=alert.id, sequence_id=seq.id)) + await session.commit() + return alert + + +def _user_auth(user_idx: int) -> Dict[str, str]: + user = pytest.user_table[user_idx] + return pytest.get_token(user["id"], user["role"].split(), user["organization_id"]) + + +@pytest.mark.asyncio +async def test_alerts_count_per_day(async_client: AsyncClient, camera_session: AsyncSession): + day = datetime(2025, 7, 14) + await _create_alert_started_at(camera_session, 1, day + timedelta(hours=10)) + await _create_alert_started_at(camera_session, 1, day + timedelta(hours=23, minutes=30)) + await _create_alert_started_at(camera_session, 1, day + timedelta(days=2, hours=12)) + await _create_alert_started_at(camera_session, 2, day + timedelta(hours=10)) + period = "from_date=2025-07-14&to_date=2025-07-16" + + # Agent: its organization only, days without alerts are omitted + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=_user_auth(1)) + assert resp.status_code == 200, resp.text + assert resp.json() == [{"date": "2025-07-14", "count": 2}, {"date": "2025-07-16", "count": 1}] + # Consistent with the single-date count + resp = await async_client.get("/alerts/all/fromdate/count?from_date=2025-07-14", headers=_user_auth(1)) + assert resp.json() == {"count": 2} + + # Admin: every organization + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=_user_auth(0)) + assert resp.json() == [{"date": "2025-07-14", "count": 3}, {"date": "2025-07-16", "count": 1}] + + # 23:30 UTC is already the next day in Paris (UTC+2 in July) + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}&tz=Europe/Paris", headers=_user_auth(1)) + assert resp.json() == [ + {"date": "2025-07-14", "count": 1}, + {"date": "2025-07-15", "count": 1}, + {"date": "2025-07-16", "count": 1}, + ] + + resp = await async_client.get( + "/alerts/all/fromdate/count-per-day?from_date=2025-07-15&to_date=2025-07-15", headers=_user_auth(1) + ) + assert resp.json() == [] + + +@pytest.mark.parametrize( + ("query", "status_code"), + [ + ("from_date=2025-07-16&to_date=2025-07-14", 422), + ("from_date=2025-07-01&to_date=2025-09-01", 422), # 63 days + ("from_date=2025-07-01&to_date=2025-08-31", 200), # 62 days + ("from_date=2025-07-01&to_date=2025-07-31&tz=Mars/Olympus_Mons", 422), + ("from_date=2025-07-01", 422), + ], +) +@pytest.mark.asyncio +async def test_alerts_count_per_day_validation( + async_client: AsyncClient, camera_session: AsyncSession, query: str, status_code: int +): + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{query}", headers=_user_auth(1)) + assert resp.status_code == status_code, resp.text + + +@pytest.mark.asyncio +async def test_alerts_count_per_day_unauthenticated(async_client: AsyncClient): + resp = await async_client.get("/alerts/all/fromdate/count-per-day?from_date=2025-07-14&to_date=2025-07-16") + assert resp.status_code == 401, resp.text + + +@pytest.mark.asyncio +async def test_alerts_count_per_day_risk_scores_per_day(async_client: AsyncClient, camera_session: AsyncSession): + camera_id = pytest.camera_table[0]["id"] + day = datetime(2025, 7, 14) + # Two low-confidence alerts, but only the first day is low risk (threshold above 0.2) + await _create_alert_started_at(camera_session, 1, day + timedelta(hours=10), camera_id, max_conf=0.2) + await _create_alert_started_at(camera_session, 1, day + timedelta(days=1, hours=10), camera_id, max_conf=0.2) + period = "from_date=2025-07-14&to_date=2025-07-15" + + def scores_for_date(target_date, organization_id=None): + return {camera_id: "low"} if target_date == day.date() else {} + + scores_mock = AsyncMock(side_effect=scores_for_date) + with patch("app.api.api_v1.endpoints.alerts.risk_service.get_scores_for_date", new=scores_mock): + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=_user_auth(1)) + assert resp.status_code == 200, resp.text + assert resp.json() == [{"date": "2025-07-15", "count": 1}] + assert scores_mock.await_count == 2 + assert {call.kwargs["organization_id"] for call in scores_mock.await_args_list} == {1} + + # A risk score override applies to every day and skips the risk API + scores_mock.reset_mock() + resp = await async_client.get( + f"/alerts/all/fromdate/count-per-day?{period}&risk_score=moderate", headers=_user_auth(1) + ) + assert resp.json() == [{"date": "2025-07-14", "count": 1}, {"date": "2025-07-15", "count": 1}] + scores_mock.assert_not_awaited() + + @pytest.mark.asyncio async def test_triangulation_creates_single_alert( async_client: AsyncClient, detection_session: AsyncSession, mock_img: bytes From 2cf20d7ac8b56b566fb25856325f21f86b521896 Mon Sep 17 00:00:00 2001 From: Majyles <123649730+Majyles@users.noreply.github.com> Date: Mon, 26 Oct 2026 22:30:55 +0100 Subject: [PATCH 2/4] fix(alerts): address review of the per-day alert count - Check the period length on the dates, before listing the days. - Filter started_at on UTC ranges: day boundaries are computed in Python with zoneinfo and days are bucketed with a CASE, so Postgres no longer converts timezones (index-friendly, consistent tz validation). - Clip the period to today in tz: no risk lookup for future days. - RiskService: shared httpx client closed in the app lifespan, and a semaphore bounding concurrent calls. - Share the sequence filter of /all/fromdate in _filter_alerts_by_sequences. - Send telemetry after validation, chain the tz exception with from None, inline get_token in tests. Co-Authored-By: Claude Opus 5.5 --- src/app/api/api_v1/endpoints/alerts.py | 83 ++++++++++++++++---------- src/app/main.py | 1 + src/app/services/risk.py | 22 +++++-- src/tests/endpoints/test_alerts.py | 65 ++++++++++++++++---- src/tests/services/test_risk.py | 45 ++++++++++++-- src/tests/test_main.py | 6 +- 6 files changed, 168 insertions(+), 54 deletions(-) diff --git a/src/app/api/api_v1/endpoints/alerts.py b/src/app/api/api_v1/endpoints/alerts.py index 52b702bc..2151b168 100644 --- a/src/app/api/api_v1/endpoints/alerts.py +++ b/src/app/api/api_v1/endpoints/alerts.py @@ -7,14 +7,14 @@ import asyncio import csv import io -from datetime import date, datetime, time, timedelta -from typing import Any, AsyncIterator, Dict, Iterable, Iterator, List, Tuple, Union, cast +from datetime import date, datetime, time, timedelta, timezone +from typing import Any, AsyncIterator, Dict, Iterable, Iterator, List, Tuple, TypeVar, Union, cast from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from fastapi import APIRouter, Depends, HTTPException, Path, Query, Security, status from fastapi.responses import StreamingResponse -from sqlalchemy import asc, desc, tuple_ -from sqlalchemy.sql import ColumnElement +from sqlalchemy import and_, asc, case, desc, tuple_ +from sqlalchemy.sql import ColumnElement, Select from sqlmodel import delete, func, select from sqlmodel.ext.asyncio.session import AsyncSession from sqlmodel.sql.expression import SelectOfScalar @@ -127,13 +127,23 @@ async def _build_alerts_from_date_stmt( alerts_stmt: Any = select(Alert).where(func.date(Alert.started_at) == from_date) if not is_admin: alerts_stmt = alerts_stmt.where(Alert.organization_id == organization_id) - if seq_filter is not None: - seq_match: Any = select(AlertSequence.alert_id).join( - Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id) - ) - seq_match = seq_match.where(seq_filter) - alerts_stmt = alerts_stmt.where(cast(Any, Alert.id).in_(seq_match)) - return alerts_stmt, seq_filter + return _filter_alerts_by_sequences(alerts_stmt, seq_filter), seq_filter + + +_SelectT = TypeVar("_SelectT", bound=Select) + + +def _filter_alerts_by_sequences(alerts_stmt: _SelectT, seq_filter: Union[ColumnElement[bool], None]) -> _SelectT: + """Keep the alerts with at least one sequence passing ``seq_filter`` (no-op when None).""" + if seq_filter is None: + return alerts_stmt + seq_match: Any = select(AlertSequence.alert_id).join(Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id)) + return alerts_stmt.where(cast(Any, Alert.id).in_(seq_match.where(seq_filter))) + + +def _utc_start_of_day(day: date, zone: ZoneInfo) -> datetime: + """Start of ``day`` in ``zone``, as a naive UTC datetime like the stored timestamps.""" + return datetime.combine(day, time.min, tzinfo=zone).astimezone(timezone.utc).replace(tzinfo=None) def _serialize_sequence(sequence: Sequence, detections_count: int = 0) -> SequenceRead: @@ -480,26 +490,33 @@ async def count_alerts_per_day( session: AsyncSession = Depends(get_session), token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]), ) -> List[AlertDayCount]: - telemetry_client.capture( - token_payload.sub, - event="alerts-count-per-day", - properties={"from_date": from_date.isoformat(), "to_date": to_date.isoformat(), "tz": tz}, - ) if to_date < from_date: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="to_date must be on or after from_date", ) - days = [from_date + timedelta(days=offset) for offset in range((to_date - from_date).days + 1)] - if len(days) > MAX_COUNT_PER_DAY_PERIOD: + if (to_date - from_date).days >= MAX_COUNT_PER_DAY_PERIOD: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"The period cannot exceed {MAX_COUNT_PER_DAY_PERIOD} days", ) try: - ZoneInfo(tz) + zone = ZoneInfo(tz) except (ZoneInfoNotFoundError, ValueError): - raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Unknown timezone: {tz}") + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Unknown timezone: {tz}" + ) from None + telemetry_client.capture( + token_payload.sub, + event="alerts-count-per-day", + properties={"from_date": from_date.isoformat(), "to_date": to_date.isoformat(), "tz": tz}, + ) + + # Days after today cannot have alerts yet: skip them, and their risk lookups + last_day = min(to_date, datetime.now(zone).date()) + days = [from_date + timedelta(days=offset) for offset in range((last_day - from_date).days + 1)] + if not days: + return [] # Same alerts as /all/fromdate/count, each day with its own risk scores: days sharing the # same scores are counted by a single query. @@ -518,20 +535,26 @@ async def count_alerts_per_day( for day, scores in zip(days, scores_per_day, strict=True): days_by_scores.setdefault(tuple(sorted(scores.items())), []).append(day) - # started_at is a naive UTC datetime: convert it to the requested timezone before taking its day - local_day: Any = func.date(func.timezone(tz, func.timezone("UTC", Alert.started_at))) counts: Dict[date, int] = {} for scores_key, scores_days in days_by_scores.items(): - stmt: Any = select(local_day, func.count()).where(local_day.in_(scores_days)).group_by(local_day) + # Day boundaries are computed here with the validated timezone, so the query only filters + # started_at on UTC ranges: index-friendly, and no timezone conversion in the database. + bounds = [ + (day, _utc_start_of_day(day, zone), _utc_start_of_day(day + timedelta(days=1), zone)) for day in scores_days + ] + local_day: Any = cast(Any, case)( + *((and_(Alert.started_at >= start, Alert.started_at < end), day) for day, start, end in bounds) + ).label("day") + stmt: Any = ( + select(local_day, func.count()) + .where(Alert.started_at >= bounds[0][1], Alert.started_at < bounds[-1][2]) + .group_by("day") + ) if not token_payload.is_admin: stmt = stmt.where(Alert.organization_id == token_payload.organization_id) - seq_filter = max_conf_filter_clause(dict(scores_key)) - if seq_filter is not None: - seq_match: Any = select(AlertSequence.alert_id).join( - Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id) - ) - stmt = stmt.where(cast(Any, Alert.id).in_(seq_match.where(seq_filter))) - counts.update((day, count) for day, count in (await session.exec(stmt)).all()) + stmt = _filter_alerts_by_sequences(stmt, max_conf_filter_clause(dict(scores_key))) + # Days of the range counted by another query (other risk scores) fall outside the CASE + counts.update((day, count) for day, count in (await session.exec(stmt)).all() if day is not None) return [AlertDayCount(date=day, count=count) for day, count in sorted(counts.items())] diff --git a/src/app/main.py b/src/app/main.py index b4bb7437..e950ed72 100644 --- a/src/app/main.py +++ b/src/app/main.py @@ -88,6 +88,7 @@ async def lifespan(_: FastAPI): with contextlib.suppress(asyncio.CancelledError): await task await temporal_service.aclose() + await risk_service.aclose() app = FastAPI( diff --git a/src/app/services/risk.py b/src/app/services/risk.py index deca342f..d3813578 100644 --- a/src/app/services/risk.py +++ b/src/app/services/risk.py @@ -3,6 +3,7 @@ # This program is licensed under the Apache License 2.0. # See LICENSE or go to for full license details. +import asyncio import logging from datetime import date from typing import Literal, Union, cast @@ -62,8 +63,19 @@ class RiskService: Fail-open: if the API is unreachable or a camera is unknown, no filter is applied. """ + # Bounds concurrent calls, e.g. one score lookup per day of a calendar month + MAX_CONCURRENT_REQUESTS: int = 8 + def __init__(self) -> None: self._scores: dict[int, str] = {} + self._client: Union[httpx.AsyncClient, None] = None + self._semaphore = asyncio.Semaphore(self.MAX_CONCURRENT_REQUESTS) + + async def aclose(self) -> None: + """Close the HTTP client on shutdown.""" + if self._client is not None: + await self._client.aclose() + self._client = None @property def is_configured(self) -> bool: @@ -89,14 +101,16 @@ async def _fetch(self, path: str, params: Union[dict, None] = None) -> object: return None host = risk_api_url.rstrip("/") try: - async with httpx.AsyncClient(timeout=5.0) as client: - response = await client.get( + async with self._semaphore: + if self._client is None: + self._client = httpx.AsyncClient(timeout=5.0) + response = await self._client.get( f"{host}/{path.lstrip('/')}", params=params, auth=(risk_api_login, risk_api_pwd), ) - response.raise_for_status() - return response.json() + response.raise_for_status() + return response.json() except (httpx.HTTPError, ValueError) as exc: logger.warning("Risk API call %s failed (%s)", path, exc) return None diff --git a/src/tests/endpoints/test_alerts.py b/src/tests/endpoints/test_alerts.py index cf319893..59b352e8 100644 --- a/src/tests/endpoints/test_alerts.py +++ b/src/tests/endpoints/test_alerts.py @@ -351,13 +351,14 @@ async def _create_alert_started_at( return alert -def _user_auth(user_idx: int) -> Dict[str, str]: - user = pytest.user_table[user_idx] - return pytest.get_token(user["id"], user["role"].split(), user["organization_id"]) - - @pytest.mark.asyncio async def test_alerts_count_per_day(async_client: AsyncClient, camera_session: AsyncSession): + admin_auth = pytest.get_token( + pytest.user_table[0]["id"], pytest.user_table[0]["role"].split(), pytest.user_table[0]["organization_id"] + ) + agent_auth = pytest.get_token( + pytest.user_table[1]["id"], pytest.user_table[1]["role"].split(), pytest.user_table[1]["organization_id"] + ) day = datetime(2025, 7, 14) await _create_alert_started_at(camera_session, 1, day + timedelta(hours=10)) await _create_alert_started_at(camera_session, 1, day + timedelta(hours=23, minutes=30)) @@ -366,19 +367,19 @@ async def test_alerts_count_per_day(async_client: AsyncClient, camera_session: A period = "from_date=2025-07-14&to_date=2025-07-16" # Agent: its organization only, days without alerts are omitted - resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=_user_auth(1)) + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=agent_auth) assert resp.status_code == 200, resp.text assert resp.json() == [{"date": "2025-07-14", "count": 2}, {"date": "2025-07-16", "count": 1}] # Consistent with the single-date count - resp = await async_client.get("/alerts/all/fromdate/count?from_date=2025-07-14", headers=_user_auth(1)) + resp = await async_client.get("/alerts/all/fromdate/count?from_date=2025-07-14", headers=agent_auth) assert resp.json() == {"count": 2} # Admin: every organization - resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=_user_auth(0)) + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=admin_auth) assert resp.json() == [{"date": "2025-07-14", "count": 3}, {"date": "2025-07-16", "count": 1}] # 23:30 UTC is already the next day in Paris (UTC+2 in July) - resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}&tz=Europe/Paris", headers=_user_auth(1)) + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}&tz=Europe/Paris", headers=agent_auth) assert resp.json() == [ {"date": "2025-07-14", "count": 1}, {"date": "2025-07-15", "count": 1}, @@ -386,7 +387,7 @@ async def test_alerts_count_per_day(async_client: AsyncClient, camera_session: A ] resp = await async_client.get( - "/alerts/all/fromdate/count-per-day?from_date=2025-07-15&to_date=2025-07-15", headers=_user_auth(1) + "/alerts/all/fromdate/count-per-day?from_date=2025-07-15&to_date=2025-07-15", headers=agent_auth ) assert resp.json() == [] @@ -397,6 +398,7 @@ async def test_alerts_count_per_day(async_client: AsyncClient, camera_session: A ("from_date=2025-07-16&to_date=2025-07-14", 422), ("from_date=2025-07-01&to_date=2025-09-01", 422), # 63 days ("from_date=2025-07-01&to_date=2025-08-31", 200), # 62 days + ("from_date=0001-01-01&to_date=9999-12-31", 422), # rejected before listing the days ("from_date=2025-07-01&to_date=2025-07-31&tz=Mars/Olympus_Mons", 422), ("from_date=2025-07-01", 422), ], @@ -405,7 +407,10 @@ async def test_alerts_count_per_day(async_client: AsyncClient, camera_session: A async def test_alerts_count_per_day_validation( async_client: AsyncClient, camera_session: AsyncSession, query: str, status_code: int ): - resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{query}", headers=_user_auth(1)) + agent_auth = pytest.get_token( + pytest.user_table[1]["id"], pytest.user_table[1]["role"].split(), pytest.user_table[1]["organization_id"] + ) + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{query}", headers=agent_auth) assert resp.status_code == status_code, resp.text @@ -417,6 +422,9 @@ async def test_alerts_count_per_day_unauthenticated(async_client: AsyncClient): @pytest.mark.asyncio async def test_alerts_count_per_day_risk_scores_per_day(async_client: AsyncClient, camera_session: AsyncSession): + agent_auth = pytest.get_token( + pytest.user_table[1]["id"], pytest.user_table[1]["role"].split(), pytest.user_table[1]["organization_id"] + ) camera_id = pytest.camera_table[0]["id"] day = datetime(2025, 7, 14) # Two low-confidence alerts, but only the first day is low risk (threshold above 0.2) @@ -429,7 +437,7 @@ def scores_for_date(target_date, organization_id=None): scores_mock = AsyncMock(side_effect=scores_for_date) with patch("app.api.api_v1.endpoints.alerts.risk_service.get_scores_for_date", new=scores_mock): - resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=_user_auth(1)) + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=agent_auth) assert resp.status_code == 200, resp.text assert resp.json() == [{"date": "2025-07-15", "count": 1}] assert scores_mock.await_count == 2 @@ -438,7 +446,7 @@ def scores_for_date(target_date, organization_id=None): # A risk score override applies to every day and skips the risk API scores_mock.reset_mock() resp = await async_client.get( - f"/alerts/all/fromdate/count-per-day?{period}&risk_score=moderate", headers=_user_auth(1) + f"/alerts/all/fromdate/count-per-day?{period}&risk_score=moderate", headers=agent_auth ) assert resp.json() == [{"date": "2025-07-14", "count": 1}, {"date": "2025-07-15", "count": 1}] scores_mock.assert_not_awaited() @@ -1147,3 +1155,34 @@ async def test_alert_changes_roll_back_on_late_failure( assert set(links) == set(sequence_ids) detections = (await observer.exec(select(Detection).where(Detection.sequence_id == sid))).all() assert len(detections) == counts[0] + + +@pytest.mark.asyncio +async def test_alerts_count_per_day_skips_future_days(async_client: AsyncClient, camera_session: AsyncSession): + agent_auth = pytest.get_token( + pytest.user_table[1]["id"], pytest.user_table[1]["role"].split(), pytest.user_table[1]["organization_id"] + ) + today = utcnow().date() + # Midnight today: always in the past and always today, whenever the test runs + await _create_alert_started_at(camera_session, 1, datetime.combine(today, datetime.min.time())) + period = f"from_date={(today - timedelta(days=2)).isoformat()}&to_date={(today + timedelta(days=10)).isoformat()}" + + scores_mock = AsyncMock(return_value={}) + with patch("app.api.api_v1.endpoints.alerts.risk_service.get_scores_for_date", new=scores_mock): + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=agent_auth) + assert resp.status_code == 200, resp.text + assert resp.json() == [{"date": today.isoformat(), "count": 1}] + # Risk scores are only fetched up to today + assert sorted(call.args[0] for call in scores_mock.await_args_list) == [ + today - timedelta(days=2), + today - timedelta(days=1), + today, + ] + + # A period entirely in the future is empty, without any lookup + scores_mock.reset_mock() + future = f"from_date={(today + timedelta(days=1)).isoformat()}&to_date={(today + timedelta(days=5)).isoformat()}" + with patch("app.api.api_v1.endpoints.alerts.risk_service.get_scores_for_date", new=scores_mock): + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{future}", headers=agent_auth) + assert resp.json() == [] + scores_mock.assert_not_awaited() diff --git a/src/tests/services/test_risk.py b/src/tests/services/test_risk.py index 5f5085e2..22b3d639 100644 --- a/src/tests/services/test_risk.py +++ b/src/tests/services/test_risk.py @@ -3,6 +3,7 @@ # This program is licensed under the Apache License 2.0. # See LICENSE or go to for full license details. +import asyncio from datetime import date from unittest.mock import AsyncMock, MagicMock, patch @@ -14,7 +15,7 @@ def _fake_httpx_client(*, json_data=None, raise_exc=None, side_effect_per_call=None): - """Return a context manager replacing ``app.services.risk.httpx.AsyncClient``. + """Return a factory replacing ``app.services.risk.httpx.AsyncClient``, and the client it builds. Either pass ``json_data`` (single response), ``raise_exc`` (raised on get()), or ``side_effect_per_call`` (list of (json_or_exc) values returned in order). @@ -40,11 +41,9 @@ def _fake_httpx_client(*, json_data=None, raise_exc=None, side_effect_per_call=N else: inner.get = AsyncMock(return_value=response) - cm = MagicMock() - cm.__aenter__ = AsyncMock(return_value=inner) - cm.__aexit__ = AsyncMock(return_value=None) + inner.aclose = AsyncMock() - factory = MagicMock(return_value=cm) + factory = MagicMock(return_value=inner) return factory, inner @@ -176,3 +175,39 @@ async def test_get_scores_for_date_no_org_param_when_none(configured_risk): await service.get_scores_for_date(date(2026, 5, 5)) _, kwargs = inner.get.await_args assert kwargs["params"] is None + + +@pytest.mark.asyncio +async def test_risk_client_is_shared_and_closed(configured_risk): + service = RiskService() + factory, inner = _fake_httpx_client(json_data=[]) + with patch("app.services.risk.httpx.AsyncClient", factory): + await service.get_scores_for_date(date(2026, 5, 5)) + await service.get_scores_for_date(date(2026, 5, 6)) + await service.aclose() + factory.assert_called_once() + assert inner.get.await_count == 2 + inner.aclose.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_risk_concurrent_requests_are_bounded(configured_risk): + service = RiskService() + factory, inner = _fake_httpx_client(json_data=[]) + in_flight, peak = 0, 0 + + async def slow_get(*args, **kwargs): + nonlocal in_flight, peak + in_flight += 1 + peak = max(peak, in_flight) + await asyncio.sleep(0.01) + in_flight -= 1 + response = MagicMock() + response.json = MagicMock(return_value=[]) + return response + + inner.get = AsyncMock(side_effect=slow_get) + with patch("app.services.risk.httpx.AsyncClient", factory): + await asyncio.gather(*(service.get_scores_for_date(date(2026, 5, day)) for day in range(1, 31))) + assert inner.get.await_count == 30 + assert peak == RiskService.MAX_CONCURRENT_REQUESTS diff --git a/src/tests/test_main.py b/src/tests/test_main.py index 99074569..aa2a59f7 100644 --- a/src/tests/test_main.py +++ b/src/tests/test_main.py @@ -115,7 +115,7 @@ async def test_risk_refresh_loop_swallows_refresh_errors_and_continues(): @pytest.mark.asyncio async def test_lifespan_refreshes_and_cancels_background_tasks_when_risk_api_configured(): - fake_service = SimpleNamespace(is_configured=True, refresh=AsyncMock()) + fake_service = SimpleNamespace(is_configured=True, refresh=AsyncMock(), aclose=AsyncMock()) fake_tasks = [_CancelledTask(), _CancelledTask()] def fake_create_task(coro): @@ -133,11 +133,12 @@ def fake_create_task(coro): assert all(task.cancel_called is False for task in fake_tasks) assert all(task.cancel_called is True for task in fake_tasks) + fake_service.aclose.assert_awaited_once() @pytest.mark.asyncio async def test_lifespan_skips_risk_refresh_but_starts_validation_worker_when_risk_api_not_configured(): - fake_service = SimpleNamespace(is_configured=False, refresh=AsyncMock()) + fake_service = SimpleNamespace(is_configured=False, refresh=AsyncMock(), aclose=AsyncMock()) fake_task = _CancelledTask() def fake_create_task(coro): @@ -157,6 +158,7 @@ def fake_create_task(coro): assert fake_task.cancel_called is True close_client.assert_awaited_once() + fake_service.aclose.assert_awaited_once() @pytest.mark.asyncio From 0f756dc6075b54f20bb24066b2bb5fca737e03d1 Mon Sep 17 00:00:00 2001 From: Majyles <123649730+Majyles@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:53:01 +0100 Subject: [PATCH 3/4] refactor(alerts): count alerts per local day in Python Select started_at on the plain UTC range and count the alerts per local day with zoneinfo, instead of a CASE over every day boundary: simpler to read, and the timezone is only handled by zoneinfo. Also make the period length check read as a number of days. Co-Authored-By: Claude Opus 5.5 --- src/app/api/api_v1/endpoints/alerts.py | 29 ++++++++++++-------------- 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/src/app/api/api_v1/endpoints/alerts.py b/src/app/api/api_v1/endpoints/alerts.py index 2151b168..2600e92e 100644 --- a/src/app/api/api_v1/endpoints/alerts.py +++ b/src/app/api/api_v1/endpoints/alerts.py @@ -13,7 +13,7 @@ from fastapi import APIRouter, Depends, HTTPException, Path, Query, Security, status from fastapi.responses import StreamingResponse -from sqlalchemy import and_, asc, case, desc, tuple_ +from sqlalchemy import asc, desc, tuple_ from sqlalchemy.sql import ColumnElement, Select from sqlmodel import delete, func, select from sqlmodel.ext.asyncio.session import AsyncSession @@ -495,7 +495,7 @@ async def count_alerts_per_day( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="to_date must be on or after from_date", ) - if (to_date - from_date).days >= MAX_COUNT_PER_DAY_PERIOD: + if (to_date - from_date).days + 1 > MAX_COUNT_PER_DAY_PERIOD: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"The period cannot exceed {MAX_COUNT_PER_DAY_PERIOD} days", @@ -537,24 +537,21 @@ async def count_alerts_per_day( counts: Dict[date, int] = {} for scores_key, scores_days in days_by_scores.items(): - # Day boundaries are computed here with the validated timezone, so the query only filters - # started_at on UTC ranges: index-friendly, and no timezone conversion in the database. - bounds = [ - (day, _utc_start_of_day(day, zone), _utc_start_of_day(day + timedelta(days=1), zone)) for day in scores_days - ] - local_day: Any = cast(Any, case)( - *((and_(Alert.started_at >= start, Alert.started_at < end), day) for day, start, end in bounds) - ).label("day") - stmt: Any = ( - select(local_day, func.count()) - .where(Alert.started_at >= bounds[0][1], Alert.started_at < bounds[-1][2]) - .group_by("day") + # The database returns the alerts of a plain UTC range of started_at (index-friendly), and + # they are counted per local day here: the timezone is only handled by zoneinfo. + stmt: Any = select(Alert.started_at).where( + Alert.started_at >= _utc_start_of_day(scores_days[0], zone), + Alert.started_at < _utc_start_of_day(scores_days[-1] + timedelta(days=1), zone), ) if not token_payload.is_admin: stmt = stmt.where(Alert.organization_id == token_payload.organization_id) stmt = _filter_alerts_by_sequences(stmt, max_conf_filter_clause(dict(scores_key))) - # Days of the range counted by another query (other risk scores) fall outside the CASE - counts.update((day, count) for day, count in (await session.exec(stmt)).all() if day is not None) + for started_at in (await session.exec(stmt)).all(): + # started_at is a naive UTC datetime: its day in the requested timezone + day = started_at.replace(tzinfo=timezone.utc).astimezone(zone).date() + # The range may include days counted by another query (other risk scores) + if day in scores_days: + counts[day] = counts.get(day, 0) + 1 return [AlertDayCount(date=day, count=count) for day, count in sorted(counts.items())] From deee618b5f5443b7e41a4a59742ee7d23c9ae2f4 Mon Sep 17 00:00:00 2001 From: Majyles <123649730+Majyles@users.noreply.github.com> Date: Fri, 30 Oct 2026 01:00:27 +0100 Subject: [PATCH 4/4] feat(db): index alerts by organization and start time Co-Authored-By: Claude Opus 5.5 --- src/app/models.py | 2 ++ ...3f_index_alerts_organization_started_at.py | 25 +++++++++++++++++++ 2 files changed, 27 insertions(+) create mode 100644 src/migrations/versions/2026_10_29_1000-ff5b6d1f9e3f_index_alerts_organization_started_at.py diff --git a/src/app/models.py b/src/app/models.py index 599531a8..829588ad 100644 --- a/src/app/models.py +++ b/src/app/models.py @@ -227,6 +227,8 @@ class Sequence(SQLModel, table=True): class Alert(SQLModel, table=True): __tablename__ = "alerts" + # Alerts of an organization over a period (History, per-day counts) + __table_args__ = (Index("ix_alerts_organization_id_started_at", "organization_id", "started_at"),) id: int = Field(None, primary_key=True) organization_id: int = Field(..., foreign_key="organizations.id", nullable=False) lat: Union[float, None] = Field(default=None, gt=-90, lt=90, nullable=True) diff --git a/src/migrations/versions/2026_10_29_1000-ff5b6d1f9e3f_index_alerts_organization_started_at.py b/src/migrations/versions/2026_10_29_1000-ff5b6d1f9e3f_index_alerts_organization_started_at.py new file mode 100644 index 00000000..dd958f75 --- /dev/null +++ b/src/migrations/versions/2026_10_29_1000-ff5b6d1f9e3f_index_alerts_organization_started_at.py @@ -0,0 +1,25 @@ +"""index alerts by organization and start time + +Revision ID: ff5b6d1f9e3f +Revises: e7a4b9c3d5f2 +Create Date: 2026-10-29 10:00:00.000000 + +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = "ff5b6d1f9e3f" +down_revision: Union[str, None] = "e7a4b9c3d5f2" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Deployment stops the backend before migration; transactional builds avoid invalid indexes. + op.create_index("ix_alerts_organization_id_started_at", "alerts", ["organization_id", "started_at"]) + + +def downgrade() -> None: + op.drop_index("ix_alerts_organization_id_started_at", table_name="alerts")