feat: add pagination to alerts and messages endpoints

Modify the API to add pagination to those endpoints, to be able
to render faster in the browser

Closes: #1020

Author: yrobla

src/codegate/api/v1.py

@@ -1,13 +1,14 @@
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 from uuid import UUID
 
 import requests
 import structlog
-from fastapi import APIRouter, Depends, HTTPException, Response
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
 from fastapi.responses import StreamingResponse
 from fastapi.routing import APIRoute
 from pydantic import BaseModel, ValidationError
 
+from codegate.config import API_DEFAULT_PAGE_SIZE, API_MAX_PAGE_SIZE
 import codegate.muxing.models as mux_models
 from codegate import __version__
 from codegate.api import v1_models, v1_processing
@@ -378,7 +379,11 @@ async def hard_delete_workspace(workspace_name: str):
     tags=["Workspaces"],
     generate_unique_id_function=uniq_name,
 )
-async def get_workspace_alerts(workspace_name: str) -> List[Optional[v1_models.AlertConversation]]:
+async def get_workspace_alerts(
+    workspace_name: str,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(API_DEFAULT_PAGE_SIZE, get=1, le=API_MAX_PAGE_SIZE),
+) -> Dict[str, Any]:
     """Get alerts for a workspace."""
     try:
         ws = await wscrud.get_workspace_by_name(workspace_name)
@@ -388,13 +393,35 @@ async def get_workspace_alerts(workspace_name: str) -> List[Optional[v1_models.A
         logger.exception("Error while getting workspace")
         raise HTTPException(status_code=500, detail="Internal server error")
 
-    try:
-        alerts = await dbreader.get_alerts_by_workspace(ws.id, AlertSeverity.CRITICAL.value)
-        prompts_outputs = await dbreader.get_prompts_with_output(ws.id)
-        return await v1_processing.parse_get_alert_conversation(alerts, prompts_outputs)
-    except Exception:
-        logger.exception("Error while getting alerts and messages")
-        raise HTTPException(status_code=500, detail="Internal server error")
+    total_alerts = 0
+    fetched_alerts = []
+    offset = (page - 1) * page_size
+    batch_size = page_size * 2  # fetch more alerts per batch to allow deduplication
+
+    while len(fetched_alerts) < page_size:
+        alerts_batch, total_alerts = await dbreader.get_alerts_by_workspace(
+            ws.id, AlertSeverity.CRITICAL.value, page_size, offset
+        )
+        if not alerts_batch:
+            break
+
+        dedup_alerts = await v1_processing.remove_duplicate_alerts(alerts_batch)
+        fetched_alerts.extend(dedup_alerts)
+        offset += batch_size
+
+    final_alerts = fetched_alerts[:page_size]
+    prompt_ids = list({alert.prompt_id for alert in final_alerts if alert.prompt_id})
+    prompts_outputs = await dbreader.get_prompts_with_output(prompt_ids)
+    alert_conversations = await v1_processing.parse_get_alert_conversation(
+        final_alerts, prompts_outputs
+    )
+    return {
+        "page": page,
+        "page_size": page_size,
+        "total_alerts": total_alerts,
+        "total_pages": (total_alerts + page_size - 1) // page_size,
+        "alerts": alert_conversations,
+    }
 
 
 @v1.get(

src/codegate/config.py

@@ -25,6 +25,9 @@
     "llamacpp": "./codegate_volume/models",  # Default LlamaCpp model path
 }
 
+API_DEFAULT_PAGE_SIZE = 50
+API_MAX_PAGE_SIZE = 100
+
 
 @dataclass
 class Config:

src/codegate/db/connection.py

@@ -2,7 +2,7 @@
 import json
 import uuid
 from pathlib import Path
-from typing import Dict, List, Optional, Type
+from typing import Dict, List, Optional, Tuple, Type
 
 import structlog
 from alembic import command as alembic_command
@@ -13,6 +13,7 @@
 from sqlalchemy.exc import IntegrityError, OperationalError
 from sqlalchemy.ext.asyncio import create_async_engine
 
+from codegate.config import API_DEFAULT_PAGE_SIZE
 from codegate.db.fim_cache import FimCache
 from codegate.db.models import (
     ActiveWorkspace,
@@ -569,7 +570,10 @@ async def _exec_select_conditions_to_pydantic(
                     raise e
                 return None
 
-    async def get_prompts_with_output(self, workpace_id: str) -> List[GetPromptWithOutputsRow]:
+    async def get_prompts_with_output(self, prompt_ids: List[str]) -> List[GetPromptWithOutputsRow]:
+        if not prompt_ids:
+            return []
+
         sql = text(
             """
             SELECT
@@ -583,11 +587,11 @@ async def get_prompts_with_output(self, workpace_id: str) -> List[GetPromptWithO
                 o.output_cost
             FROM prompts p
             LEFT JOIN outputs o ON p.id = o.prompt_id
-            WHERE p.workspace_id = :workspace_id
+            WHERE p.id IN :prompt_ids
             ORDER BY o.timestamp DESC
             """
         )
-        conditions = {"workspace_id": workpace_id}
+        conditions = {"prompt_ids": tuple(prompt_ids)}
         prompts = await self._exec_select_conditions_to_pydantic(
             GetPromptWithOutputsRow, sql, conditions, should_raise=True
         )
@@ -656,8 +660,12 @@ async def get_prompts_with_output_alerts_usage_by_workspace_id(
         return list(prompts_dict.values())
 
     async def get_alerts_by_workspace(
-        self, workspace_id: str, trigger_category: Optional[str] = None
-    ) -> List[Alert]:
+        self,
+        workspace_id: str,
+        trigger_category: Optional[str] = None,
+        limit: int = API_DEFAULT_PAGE_SIZE,
+        offset: int = 0,
+    ) -> Tuple[List[Alert], int]:
         sql = text(
             """
             SELECT
@@ -679,12 +687,28 @@ async def get_alerts_by_workspace(
             sql = text(sql.text + " AND a.trigger_category = :trigger_category")
             conditions["trigger_category"] = trigger_category
 
-        sql = text(sql.text + " ORDER BY a.timestamp DESC")
+        sql = text(sql.text + " ORDER BY a.timestamp DESC LIMIT :limit OFFSET :offset")
+        conditions["limit"] = limit
+        conditions["offset"] = offset
 
-        prompts = await self._exec_select_conditions_to_pydantic(
+        alerts = await self._exec_select_conditions_to_pydantic(
             Alert, sql, conditions, should_raise=True
         )
-        return prompts
+
+        # Count total alerts for pagination
+        count_sql = text(
+            """
+            SELECT COUNT(*)
+            FROM alerts a
+            INNER JOIN prompts p ON p.id = a.prompt_id
+            WHERE p.workspace_id = :workspace_id
+        """
+        )
+        if trigger_category:
+            count_sql = text(count_sql.text + " AND a.trigger_category = :trigger_category")
+
+        total_alerts = await self._exec_select_count(count_sql, conditions)
+        return alerts, total_alerts
 
     async def get_workspaces(self) -> List[WorkspaceWithSessionInfo]:
         sql = text(