perf: batch database reads and bound data loading

client/docs/source/changelog.rst

@@ -1,6 +1,15 @@
 Changelog
 =========
 
+Unreleased
+----------
+
+Detection listing is paginated: ``Client.fetch_detections(limit=100, offset=0)``
+returns up to 100 detections by default, with a maximum page size of 500. Advance
+``offset`` by the number of rows returned until a page is empty to retrieve the
+full list. Alert and sequence date-list endpoints accept page sizes from 1 to
+100 and nonnegative offsets.
+
 v0.1.2 (2022-07-01)
 -------------------
 Release note: `v0.1.2 <https://github.com/pyronear/pyro-api/releases/tag/v0.1.2>`_

client/pyroclient/client.py

@@ -422,8 +422,10 @@ def get_detection_url(self, detection_id: int) -> Response:
             timeout=self.timeout,
         )
 
-    def fetch_detections(self) -> Response:
-        """List the detections accessible to the authenticated user
+    def fetch_detections(self, limit: int = 100, offset: int = 0) -> Response:
+        """List a page of detections accessible to the authenticated user.
+
+        Pages default to 100 rows (maximum 500), ordered by ID. Advance offset to fetch more.
 
         >>> from pyroclient import client
         >>> api_client = Client("MY_USER_TOKEN")
