feat: add workspaces list endpoint by provider id (#1099)

* feat: add workspaces list endpoint by provider id

* leftover

* refactor: simplify return values

Author: peppescg

src/codegate/api/v1.py

@@ -12,7 +12,7 @@
 from codegate import __version__
 from codegate.api import v1_models, v1_processing
 from codegate.db.connection import AlreadyExistsError, DbReader
-from codegate.db.models import AlertSeverity
+from codegate.db.models import AlertSeverity, WorkspaceWithModel
 from codegate.providers import crud as provendcrud
 from codegate.workspaces import crud
 
@@ -532,6 +532,22 @@ async def set_workspace_muxes(
     return Response(status_code=204)
 
 
+@v1.get(
+    "/workspaces/{provider_id}",
+    tags=["Workspaces"],
+    generate_unique_id_function=uniq_name,
+)
+async def list_workspaces_by_provider(
+    provider_id: UUID,
+) -> List[WorkspaceWithModel]:
+    """List workspaces by provider ID."""
+    try:
+        return await wscrud.workspaces_by_provider(provider_id)
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
 @v1.get("/alerts_notification", tags=["Dashboard"], generate_unique_id_function=uniq_name)
 async def stream_sse():
     """

src/codegate/db/connection.py

@@ -28,6 +28,7 @@
     ProviderModel,
     Session,
     WorkspaceRow,
+    WorkspaceWithModel,
     WorkspaceWithSessionInfo,
 )
 from codegate.db.token_usage import TokenUsageParser
@@ -721,6 +722,23 @@ async def get_workspace_by_name(self, name: str) -> Optional[WorkspaceRow]:
         )
         return workspaces[0] if workspaces else None
 
+    async def get_workspaces_by_provider(self, provider_id: str) -> List[WorkspaceWithModel]:
+        sql = text(
+            """
+            SELECT
+                w.id, w.name, m.provider_model_name
+            FROM workspaces w
+            JOIN muxes m ON w.id = m.workspace_id
+            WHERE m.provider_endpoint_id = :provider_id
+            AND w.deleted_at IS NULL
+            """
+        )
+        conditions = {"provider_id": provider_id}
+        workspaces = await self._exec_select_conditions_to_pydantic(
+            WorkspaceWithModel, sql, conditions, should_raise=True
+        )
+        return workspaces
+
     async def get_archived_workspace_by_name(self, name: str) -> Optional[WorkspaceRow]:
         sql = text(
             """

src/codegate/db/models.py

@@ -189,6 +189,14 @@ class WorkspaceWithSessionInfo(BaseModel):
     session_id: Optional[str]
 
 
+class WorkspaceWithModel(BaseModel):
+    """Returns a workspace ID with model name"""
+
+    id: str
+    name: WorkspaceNameStr
+    provider_model_name: str
+
+
 class ActiveWorkspace(BaseModel):
     """Returns a full active workspace object with the
     with the session information.

src/codegate/workspaces/crud.py

@@ -218,6 +218,13 @@ async def get_workspace_by_name(self, workspace_name: str) -> db_models.Workspac
             raise WorkspaceDoesNotExistError(f"Workspace {workspace_name} does not exist.")
         return workspace
 
+    async def workspaces_by_provider(self, provider_id: uuid) -> List[db_models.WorkspaceWithModel]:
+        """Get the workspaces by provider."""
+
+        workspaces = await self._db_reader.get_workspaces_by_provider(str(provider_id))
+
+        return workspaces
+
     async def get_muxes(self, workspace_name: str) -> List[mux_models.MuxRule]:
         # Verify if workspace exists
         workspace = await self._db_reader.get_workspace_by_name(workspace_name)