feat(workspaces): implement workspace limits feature with backend integration and UI updates

This commit is contained in:
Anish Sarkar 2026-07-16 15:44:49 +05:30
parent 1385ff4789
commit 38b784fbac
12 changed files with 176 additions and 18 deletions

View file

@ -850,6 +850,9 @@ class Config:
# Auth
AUTH_TYPE = os.getenv("AUTH_TYPE", "LOCAL")
REGISTRATION_ENABLED = os.getenv("REGISTRATION_ENABLED", "TRUE").upper() == "TRUE"
# Max workspaces a user may own. The frontend reads this through the
# workspace limits route; do not duplicate this value client-side.
MAX_WORKSPACES_PER_USER = int(os.getenv("MAX_WORKSPACES_PER_USER", "100"))
# Google OAuth
GOOGLE_OAUTH_CLIENT_ID = os.getenv("GOOGLE_OAUTH_CLIENT_ID")

View file

@ -6,6 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from app.auth.context import AuthContext
from app.config import config
from app.db import (
Permission,
Workspace,
@ -81,6 +82,22 @@ async def create_workspace(
user = auth.user
try:
workspace_data = workspace.model_dump()
not_deleting = ~Workspace.name.startswith("[DELETING] ")
owned_count = (
await session.execute(
select(func.count())
.select_from(Workspace)
.filter(Workspace.user_id == user.id, not_deleting)
)
).scalar_one()
if owned_count >= config.MAX_WORKSPACES_PER_USER:
raise HTTPException(
status_code=409,
detail=(
"Workspace limit reached. You can own at most "
f"{config.MAX_WORKSPACES_PER_USER} workspaces."
),
)
# citations_enabled defaults to True (handled by Pydantic schema)
# qna_custom_instructions defaults to None/empty (handled by DB)
@ -207,6 +224,11 @@ async def read_workspaces(
) from e
@router.get("/workspaces/limits")
async def read_workspace_limits(_auth: AuthContext = Depends(allow_any_principal)):
return {"max_workspaces_per_user": config.MAX_WORKSPACES_PER_USER}
@router.get("/workspaces/{workspace_id}", response_model=WorkspaceRead)
async def read_workspace(
workspace_id: int,

View file

@ -0,0 +1,63 @@
from __future__ import annotations
from types import SimpleNamespace
import pytest
from fastapi import HTTPException
from app.routes import workspaces_routes
from app.schemas import WorkspaceCreate
pytestmark = pytest.mark.unit
class _CountResult:
def __init__(self, count: int):
self.count = count
def scalar_one(self) -> int:
return self.count
class _FakeSession:
def __init__(self, owned_count: int):
self.owned_count = owned_count
async def execute(self, _statement):
return _CountResult(self.owned_count)
@pytest.mark.asyncio
async def test_read_workspace_limits_uses_backend_config(monkeypatch):
monkeypatch.setattr(
workspaces_routes.config,
"MAX_WORKSPACES_PER_USER",
37,
raising=False,
)
result = await workspaces_routes.read_workspace_limits(_auth=SimpleNamespace())
assert result == {"max_workspaces_per_user": 37}
@pytest.mark.asyncio
async def test_create_workspace_rejects_when_owned_limit_reached(monkeypatch):
monkeypatch.setattr(
workspaces_routes.config,
"MAX_WORKSPACES_PER_USER",
2,
raising=False,
)
auth = SimpleNamespace(user=SimpleNamespace(id="user-1"))
session = _FakeSession(owned_count=2)
with pytest.raises(HTTPException) as exc_info:
await workspaces_routes.create_workspace(
WorkspaceCreate(name="Extra", description=""),
session=session,
auth=auth,
)
assert exc_info.value.status_code == 409
assert "at most 2 workspaces" in exc_info.value.detail