@@ -434,6 +436,7 @@ def fetch_detections(self) -> Response:
         """
         return requests.get(
             urljoin(self._route_prefix, ClientRoute.DETECTIONS_FETCH),
+            params={"limit": limit, "offset": offset},
             headers=self.headers,
             timeout=self.timeout,
         )

client/tests/test_client.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from unittest.mock import Mock
 
 import pytest
 import requests
@@ -9,6 +10,14 @@
 from pyroclient.exceptions import HTTPRequestError
 
 
+def test_client_forwards_detection_pagination(monkeypatch):
+    get = Mock(return_value=Mock(status_code=200))
+    monkeypatch.setattr("pyroclient.client.requests.get", get)
+    client = Client("test-token", host="http://api.test")
+    client.fetch_detections(limit=25, offset=50)
+    assert get.call_args.kwargs["params"] == {"limit": 25, "offset": 50}
+
+
 @pytest.mark.parametrize(
     ("token", "host", "timeout", "expected_error"),
     [

src/app/api/api_v1/endpoints/alerts.py

@@ -7,11 +7,11 @@
 import csv
 import io
 from datetime import date, datetime, time, timedelta
-from typing import Any, Dict, Iterable, Iterator, List, Tuple, Union, cast
+from typing import Any, AsyncIterator, Dict, Iterable, Iterator, List, Tuple, Union, cast
 
 from fastapi import APIRouter, Depends, HTTPException, Path, Query, Security, status
 from fastapi.responses import StreamingResponse
-from sqlalchemy import asc, desc
+from sqlalchemy import asc, desc, tuple_
 from sqlalchemy.sql import ColumnElement
 from sqlmodel import delete, func, select
 from sqlmodel.ext.asyncio.session import AsyncSession
@@ -20,7 +20,7 @@
 from app.api.dependencies import get_alert_crud, get_camera_crud, get_jwt, get_sequence_crud
 from app.core.time import utcnow
 from app.crud import AlertCRUD, CameraCRUD, SequenceCRUD
-from app.db import get_session
+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.login import TokenPayload
@@ -32,6 +32,7 @@
 from app.services.telemetry import telemetry_client
 
 router = APIRouter()
+_ALERT_EXPORT_BATCH_SIZE = 100
 
 
 def verify_org_rights(organization_id: int, alert: Alert) -> None:
@@ -188,14 +189,6 @@ async def _serialize_alerts_page(
 }
 
 
-async def _fetch_camera_names_by_ids(session: AsyncSession, camera_ids: Iterable[int]) -> Dict[int, str]:
-    ids = list(set(camera_ids))
-    if not ids:
-        return {}
-    stmt: Any = select(Camera.id, Camera.name).where(cast(Any, Camera.id).in_(ids))
-    return {cid: name for cid, name in (await session.exec(stmt)).all()}
-
-
 def _alert_cells(alert: Alert) -> List[Any]:
     return [
         alert.id,
@@ -226,6 +219,8 @@ def _iter_alerts_csv(
     alerts: Iterable[Alert],
     seq_map: Dict[int, List[Sequence]],
     camera_names_by_id: Dict[int, str],
+    *,
+    include_header: bool = True,
 ) -> Iterator[str]:
     buf = io.StringIO()
     writer = csv.writer(buf)
@@ -236,8 +231,9 @@ def drain() -> str:
         buf.truncate(0)
         return value
 
-    writer.writerow(_ALERT_EXPORT_COLUMNS)
-    yield drain()
+    if include_header:
+        writer.writerow(_ALERT_EXPORT_COLUMNS)
+        yield drain()
 
     for alert in alerts:
         alert_cells = _alert_cells(alert)
@@ -248,20 +244,44 @@ def drain() -> str:
             yield drain()
 
 
-def _build_alerts_csv_response(
-    alerts: List[Alert],
-    seq_map: Dict[int, List[Sequence]],
-    camera_names_by_id: Dict[int, str],
-    from_date: date,
-    to_date: date,
-) -> StreamingResponse:
-    filename = f"alerts_{from_date.isoformat()}_{to_date.isoformat()}.csv"
-    headers = {"Content-Disposition": f'attachment; filename="{filename}"'}
-    return StreamingResponse(
-        _iter_alerts_csv(alerts, seq_map, camera_names_by_id),
-        media_type="text/csv",
-        headers=headers,
+async def _stream_alerts_csv(start_dt: datetime, end_dt: datetime, organization_id: int | None) -> AsyncIterator[str]:
+    # Own short-lived sessions: streaming can outlive request dependency cleanup and a
+    # slow download should not retain a database connection between batches.
+    cursor: tuple[datetime, int, datetime, int] | None = None
+    yield next(_iter_alerts_csv([], {}, {}))
+    order = (
+        cast(Any, Alert.started_at),
+        cast(Any, Alert.id),
+        cast(Any, Sequence.started_at),
+        cast(Any, Sequence.id),
     )
+    while True:
+        stmt: Any = (
+            select(Alert, Sequence, Camera.name)
+            .join(AlertSequence, cast(Any, AlertSequence.alert_id) == Alert.id)
+            .join(Sequence, cast(Any, Sequence.id) == AlertSequence.sequence_id)
+            .outerjoin(Camera, cast(Any, Camera.id) == Sequence.camera_id)
+            .where(Alert.started_at >= start_dt, Alert.started_at <= end_dt)
+            .order_by(*order)
+            .limit(_ALERT_EXPORT_BATCH_SIZE)
+        )
+        if organization_id is not None:
+            stmt = stmt.where(Alert.organization_id == organization_id)
+        if cursor is not None:
+            stmt = stmt.where(tuple_(*order) > tuple_(*cursor))
+        async with session_factory() as session:
+            rows = (await session.exec(stmt)).all()
+        if not rows:
+            return
+        for alert, sequence, camera_name in rows:
+            for line in _iter_alerts_csv(
+                [alert], {alert.id: [sequence]}, {sequence.camera_id: camera_name or ""}, include_header=False
+            ):
+                yield line
+        last_alert, last_sequence, _ = rows[-1]
+        cursor = (last_alert.started_at, last_alert.id, last_sequence.started_at, last_sequence.id)
+        if len(rows) < _ALERT_EXPORT_BATCH_SIZE:
+            return
 
 
 @router.get(
@@ -273,7 +293,6 @@ def _build_alerts_csv_response(
 async def export_alerts_csv(
     from_date: date = Query(..., description="Inclusive lower bound on started_at (UTC date)"),
     to_date: date = Query(..., description="Inclusive upper bound on started_at (UTC date)"),
-    session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
 ) -> StreamingResponse:
     telemetry_client.capture(
@@ -292,22 +311,12 @@ async def export_alerts_csv(
     start_dt = datetime.combine(from_date, time.min)
     end_dt = datetime.combine(to_date, time.max)
 
-    stmt: Any = (
-        select(Alert)
-        .where(Alert.started_at >= start_dt)
-        .where(Alert.started_at <= end_dt)
-        .order_by(Alert.started_at.asc())  # type: ignore[attr-defined]
-    )
-    # Admins export every organization's alerts
-    if not token_payload.is_admin:
-        stmt = stmt.where(Alert.organization_id == token_payload.organization_id)
-    alerts = list((await session.exec(stmt)).all())
-    seq_map = await _fetch_sequences_by_alert_ids(session, [alert.id for alert in alerts])
-    camera_names_by_id = await _fetch_camera_names_by_ids(
-        session,
-        (sequence.camera_id for sequences in seq_map.values() for sequence in sequences),
+    filename = f"alerts_{from_date.isoformat()}_{to_date.isoformat()}.csv"
+    return StreamingResponse(
+        _stream_alerts_csv(start_dt, end_dt, None if token_payload.is_admin else token_payload.organization_id),
+        media_type="text/csv",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
     )
-    return _build_alerts_csv_response(alerts, seq_map, camera_names_by_id, from_date, to_date)
 
 
 @router.get("/{alert_id}", status_code=status.HTTP_200_OK, summary="Fetch the information of a specific alert")
@@ -363,8 +372,8 @@ async def fetch_alert_sequences(
     summary="Fetch all the alerts with unlabeled sequences from the last 24 hours",
 )
 async def fetch_latest_unlabeled_alerts(
-    limit: Union[int, None] = Query(15, ge=1, description="Maximum number of alerts to fetch"),
-    offset: Union[int, None] = Query(0, description="Number of alerts to skip before starting to fetch"),
+    limit: int = Query(15, ge=1, le=100, description="Maximum number of alerts to fetch"),
+    offset: int = Query(0, ge=0, description="Number of alerts to skip before starting to fetch"),
     risk_score: Union[FwiClass, None] = Query(
         None,
         description="Override FWI class applied to every sequence; bypasses risk-api lookup. Ignored for admins.",
@@ -378,7 +387,7 @@ async def fetch_latest_unlabeled_alerts(
     alerts_stmt, seq_filter = await _build_unlabeled_alerts_stmt(
         session, token_payload.organization_id, risk_score, is_admin=is_admin
     )
-    alerts_stmt = alerts_stmt.order_by(Alert.started_at.desc()).limit(limit).offset(offset)  # type: ignore[attr-defined]
+    alerts_stmt = alerts_stmt.order_by(Alert.started_at.desc(), cast(Any, Alert.id).desc()).limit(limit).offset(offset)  # type: ignore[attr-defined]
     return await _serialize_alerts_page(session, alerts_stmt, seq_filter)
 
 
@@ -407,8 +416,8 @@ async def count_latest_unlabeled_alerts(
 @router.get("/all/fromdate", status_code=status.HTTP_200_OK, summary="Fetch all the alerts for a specific date")
 async def fetch_alerts_from_date(
     from_date: date = Query(),
-    limit: Union[int, None] = Query(15, description="Maximum number of alerts to fetch"),
-    offset: Union[int, None] = Query(0, description="Number of alerts to skip before starting to fetch"),
+    limit: int = Query(15, ge=1, le=100, description="Maximum number of alerts to fetch"),
+    offset: int = Query(0, ge=0, description="Number of alerts to skip before starting to fetch"),
     risk_score: Union[FwiClass, None] = Query(
         None,
         description="Override FWI class applied to every sequence; bypasses risk-api lookup. Ignored for admins.",
@@ -422,7 +431,7 @@ async def fetch_alerts_from_date(
     alerts_stmt, seq_filter = await _build_alerts_from_date_stmt(
         session, token_payload.organization_id, from_date, risk_score, is_admin=is_admin
     )
-    alerts_stmt = alerts_stmt.order_by(Alert.started_at.desc()).limit(limit).offset(offset)  # type: ignore[attr-defined]
+    alerts_stmt = alerts_stmt.order_by(Alert.started_at.desc(), cast(Any, Alert.id).desc()).limit(limit).offset(offset)  # type: ignore[attr-defined]
     return await _serialize_alerts_page(session, alerts_stmt, seq_filter)
 
 

src/app/api/api_v1/endpoints/cameras.py

@@ -6,6 +6,7 @@
 import asyncio
 from typing import Any, List, cast
 
+from anyio import CapacityLimiter, to_thread
 from fastapi import APIRouter, Depends, File, HTTPException, Path, Security, UploadFile, status
 
 from app.api.dependencies import get_camera_crud, get_jwt, get_pose_crud
@@ -32,6 +33,20 @@
 from app.services.telemetry import telemetry_client
 
 router = APIRouter()
+_image_url_limiter = CapacityLimiter(8)
+
+
+async def _last_image_url(organization_id: int, last_image: str | None) -> str | None:
+    if not last_image:
+        return None
+    bucket = await to_thread.run_sync(
+        s3_service.get_bucket, s3_service.resolve_bucket_name(organization_id), limiter=_image_url_limiter
+    )
+    exists = await to_thread.run_sync(bucket.check_file_existence, last_image, limiter=_image_url_limiter)
+    if not exists:
+        return None
+    # Presigning is local; keep its mutable URL cache on the event loop.
+    return bucket.get_public_url(last_image, verify_exists=False)
 
 
 @router.post("/", status_code=status.HTTP_201_CREATED, summary="Register a new camera")
@@ -70,11 +85,7 @@ async def get_camera(
     )
     pose_reads = [PoseReadWithoutImgInfo(**p.model_dump()) for p in cam_poses]
 
-    bucket = s3_service.get_bucket(s3_service.resolve_bucket_name(camera.organization_id))
-    try:
-        last_image_url = bucket.get_public_url(camera.last_image) if camera.last_image else None
-    except HTTPException:
-        last_image_url = None
+    last_image_url = await _last_image_url(camera.organization_id, camera.last_image)
     return CameraRead(**camera.model_dump(), last_image_url=last_image_url, poses=pose_reads)
 
 
@@ -91,46 +102,28 @@ async def fetch_cameras(
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
 ) -> List[CameraRead]:
     telemetry_client.capture(token_payload.sub, event="cameras-fetch")
-    trustable_filter: list[tuple[str, Any]] | None = None if include_non_trustable else [("is_trustable", True)]
-    if token_payload.is_admin:
-        cams = [elt for elt in await cameras.fetch_all(order_by="id", filters=trustable_filter)]
-
-        async def get_url_for_cam(cam: Camera) -> str | None:  # ruff:ignore[unused-async]
-            if cam.last_image:
-                bucket = s3_service.get_bucket(s3_service.resolve_bucket_name(cam.organization_id))
-                try:
-                    return bucket.get_public_url(cam.last_image)
-                except HTTPException:
-                    return None
-            return None
-
-        urls = await asyncio.gather(*[get_url_for_cam(cam) for cam in cams])
-    else:
-        bucket = s3_service.get_bucket(s3_service.resolve_bucket_name(token_payload.organization_id))
-        org_filters: list[tuple[str, Any]] = [("organization_id", token_payload.organization_id)]
-        if not include_non_trustable:
-            org_filters.append(("is_trustable", True))
-        cams = [elt for elt in await cameras.fetch_all(order_by="id", filters=org_filters)]
-
-        async def get_url_for_cam_single_bucket(cam: Camera) -> str | None:  # ruff:ignore[unused-async]
-            if cam.last_image:
-                try:
-                    return bucket.get_public_url(cam.last_image)
-                except HTTPException:
-                    return None
-            return None
-
-        urls = await asyncio.gather(*[get_url_for_cam_single_bucket(cam) for cam in cams])
-
-    async def get_poses(cam: Camera) -> list[PoseReadWithoutImgInfo]:
-        p = await poses.fetch_all(filters=[("camera_id", cam.id), ("active", True)])
-        return [PoseReadWithoutImgInfo(**pose.model_dump()) for pose in p]
-
-    poses_list = await asyncio.gather(*[get_poses(cam) for cam in cams])
+    filters: list[tuple[str, Any]] = []
+    if not include_non_trustable:
+        filters.append(("is_trustable", True))
+    if not token_payload.is_admin:
+        filters.append(("organization_id", token_payload.organization_id))
+    cams = await cameras.fetch_all(order_by="id", filters=filters)
+    if not cams:
+        return []
+
+    # One query for every camera; an AsyncSession cannot run concurrent pose queries.
+    active_poses = await poses.fetch_all(
+        filters=("active", True), in_pair=("camera_id", [cam.id for cam in cams]), order_by="id"
+    )
+    poses_by_camera: dict[int, list[PoseReadWithoutImgInfo]] = {}
+    for pose in active_poses:
+        poses_by_camera.setdefault(pose.camera_id, []).append(PoseReadWithoutImgInfo(**pose.model_dump()))
 
+    # S3 existence checks are blocking I/O; keep them off the event loop and cap concurrency.
+    urls = await asyncio.gather(*[_last_image_url(cam.organization_id, cam.last_image) for cam in cams])
     return [
-        CameraRead(**cam.model_dump(), last_image_url=url, poses=cam_poses)
-        for cam, url, cam_poses in zip(cams, urls, poses_list, strict=False)
+        CameraRead(**cam.model_dump(), last_image_url=url, poses=poses_by_camera.get(cam.id, []))
+        for cam, url in zip(cams, urls, strict=True)
     ]