endpoint to get full workspace config + free

Author: alex-mcgovern

src/codegate/api/v1.py

@@ -3,7 +3,7 @@
 
 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
@@ -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, WorkspaceWithModel
+from codegate.db.models import AlertSeverity
 from codegate.providers import crud as provendcrud
 from codegate.workspaces import crud
 
@@ -209,13 +209,32 @@ async def delete_provider_endpoint(
 
 
 @v1.get("/workspaces", tags=["Workspaces"], generate_unique_id_function=uniq_name)
-async def list_workspaces() -> v1_models.ListWorkspacesResponse:
-    """List all workspaces."""
-    wslist = await wscrud.get_workspaces()
+async def list_workspaces(
+    provider_id: Optional[UUID] = Query(None),
+) -> v1_models.ListWorkspacesResponse:
+    """
+    List all workspaces.
 
-    resp = v1_models.ListWorkspacesResponse.from_db_workspaces_with_sessioninfo(wslist)
+    Args:
+        provider_id (Optional[UUID]): Filter workspaces by provider ID. If provided,
+        will return workspaces where models from the specified provider (e.g., OpenAI,
+        Anthropic) have been used in workspace muxing rules. Note that you must
+        refer to a provider by ID, not by name.
 
-    return resp
+    Returns:
+        ListWorkspacesResponse: A response object containing the list of workspaces.
+    """
+    try:
+        if provider_id:
+            wslist = await wscrud.workspaces_by_provider(provider_id)
+            resp = v1_models.ListWorkspacesResponse.from_db_workspaces(wslist)
+            return resp
+        else:
+            wslist = await wscrud.get_workspaces()
+            resp = v1_models.ListWorkspacesResponse.from_db_workspaces_with_sessioninfo(wslist)
+            return resp
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
 
 
 @v1.get("/workspaces/active", tags=["Workspaces"], generate_unique_id_function=uniq_name)
@@ -584,17 +603,28 @@ async def set_workspace_muxes(
 
 
 @v1.get(
-    "/workspaces/{provider_id}",
+    "/workspaces/{workspace_name}",
     tags=["Workspaces"],
     generate_unique_id_function=uniq_name,
 )
-async def list_workspaces_by_provider(
-    provider_id: UUID,
-) -> List[WorkspaceWithModel]:
+async def get_workspace_by_name(
+    workspace_name: str,
+) -> v1_models.FullWorkspace:
     """List workspaces by provider ID."""
     try:
-        return await wscrud.workspaces_by_provider(provider_id)
+        ws = await wscrud.get_workspace_by_name(workspace_name)
+        muxes = await wscrud.get_muxes(workspace_name)
+
+        return v1_models.FullWorkspace(
+            name=ws.name,
+            config=v1_models.WorkspaceConfig(
+                custom_instructions=ws.custom_instructions or "",
+                muxing_rules=muxes,
+            ),
+        )
 
+    except crud.WorkspaceDoesNotExistError:
+        raise HTTPException(status_code=404, detail="Workspace does not exist")
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 

src/codegate/db/connection.py

@@ -35,7 +35,6 @@
     ProviderModel,
     Session,
     WorkspaceRow,
-    WorkspaceWithModel,
     WorkspaceWithSessionInfo,
 )
 from codegate.db.token_usage import TokenUsageParser
@@ -820,11 +819,13 @@ 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]:
+    async def get_workspaces_by_provider(self, provider_id: str) -> List[WorkspaceRow]:
         sql = text(
             """
             SELECT
-                w.id, w.name, m.provider_model_name
+                w.id,
+                w.name,
+                w.custom_instructions
             FROM workspaces w
             JOIN muxes m ON w.id = m.workspace_id
             WHERE m.provider_endpoint_id = :provider_id
@@ -833,7 +834,7 @@ async def get_workspaces_by_provider(self, provider_id: str) -> List[WorkspaceWi
         )
         conditions = {"provider_id": provider_id}
         workspaces = await self._exec_select_conditions_to_pydantic(
-            WorkspaceWithModel, sql, conditions, should_raise=True
+            WorkspaceRow, sql, conditions, should_raise=True
         )
         return workspaces
 

src/codegate/workspaces/crud.py

@@ -281,7 +281,9 @@ 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]:
+    async def workspaces_by_provider(
+        self, provider_id: uuid
+    ) -> List[db_models.WorkspaceWithSessionInfo]:
         """Get the workspaces by provider."""
 
         workspaces = await self._db_reader.get_workspaces_by_provider(str(provider_id))

tests/api/test_v1_workspaces.py

@@ -70,6 +70,184 @@ def mock_pipeline_factory():
     return mock_factory
 
 
+@pytest.mark.asyncio
+async def test_get_workspaces(
+    mock_pipeline_factory, mock_workspace_crud, mock_provider_crud
+) -> None:
+    with (
+        patch("codegate.api.v1.wscrud", mock_workspace_crud),
+        patch("codegate.api.v1.pcrud", mock_provider_crud),
+        patch(
+            "codegate.providers.openai.provider.OpenAIProvider.models",
+            return_value=["foo-bar-001", "foo-bar-002"],
+        ),
+    ):
+        """Test getting all workspaces."""
+        app = init_app(mock_pipeline_factory)
+
+        async with AsyncClient(
+            transport=httpx.ASGITransport(app=app), base_url="http://test"
+        ) as ac:
+            # Create a provider for muxing rules
+            provider_payload = {
+                "name": "test-provider",
+                "description": "",
+                "auth_type": "none",
+                "provider_type": "openai",
+                "endpoint": "https://api.openai.com",
+                "api_key": "sk-proj-foo-bar-123-xzy",
+            }
+            response = await ac.post("/api/v1/provider-endpoints", json=provider_payload)
+            assert response.status_code == 201
+            provider = response.json()
+
+            # Create first workspace
+            name_1 = str(uuid())
+            workspace_1 = {
+                "name": name_1,
+                "config": {
+                    "custom_instructions": "Respond in haiku format",
+                    "muxing_rules": [
+                        {
+                            "provider_id": provider["id"],
+                            "model": "foo-bar-001",
+                            "matcher": "*.py",
+                            "matcher_type": "filename_match",
+                        }
+                    ],
+                },
+            }
+            response = await ac.post("/api/v1/workspaces", json=workspace_1)
+            assert response.status_code == 201
+
+            # Create second workspace
+            name_2 = str(uuid())
+            workspace_2 = {
+                "name": name_2,
+                "config": {
+                    "custom_instructions": "Respond in prose",
+                    "muxing_rules": [
+                        {
+                            "provider_id": provider["id"],
+                            "model": "foo-bar-002",
+                            "matcher": "*.js",
+                            "matcher_type": "filename_match",
+                        }
+                    ],
+                },
+            }
+            response = await ac.post("/api/v1/workspaces", json=workspace_2)
+            assert response.status_code == 201
+
+            response = await ac.get("/api/v1/workspaces")
+            assert response.status_code == 200
+            workspaces = response.json()["workspaces"]
+
+            # Verify response structure
+            assert isinstance(workspaces, list)
+            assert len(workspaces) >= 2
+
+            workspace_names = [w["name"] for w in workspaces]
+            assert name_1 in workspace_names
+            assert name_2 in workspace_names
+            assert len([n for n in workspace_names if n in [name_1, name_2]]) == 2
+
+
+@pytest.mark.asyncio
+async def test_get_workspaces_filter_by_provider(
+    mock_pipeline_factory, mock_workspace_crud, mock_provider_crud
+) -> None:
+    with (
+        patch("codegate.api.v1.wscrud", mock_workspace_crud),
+        patch("codegate.api.v1.pcrud", mock_provider_crud),
+        patch(
+            "codegate.providers.openai.provider.OpenAIProvider.models",
+            return_value=["foo-bar-001", "foo-bar-002"],
+        ),
+    ):
+        """Test filtering workspaces by provider ID."""
+        app = init_app(mock_pipeline_factory)
+
+        async with AsyncClient(
+            transport=httpx.ASGITransport(app=app), base_url="http://test"
+        ) as ac:
+            # Create first provider
+            provider_payload_1 = {
+                "name": "provider-1",
+                "description": "",
+                "auth_type": "none",
+                "provider_type": "openai",
+                "endpoint": "https://api.openai.com",
+                "api_key": "sk-proj-foo-bar-123-xyz",
+            }
+            response = await ac.post("/api/v1/provider-endpoints", json=provider_payload_1)
+            assert response.status_code == 201
+            provider_1 = response.json()
+
+            # Create second provider
+            provider_payload_2 = {
+                "name": "provider-2",
+                "description": "",
+                "auth_type": "none",
+                "provider_type": "openai",
+                "endpoint": "https://api.openai.com",
+                "api_key": "sk-proj-foo-bar-456-xyz",
+            }
+            response = await ac.post("/api/v1/provider-endpoints", json=provider_payload_2)
+            assert response.status_code == 201
+            provider_2 = response.json()
+
+            # Create workspace using provider 1
+            workspace_1 = {
+                "name": str(uuid()),
+                "config": {
+                    "custom_instructions": "Instructions 1",
+                    "muxing_rules": [
+                        {
+                            "provider_id": provider_1["id"],
+                            "model": "foo-bar-001",
+                            "matcher": "*.py",
+                            "matcher_type": "filename_match",
+                        }
+                    ],
+                },
+            }
+            response = await ac.post("/api/v1/workspaces", json=workspace_1)
+            assert response.status_code == 201
+
+            # Create workspace using provider 2
+            workspace_2 = {
+                "name": str(uuid()),
+                "config": {
+                    "custom_instructions": "Instructions 2",
+                    "muxing_rules": [
+                        {
+                            "provider_id": provider_2["id"],
+                            "model": "foo-bar-002",
+                            "matcher": "*.js",
+                            "matcher_type": "filename_match",
+                        }
+                    ],
+                },
+            }
+            response = await ac.post("/api/v1/workspaces", json=workspace_2)
+            assert response.status_code == 201
+
+            # Test filtering by provider 1
+            response = await ac.get(f"/api/v1/workspaces?provider_id={provider_1['id']}")
+            assert response.status_code == 200
+            workspaces = response.json()["workspaces"]
+            assert len(workspaces) == 1
+            assert workspaces[0]["name"] == workspace_1["name"]
+
+            # Test filtering by provider 2
+            response = await ac.get(f"/api/v1/workspaces?provider_id={provider_2['id']}")
+            assert response.status_code == 200
+            workspaces = response.json()["workspaces"]
+            assert len(workspaces) == 1
+            assert workspaces[0]["name"] == workspace_2["name"]
+
+
 @pytest.mark.asyncio
 async def test_create_update_workspace_happy_path(
     mock_pipeline_factory, mock_workspace_crud, mock_provider_crud
@@ -146,6 +324,10 @@ async def test_create_update_workspace_happy_path(
 
             response = await ac.post("/api/v1/workspaces", json=payload_create)
             assert response.status_code == 201
+
+            # Verify created workspace
+            response = await ac.get(f"/api/v1/workspaces/{name_1}")
+            assert response.status_code == 200
             response_body = response.json()
 
             assert response_body["name"] == name_1
@@ -184,6 +366,10 @@ async def test_create_update_workspace_happy_path(
 
             response = await ac.put(f"/api/v1/workspaces/{name_1}", json=payload_update)
             assert response.status_code == 201
+
+            # Verify updated workspace
+            response = await ac.get(f"/api/v1/workspaces/{name_2}")
+            assert response.status_code == 200
             response_body = response.json()
 
             assert response_body["name"] == name_2
@@ -222,8 +408,15 @@ async def test_create_update_workspace_name_only(
             response = await ac.post("/api/v1/workspaces", json=payload_create)
             assert response.status_code == 201
             response_body = response.json()
+            assert response_body["name"] == name_1
 
+            # Verify created workspace
+            response = await ac.get(f"/api/v1/workspaces/{name_1}")
+            assert response.status_code == 200
+            response_body = response.json()
             assert response_body["name"] == name_1
+            assert response_body["config"]["custom_instructions"] == ""
+            assert response_body["config"]["muxing_rules"] == []
 
             name_2: str = str(uuid())
 
@@ -234,8 +427,15 @@ async def test_create_update_workspace_name_only(
             response = await ac.put(f"/api/v1/workspaces/{name_1}", json=payload_update)
             assert response.status_code == 201
             response_body = response.json()
+            assert response_body["name"] == name_2
 
+            # Verify updated workspace
+            response = await ac.get(f"/api/v1/workspaces/{name_2}")
+            assert response.status_code == 200
+            response_body = response.json()
             assert response_body["name"] == name_2
+            assert response_body["config"]["custom_instructions"] == ""
+            assert response_body["config"]["muxing_rules"] == []
 
 
 @pytest.mark.asyncio