Agent Zero
310a9f0542
Backend:
- Widget CRUD: get_widgets, create_widget, update_widget, delete_widget
- Manager role check: is_workspace_manager
- Cross-tenant validation: verify_user_same_tenant (UserTenant)
- Default workspace seeding: seed_default_workspace with 12 standard modules
- Set user default workspace: set_user_default_workspace
- Fix create_workspace default uniqueness (unset others before insert)
- Widget CRUD routes: GET/POST/PUT/DELETE /{workspace_id}/widgets
- Set-default route: POST /{workspace_id}/set-default
- Cross-tenant validation in assign_user route
Frontend:
- workspaceStore (Zustand): central state with sessionStorage persistence
- API client interceptor: X-Workspace-ID header on all requests
- useWorkspace hook refactored to use workspaceStore
- Widget API hooks: useWorkspaceWidgets, useCreateWorkspaceWidget, etc.
- useSetDefaultWorkspace hook
- Settings route: /settings/workspaces with WorkspaceManagerPage
- Settings nav item for Workspaces
Tests:
- 25 backend tests (CRUD, modules, widgets, users, manager, seeding, context, isolation)
- 12 frontend tests (workspaceStore state, visibility, persistence, reset)
- 48/48 backend tests passing
- 12/12 frontend tests passing
665 lines
20 KiB
Python
665 lines
20 KiB
Python
"""Workspace service — CRUD, module config, user assignment, widgets.
|
|
|
|
Workspaces are UI/navigation context only. They never affect permissions.
|
|
See: docs/security_kernel.md
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select, update, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.workspace import Workspace, WorkspaceModule, WorkspaceUser, WorkspaceWidget
|
|
from app.models.user import User, UserTenant
|
|
|
|
|
|
def _workspace_to_dict(ws: Workspace, modules: list[WorkspaceModule] | None = None, user_count: int = 0) -> dict[str, Any]:
|
|
return {
|
|
"id": str(ws.id),
|
|
"name": ws.name,
|
|
"icon": ws.icon,
|
|
"description": ws.description,
|
|
"is_default": ws.is_default,
|
|
"is_active": ws.is_active,
|
|
"created_by": str(ws.created_by) if ws.created_by else None,
|
|
"created_at": ws.created_at.isoformat() if ws.created_at else None,
|
|
"updated_at": ws.updated_at.isoformat() if ws.updated_at else None,
|
|
"modules": [
|
|
{
|
|
"id": str(m.id),
|
|
"module_key": m.module_key,
|
|
"is_visible": m.is_visible,
|
|
"menu_order": m.menu_order,
|
|
"config": m.config or {},
|
|
}
|
|
for m in (modules or [])
|
|
],
|
|
"user_count": user_count,
|
|
}
|
|
|
|
|
|
async def list_workspaces(
|
|
db: AsyncSession, tenant_id: uuid.UUID
|
|
) -> dict[str, Any]:
|
|
"""List all workspaces for a tenant."""
|
|
q = select(Workspace).where(
|
|
Workspace.tenant_id == tenant_id,
|
|
Workspace.is_active == True, # noqa: E712
|
|
).order_by(Workspace.name)
|
|
result = await db.execute(q)
|
|
workspaces = result.scalars().all()
|
|
|
|
items = []
|
|
for ws in workspaces:
|
|
# Count users
|
|
count_q = select(func.count()).select_from(WorkspaceUser).where(
|
|
WorkspaceUser.workspace_id == ws.id,
|
|
WorkspaceUser.tenant_id == tenant_id,
|
|
)
|
|
count_result = await db.execute(count_q)
|
|
user_count = count_result.scalar() or 0
|
|
items.append(_workspace_to_dict(ws, user_count=user_count))
|
|
|
|
return {"items": items, "total": len(items)}
|
|
|
|
|
|
async def get_workspace(
|
|
db: AsyncSession, tenant_id: uuid.UUID, workspace_id: uuid.UUID
|
|
) -> dict[str, Any] | None:
|
|
"""Get a single workspace with modules and user count."""
|
|
q = select(Workspace).where(
|
|
Workspace.id == workspace_id,
|
|
Workspace.tenant_id == tenant_id,
|
|
)
|
|
result = await db.execute(q)
|
|
ws = result.scalar_one_or_none()
|
|
if ws is None:
|
|
return None
|
|
|
|
# Get modules
|
|
mod_q = select(WorkspaceModule).where(
|
|
WorkspaceModule.workspace_id == workspace_id,
|
|
WorkspaceModule.tenant_id == tenant_id,
|
|
).order_by(WorkspaceModule.menu_order)
|
|
mod_result = await db.execute(mod_q)
|
|
modules = mod_result.scalars().all()
|
|
|
|
# Count users
|
|
count_q = select(func.count()).select_from(WorkspaceUser).where(
|
|
WorkspaceUser.workspace_id == workspace_id,
|
|
WorkspaceUser.tenant_id == tenant_id,
|
|
)
|
|
count_result = await db.execute(count_q)
|
|
user_count = count_result.scalar() or 0
|
|
|
|
return _workspace_to_dict(ws, modules=modules, user_count=user_count)
|
|
|
|
|
|
async def create_workspace(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
name: str,
|
|
icon: str = "LayoutGrid",
|
|
description: str | None = None,
|
|
is_default: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create a new workspace."""
|
|
# If this is the default workspace, unset others FIRST (avoids unique constraint violation)
|
|
if is_default:
|
|
await db.execute(
|
|
update(Workspace)
|
|
.where(
|
|
Workspace.tenant_id == tenant_id,
|
|
)
|
|
.values(is_default=False)
|
|
)
|
|
await db.flush()
|
|
|
|
ws = Workspace(
|
|
tenant_id=tenant_id,
|
|
name=name,
|
|
icon=icon,
|
|
description=description,
|
|
is_default=is_default,
|
|
is_active=True,
|
|
created_by=user_id,
|
|
)
|
|
db.add(ws)
|
|
await db.flush()
|
|
await db.refresh(ws)
|
|
|
|
# Auto-assign creator as manager
|
|
wu = WorkspaceUser(
|
|
tenant_id=tenant_id,
|
|
workspace_id=ws.id,
|
|
user_id=user_id,
|
|
role="manager",
|
|
is_default=is_default,
|
|
assigned_by=user_id,
|
|
)
|
|
db.add(wu)
|
|
await db.flush()
|
|
|
|
return _workspace_to_dict(ws, user_count=1)
|
|
|
|
|
|
async def update_workspace(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
workspace_id: uuid.UUID,
|
|
name: str | None = None,
|
|
icon: str | None = None,
|
|
description: str | None = None,
|
|
is_default: bool | None = None,
|
|
is_active: bool | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""Update a workspace."""
|
|
q = select(Workspace).where(
|
|
Workspace.id == workspace_id,
|
|
Workspace.tenant_id == tenant_id,
|
|
)
|
|
result = await db.execute(q)
|
|
ws = result.scalar_one_or_none()
|
|
if ws is None:
|
|
return None
|
|
|
|
if name is not None:
|
|
ws.name = name
|
|
if icon is not None:
|
|
ws.icon = icon
|
|
if description is not None:
|
|
ws.description = description
|
|
if is_active is not None:
|
|
ws.is_active = is_active
|
|
if is_default is True:
|
|
# Unset other defaults
|
|
await db.execute(
|
|
update(Workspace)
|
|
.where(
|
|
Workspace.tenant_id == tenant_id,
|
|
Workspace.id != ws.id,
|
|
)
|
|
.values(is_default=False)
|
|
)
|
|
ws.is_default = True
|
|
elif is_default is False:
|
|
ws.is_default = False
|
|
|
|
await db.flush()
|
|
await db.refresh(ws)
|
|
return _workspace_to_dict(ws)
|
|
|
|
|
|
async def delete_workspace(
|
|
db: AsyncSession, tenant_id: uuid.UUID, workspace_id: uuid.UUID
|
|
) -> bool:
|
|
"""Delete a workspace (soft delete by setting is_active=False)."""
|
|
q = select(Workspace).where(
|
|
Workspace.id == workspace_id,
|
|
Workspace.tenant_id == tenant_id,
|
|
)
|
|
result = await db.execute(q)
|
|
ws = result.scalar_one_or_none()
|
|
if ws is None:
|
|
return False
|
|
ws.is_active = False
|
|
await db.flush()
|
|
return True
|
|
|
|
|
|
async def set_workspace_modules(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
workspace_id: uuid.UUID,
|
|
modules: list[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
"""Set the modules for a workspace. Replaces all existing modules."""
|
|
# Delete existing modules
|
|
existing_q = select(WorkspaceModule).where(
|
|
WorkspaceModule.workspace_id == workspace_id,
|
|
WorkspaceModule.tenant_id == tenant_id,
|
|
)
|
|
existing = await db.execute(existing_q)
|
|
for m in existing.scalars().all():
|
|
await db.delete(m)
|
|
|
|
# Insert new modules
|
|
result = []
|
|
for mod in modules:
|
|
wm = WorkspaceModule(
|
|
tenant_id=tenant_id,
|
|
workspace_id=workspace_id,
|
|
module_key=mod["module_key"],
|
|
is_visible=mod.get("is_visible", True),
|
|
menu_order=mod.get("menu_order", 0),
|
|
config=mod.get("config", {}),
|
|
)
|
|
db.add(wm)
|
|
await db.flush()
|
|
await db.refresh(wm)
|
|
result.append({
|
|
"id": str(wm.id),
|
|
"module_key": wm.module_key,
|
|
"is_visible": wm.is_visible,
|
|
"menu_order": wm.menu_order,
|
|
"config": wm.config or {},
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
async def assign_user(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
workspace_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
role: str = "member",
|
|
assigned_by: uuid.UUID | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Assign a user to a workspace."""
|
|
wu = WorkspaceUser(
|
|
tenant_id=tenant_id,
|
|
workspace_id=workspace_id,
|
|
user_id=user_id,
|
|
role=role,
|
|
is_default=False,
|
|
assigned_by=assigned_by,
|
|
)
|
|
db.add(wu)
|
|
await db.flush()
|
|
await db.refresh(wu)
|
|
return {
|
|
"id": str(wu.id),
|
|
"workspace_id": str(wu.workspace_id),
|
|
"user_id": str(wu.user_id),
|
|
"role": wu.role,
|
|
"is_default": wu.is_default,
|
|
"assigned_at": wu.assigned_at.isoformat() if wu.assigned_at else None,
|
|
}
|
|
|
|
|
|
async def remove_user(
|
|
db: AsyncSession, tenant_id: uuid.UUID, workspace_id: uuid.UUID, user_id: uuid.UUID
|
|
) -> bool:
|
|
"""Remove a user from a workspace."""
|
|
q = select(WorkspaceUser).where(
|
|
WorkspaceUser.workspace_id == workspace_id,
|
|
WorkspaceUser.user_id == user_id,
|
|
WorkspaceUser.tenant_id == tenant_id,
|
|
)
|
|
result = await db.execute(q)
|
|
wu = result.scalar_one_or_none()
|
|
if wu is None:
|
|
return False
|
|
await db.delete(wu)
|
|
await db.flush()
|
|
return True
|
|
|
|
|
|
async def get_my_workspaces(
|
|
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID
|
|
) -> dict[str, Any]:
|
|
"""Get workspaces assigned to the current user."""
|
|
q = (
|
|
select(Workspace, WorkspaceUser)
|
|
.join(WorkspaceUser, WorkspaceUser.workspace_id == Workspace.id)
|
|
.where(
|
|
WorkspaceUser.user_id == user_id,
|
|
Workspace.tenant_id == tenant_id,
|
|
Workspace.is_active == True, # noqa: E712
|
|
)
|
|
.order_by(Workspace.name)
|
|
)
|
|
result = await db.execute(q)
|
|
rows = result.all()
|
|
|
|
items = []
|
|
for ws, wu in rows:
|
|
# Get modules for this workspace
|
|
mod_q = select(WorkspaceModule).where(
|
|
WorkspaceModule.workspace_id == ws.id,
|
|
WorkspaceModule.tenant_id == tenant_id,
|
|
WorkspaceModule.is_visible == True, # noqa: E712
|
|
).order_by(WorkspaceModule.menu_order)
|
|
mod_result = await db.execute(mod_q)
|
|
modules = mod_result.scalars().all()
|
|
|
|
items.append({
|
|
"id": str(ws.id),
|
|
"name": ws.name,
|
|
"icon": ws.icon,
|
|
"description": ws.description,
|
|
"is_default": ws.is_default,
|
|
"role": wu.role,
|
|
"is_user_default": wu.is_default,
|
|
"modules": [
|
|
{
|
|
"module_key": m.module_key,
|
|
"menu_order": m.menu_order,
|
|
"config": m.config or {},
|
|
}
|
|
for m in modules
|
|
],
|
|
})
|
|
|
|
return {"items": items, "total": len(items)}
|
|
|
|
|
|
async def get_workspace_context(
|
|
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, workspace_id: uuid.UUID
|
|
) -> dict[str, Any] | None:
|
|
"""Get workspace context for a user — modules, widgets, config.
|
|
|
|
Validates:
|
|
- Workspace belongs to tenant
|
|
- User is assigned or is system admin / tenant admin
|
|
- Workspace is active
|
|
"""
|
|
# Check workspace exists and is active
|
|
ws_q = select(Workspace).where(
|
|
Workspace.id == workspace_id,
|
|
Workspace.tenant_id == tenant_id,
|
|
Workspace.is_active == True, # noqa: E712
|
|
)
|
|
ws_result = await db.execute(ws_q)
|
|
ws = ws_result.scalar_one_or_none()
|
|
if ws is None:
|
|
return None
|
|
|
|
# Check user is assigned
|
|
wu_q = select(WorkspaceUser).where(
|
|
WorkspaceUser.workspace_id == workspace_id,
|
|
WorkspaceUser.user_id == user_id,
|
|
WorkspaceUser.tenant_id == tenant_id,
|
|
)
|
|
wu_result = await db.execute(wu_q)
|
|
wu = wu_result.scalar_one_or_none()
|
|
if wu is None:
|
|
return None # User not assigned — caller can check is_system_admin
|
|
|
|
# Get visible modules
|
|
mod_q = select(WorkspaceModule).where(
|
|
WorkspaceModule.workspace_id == workspace_id,
|
|
WorkspaceModule.tenant_id == tenant_id,
|
|
WorkspaceModule.is_visible == True, # noqa: E712
|
|
).order_by(WorkspaceModule.menu_order)
|
|
mod_result = await db.execute(mod_q)
|
|
modules = mod_result.scalars().all()
|
|
|
|
# Get widgets
|
|
widget_q = select(WorkspaceWidget).where(
|
|
WorkspaceWidget.workspace_id == workspace_id,
|
|
WorkspaceWidget.tenant_id == tenant_id,
|
|
).order_by(WorkspaceWidget.position_y, WorkspaceWidget.position_x)
|
|
widget_result = await db.execute(widget_q)
|
|
widgets = widget_result.scalars().all()
|
|
|
|
return {
|
|
"workspace_id": str(ws.id),
|
|
"name": ws.name,
|
|
"icon": ws.icon,
|
|
"role": wu.role,
|
|
"modules": [
|
|
{
|
|
"module_key": m.module_key,
|
|
"menu_order": m.menu_order,
|
|
"config": m.config or {},
|
|
}
|
|
for m in modules
|
|
],
|
|
"widgets": [
|
|
{
|
|
"id": str(w.id),
|
|
"widget_key": w.widget_key,
|
|
"position_x": w.position_x,
|
|
"position_y": w.position_y,
|
|
"width": w.width,
|
|
"height": w.height,
|
|
"config": w.config or {},
|
|
}
|
|
for w in widgets
|
|
],
|
|
}
|
|
|
|
# ─── Widget CRUD ─────────────────────────────────────────────
|
|
|
|
async def get_widgets(
|
|
db: AsyncSession, tenant_id: uuid.UUID, workspace_id: uuid.UUID
|
|
) -> list[dict[str, Any]]:
|
|
"""List all widgets for a workspace."""
|
|
q = select(WorkspaceWidget).where(
|
|
WorkspaceWidget.workspace_id == workspace_id,
|
|
WorkspaceWidget.tenant_id == tenant_id,
|
|
).order_by(WorkspaceWidget.position_y, WorkspaceWidget.position_x)
|
|
result = await db.execute(q)
|
|
widgets = result.scalars().all()
|
|
return [
|
|
{
|
|
"id": str(w.id),
|
|
"workspace_id": str(w.workspace_id),
|
|
"widget_key": w.widget_key,
|
|
"position_x": w.position_x,
|
|
"position_y": w.position_y,
|
|
"width": w.width,
|
|
"height": w.height,
|
|
"config": w.config or {},
|
|
}
|
|
for w in widgets
|
|
]
|
|
|
|
|
|
async def create_widget(
|
|
db: AsyncSession, tenant_id: uuid.UUID, workspace_id: uuid.UUID,
|
|
widget_key: str, position_x: int = 0, position_y: int = 0,
|
|
width: int = 1, height: int = 1, config: dict | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new widget in a workspace. Multiple instances of the same widget_key allowed."""
|
|
w = WorkspaceWidget(
|
|
tenant_id=tenant_id,
|
|
workspace_id=workspace_id,
|
|
widget_key=widget_key,
|
|
position_x=position_x,
|
|
position_y=position_y,
|
|
width=width,
|
|
height=height,
|
|
config=config or {},
|
|
)
|
|
db.add(w)
|
|
await db.flush()
|
|
await db.refresh(w)
|
|
return {
|
|
"id": str(w.id),
|
|
"workspace_id": str(w.workspace_id),
|
|
"widget_key": w.widget_key,
|
|
"position_x": w.position_x,
|
|
"position_y": w.position_y,
|
|
"width": w.width,
|
|
"height": w.height,
|
|
"config": w.config or {},
|
|
}
|
|
|
|
|
|
async def update_widget(
|
|
db: AsyncSession, tenant_id: uuid.UUID, widget_id: uuid.UUID,
|
|
position_x: int | None = None, position_y: int | None = None,
|
|
width: int | None = None, height: int | None = None,
|
|
config: dict | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""Update a widget."""
|
|
q = select(WorkspaceWidget).where(
|
|
WorkspaceWidget.id == widget_id,
|
|
WorkspaceWidget.tenant_id == tenant_id,
|
|
)
|
|
result = await db.execute(q)
|
|
w = result.scalar_one_or_none()
|
|
if w is None:
|
|
return None
|
|
if position_x is not None:
|
|
w.position_x = position_x
|
|
if position_y is not None:
|
|
w.position_y = position_y
|
|
if width is not None:
|
|
w.width = width
|
|
if height is not None:
|
|
w.height = height
|
|
if config is not None:
|
|
w.config = config
|
|
await db.flush()
|
|
await db.refresh(w)
|
|
return {
|
|
"id": str(w.id),
|
|
"workspace_id": str(w.workspace_id),
|
|
"widget_key": w.widget_key,
|
|
"position_x": w.position_x,
|
|
"position_y": w.position_y,
|
|
"width": w.width,
|
|
"height": w.height,
|
|
"config": w.config or {},
|
|
}
|
|
|
|
|
|
async def delete_widget(
|
|
db: AsyncSession, tenant_id: uuid.UUID, widget_id: uuid.UUID
|
|
) -> bool:
|
|
"""Delete a widget."""
|
|
q = select(WorkspaceWidget).where(
|
|
WorkspaceWidget.id == widget_id,
|
|
WorkspaceWidget.tenant_id == tenant_id,
|
|
)
|
|
result = await db.execute(q)
|
|
w = result.scalar_one_or_none()
|
|
if w is None:
|
|
return False
|
|
await db.delete(w)
|
|
await db.flush()
|
|
return True
|
|
|
|
|
|
# ─── Manager Role Check ───────────────────────────────────────
|
|
|
|
async def is_workspace_manager(
|
|
db: AsyncSession, tenant_id: uuid.UUID, workspace_id: uuid.UUID, user_id: uuid.UUID
|
|
) -> bool:
|
|
"""Check if a user is a manager of a workspace (or system admin)."""
|
|
q = select(WorkspaceUser).where(
|
|
WorkspaceUser.workspace_id == workspace_id,
|
|
WorkspaceUser.user_id == user_id,
|
|
WorkspaceUser.tenant_id == tenant_id,
|
|
WorkspaceUser.role == "manager",
|
|
)
|
|
result = await db.execute(q)
|
|
return result.scalar_one_or_none() is not None
|
|
|
|
|
|
# ─── Cross-Tenant Validation ──────────────────────────────────
|
|
|
|
async def verify_user_same_tenant(
|
|
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID
|
|
) -> bool:
|
|
"""Verify that a user belongs to the same tenant. Prevents cross-tenant assignment."""
|
|
q = select(UserTenant).where(
|
|
UserTenant.user_id == user_id,
|
|
UserTenant.tenant_id == tenant_id,
|
|
)
|
|
result = await db.execute(q)
|
|
return result.scalar_one_or_none() is not None
|
|
|
|
|
|
# ─── Default Workspace Seeding ────────────────────────────────
|
|
|
|
async def seed_default_workspace(
|
|
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID
|
|
) -> dict[str, Any] | None:
|
|
"""Create a default workspace for a tenant if none exists.
|
|
Called during tenant setup or user creation.
|
|
"""
|
|
# Check if any workspace exists for this tenant
|
|
existing_q = select(Workspace).where(
|
|
Workspace.tenant_id == tenant_id,
|
|
Workspace.is_active == True, # noqa: E712
|
|
)
|
|
result = await db.execute(existing_q)
|
|
if result.scalars().first() is not None:
|
|
return None # Already has workspaces
|
|
|
|
# Create default workspace with all standard modules visible
|
|
ws = Workspace(
|
|
tenant_id=tenant_id,
|
|
name="Standard",
|
|
icon="LayoutGrid",
|
|
description="Standard-Workspace mit allen Modulen",
|
|
is_default=True,
|
|
is_active=True,
|
|
created_by=user_id,
|
|
)
|
|
db.add(ws)
|
|
await db.flush()
|
|
await db.refresh(ws)
|
|
|
|
# Auto-assign creator as manager
|
|
wu = WorkspaceUser(
|
|
tenant_id=tenant_id,
|
|
workspace_id=ws.id,
|
|
user_id=user_id,
|
|
role="manager",
|
|
is_default=True,
|
|
assigned_by=user_id,
|
|
)
|
|
db.add(wu)
|
|
|
|
# Add all standard modules visible by default
|
|
standard_modules = [
|
|
("dashboard", 0), ("contacts", 10), ("calendar", 20),
|
|
("mail", 30), ("tasks", 40), ("dms", 50),
|
|
("kommunikation", 60), ("reports", 70), ("automation", 80),
|
|
("tags", 90), ("settings", 100), ("audit", 110),
|
|
]
|
|
for key, order in standard_modules:
|
|
wm = WorkspaceModule(
|
|
tenant_id=tenant_id,
|
|
workspace_id=ws.id,
|
|
module_key=key,
|
|
is_visible=True,
|
|
menu_order=order,
|
|
config={},
|
|
)
|
|
db.add(wm)
|
|
|
|
await db.flush()
|
|
return _workspace_to_dict(ws, user_count=1)
|
|
|
|
|
|
# ─── Set User Default Workspace ───────────────────────────────
|
|
|
|
async def set_user_default_workspace(
|
|
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, workspace_id: uuid.UUID
|
|
) -> bool:
|
|
"""Set a workspace as the user's default. Unsets previous default."""
|
|
# Unset previous default
|
|
await db.execute(
|
|
update(WorkspaceUser)
|
|
.where(
|
|
WorkspaceUser.user_id == user_id,
|
|
WorkspaceUser.tenant_id == tenant_id,
|
|
WorkspaceUser.workspace_id != workspace_id,
|
|
)
|
|
.values(is_default=False)
|
|
)
|
|
# Set new default
|
|
await db.execute(
|
|
update(WorkspaceUser)
|
|
.where(
|
|
WorkspaceUser.user_id == user_id,
|
|
WorkspaceUser.tenant_id == tenant_id,
|
|
WorkspaceUser.workspace_id == workspace_id,
|
|
)
|
|
.values(is_default=True)
|
|
)
|
|
await db.flush()
|
|
return True
|