"""HTTP API for workspaces, navigation, bootstrap, and unified search. Routes (mounted under ``/api/v1/workspaces/`` and ``/api/v1/search/``): * ``GET /workspaces/`` — list user's workspaces * ``POST /workspaces/`` — create (perm: workspaces.workspace.manage) * ``GET /workspaces//`` — detail * ``GET /workspaces//bootstrap/`` — full shell payload * ``GET /workspaces//navigation/`` — nav tree * ``POST /workspaces//members/`` — add member (perm: ...member.manage) * ``DELETE /workspaces//members//`` — remove member * ``POST /workspaces//install-nav/`` — body: {"module": "..."}; install contribs * ``GET /search/`` — q=, entity= (repeatable) """ from __future__ import annotations from typing import Any from rest_framework import status from rest_framework.exceptions import NotFound, PermissionDenied, ValidationError from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from rest_framework.views import APIView from simorgh.apps.accounts.models import User from simorgh.apps.iam.models import Role from simorgh.apps.iam.permissions import HasPermission from simorgh.apps.workspaces.models import Workspace, WorkspaceAccessGrant, WorkspaceMembership, WorkspaceTemplate from simorgh.apps.workspaces.services import ( WorkspaceError, add_access_grant, add_member, create_workspace, create_workspace_from_template, get_user_org_context, get_user_workspace_context, install_navigation_from_contributions, list_access_grants, list_workspace_templates, remove_access_grant, remove_member, resolve_navigation, switch_active_node, switch_active_workspace, workspace_bootstrap, workspaces_for_user, ) from simorgh.core import search as core_search from simorgh.core.context import current_request_context from simorgh.core.view_models import serialize_workspace # --------------------------------------------------------------------------- # helpers # --------------------------------------------------------------------------- def _require_tenant(): ctx = current_request_context() if ctx.tenant is None: raise ValidationError({"detail": "tenant required (set X-Tenant header)"}) return ctx def _get_workspace(slug: str, tenant) -> Workspace: try: return Workspace.objects.get(tenant=tenant, slug=slug) except Workspace.DoesNotExist as exc: raise NotFound("workspace not found") from exc def _ensure_membership(ws: Workspace, user) -> None: if user.is_superuser: return if not WorkspaceMembership.objects.filter(workspace=ws, user=user).exists(): raise PermissionDenied("not a member of this workspace") # --------------------------------------------------------------------------- # Workspace list / create # --------------------------------------------------------------------------- class WorkspaceListView(APIView): permission_classes = (IsAuthenticated,) def get(self, request: Request) -> Response: ctx = _require_tenant() workspaces = workspaces_for_user(request.user, ctx.tenant) return Response({"results": [serialize_workspace(w) for w in workspaces]}) def post(self, request: Request) -> Response: ctx = _require_tenant() if not ( request.user.is_superuser or "workspaces.workspace.manage" in ctx.permissions ): raise PermissionDenied("workspaces.workspace.manage required") data = request.data or {} slug = (data.get("slug") or "").strip() name = (data.get("name") or "").strip() if not slug or not name: raise ValidationError({"detail": "slug and name are required"}) try: ws = create_workspace( ctx.tenant, slug=slug, name=name, kind=data.get("kind", "internal"), icon=data.get("icon", ""), theme=data.get("theme") or {}, default_module=data.get("default_module", ""), ) except WorkspaceError as exc: raise ValidationError({"detail": str(exc)}) from exc return Response(serialize_workspace(ws), status=status.HTTP_201_CREATED) class WorkspaceDetailView(APIView): permission_classes = (IsAuthenticated,) def get(self, request: Request, slug: str) -> Response: ctx = _require_tenant() ws = _get_workspace(slug, ctx.tenant) _ensure_membership(ws, request.user) return Response(serialize_workspace(ws)) class WorkspaceBootstrapView(APIView): permission_classes = (IsAuthenticated,) def get(self, request: Request, slug: str) -> Response: ctx = _require_tenant() ws = _get_workspace(slug, ctx.tenant) _ensure_membership(ws, request.user) return Response(workspace_bootstrap(ws, ctx)) class WorkspaceNavigationView(APIView): permission_classes = (IsAuthenticated,) def get(self, request: Request, slug: str) -> Response: ctx = _require_tenant() ws = _get_workspace(slug, ctx.tenant) _ensure_membership(ws, request.user) return Response({"items": resolve_navigation(ws, ctx)}) # --------------------------------------------------------------------------- # Membership # --------------------------------------------------------------------------- class WorkspaceMembersView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "workspaces.member.manage" def post(self, request: Request, slug: str) -> Response: ctx = _require_tenant() ws = _get_workspace(slug, ctx.tenant) data = request.data or {} user_id = data.get("user_id") if user_id is None: raise ValidationError({"detail": "user_id is required"}) try: user = User.objects.get(pk=user_id) except User.DoesNotExist as exc: raise ValidationError({"detail": "unknown user"}) from exc add_member(ws, user) return Response({"detail": "ok"}, status=status.HTTP_201_CREATED) def delete(self, request: Request, slug: str) -> Response: ctx = _require_tenant() ws = _get_workspace(slug, ctx.tenant) user_id = request.query_params.get("user_id") or (request.data or {}).get("user_id") if user_id is None: raise ValidationError({"detail": "user_id is required"}) try: user = User.objects.get(pk=user_id) except User.DoesNotExist as exc: raise ValidationError({"detail": "unknown user"}) from exc remove_member(ws, user) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Access grants — workspace-centric role → workspace access rules # --------------------------------------------------------------------------- class WorkspaceAccessGrantsView(APIView): """``GET|POST /workspaces//access-grants/`` GET — list all access grants for the workspace. POST — add an access grant (perm: workspaces.access_grant.manage). Body: ``{"role_id": ""}`` DELETE — remove an access grant (perm: workspaces.access_grant.manage). Body: ``{"role_id": ""}`` """ permission_classes = (IsAuthenticated,) @staticmethod def _serialize_grant(grant: WorkspaceAccessGrant) -> dict: return { "id": str(grant.pk), "role_id": str(grant.role_id), "role_code": grant.role.code, "role_name": grant.role.name, } def get(self, request: Request, slug: str) -> Response: ctx = _require_tenant() ws = _get_workspace(slug, ctx.tenant) _ensure_membership(ws, request.user) grants = list_access_grants(ws) return Response({"results": [self._serialize_grant(g) for g in grants]}) def post(self, request: Request, slug: str) -> Response: ctx = _require_tenant() if not (request.user.is_superuser or "workspaces.access_grant.manage" in ctx.permissions): raise PermissionDenied("workspaces.access_grant.manage required") ws = _get_workspace(slug, ctx.tenant) role_id = (request.data or {}).get("role_id") if not role_id: raise ValidationError({"detail": "role_id is required"}) try: role = Role.objects.get(pk=role_id) except (Role.DoesNotExist, ValueError, TypeError) as exc: raise ValidationError({"detail": "role not found"}) from exc grant = add_access_grant(ws, role) return Response(self._serialize_grant(grant), status=status.HTTP_201_CREATED) def delete(self, request: Request, slug: str) -> Response: ctx = _require_tenant() if not (request.user.is_superuser or "workspaces.access_grant.manage" in ctx.permissions): raise PermissionDenied("workspaces.access_grant.manage required") ws = _get_workspace(slug, ctx.tenant) role_id = (request.data or {}).get("role_id") if not role_id: raise ValidationError({"detail": "role_id is required"}) try: role = Role.objects.get(pk=role_id) except (Role.DoesNotExist, ValueError, TypeError) as exc: raise ValidationError({"detail": "role not found"}) from exc remove_access_grant(ws, role) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Navigation install # --------------------------------------------------------------------------- class WorkspaceInstallNavigationView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "workspaces.navigation.manage" def post(self, request: Request, slug: str) -> Response: ctx = _require_tenant() ws = _get_workspace(slug, ctx.tenant) module = (request.data or {}).get("module") if not module: raise ValidationError({"detail": "module is required"}) written = install_navigation_from_contributions(ws, module) return Response({"installed": written}) # --------------------------------------------------------------------------- # Unified search # --------------------------------------------------------------------------- class SearchView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "search.search.run" def get(self, request: Request) -> Response: _require_tenant() ctx = current_request_context() q = (request.query_params.get("q") or "").strip() entities = tuple(request.query_params.getlist("entity") or ()) limit = int(request.query_params.get("limit") or 10) hits = core_search.run_search(ctx, q, entities=entities, limit_per_entity=limit) payload: dict[str, Any] = { "query": q, "count": len(hits), "results": [h.to_dict() for h in hits], } return Response(payload) # --------------------------------------------------------------------------- # Workspace templates # --------------------------------------------------------------------------- def _serialize_template(tmpl: WorkspaceTemplate) -> dict[str, Any]: return { "slug": tmpl.slug, "name": tmpl.name, "description": tmpl.description, "kind": tmpl.kind, "icon": tmpl.icon, "theme": tmpl.theme, "default_module": tmpl.default_module, "navigation_schema": tmpl.navigation_schema, "is_system": tmpl.is_system, } class WorkspaceTemplateListView(APIView): """``GET /workspaces/templates/`` — list all workspace templates.""" permission_classes = (IsAuthenticated,) def get(self, request: Request) -> Response: _require_tenant() templates = list_workspace_templates() return Response({"results": [_serialize_template(t) for t in templates]}) class WorkspaceFromTemplateView(APIView): """``POST /workspaces/from-template//`` — instantiate a template. Body (all optional): ``name`` — override the workspace name ``slug`` — override the workspace slug """ permission_classes = (IsAuthenticated,) def post(self, request: Request, slug: str) -> Response: ctx = _require_tenant() if not ( request.user.is_superuser or "workspaces.workspace.manage" in ctx.permissions ): raise PermissionDenied("workspaces.workspace.manage required") data = request.data or {} try: ws = create_workspace_from_template( ctx.tenant, slug, name=data.get("name") or None, slug=data.get("slug") or None, ) except WorkspaceTemplate.DoesNotExist as exc: raise NotFound("workspace template not found") from exc except WorkspaceError as exc: raise ValidationError({"detail": str(exc)}) from exc return Response(serialize_workspace(ws), status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # Operational context # --------------------------------------------------------------------------- def _serialize_context(org_pref: Any, ws_pref: Any, tenant: Any) -> dict[str, Any]: from simorgh.apps.organizations.models import OrganizationNode # noqa: PLC0415 from simorgh.core.view_models import serialize_workspace # noqa: PLC0415 node = org_pref.active_organization_node if org_pref is not None else None node_data = None if node is not None: node_data = { "id": node.pk, "name": node.name, "type": node.type, "path": node.path_string, } ws = ws_pref.active_workspace if ws_pref is not None else None ws_data = serialize_workspace(ws) if ws is not None else None # available nodes = all active nodes the user can see in this tenant available = list( OrganizationNode.objects.filter(tenant=tenant, is_active=True) .values("id", "name", "type", "path_string", "depth") .order_by("path_string") ) return { "active_organization_node": node_data, "active_workspace": ws_data, "available_nodes": available, } class UserContextView(APIView): """Operational context API — independent org context and workspace. ``GET /workspaces/context/`` — read current context preferences ``POST /workspaces/context/`` — update active org node and/or workspace POST body accepts both fields independently: ``{"node_id": 5, "workspace_slug": "support-center"}`` Omitting a field leaves it unchanged. """ permission_classes = (IsAuthenticated,) def get(self, request: Request) -> Response: ctx = _require_tenant() org_pref = get_user_org_context(request.user, ctx.tenant) ws_pref = get_user_workspace_context(request.user, ctx.tenant) return Response(_serialize_context(org_pref, ws_pref, ctx.tenant)) def post(self, request: Request) -> Response: ctx = _require_tenant() data = request.data or {} if "node_id" not in data and "workspace_slug" not in data: raise ValidationError( {"detail": "at least one of node_id or workspace_slug is required"} ) if "node_id" in data: try: node_id = int(data["node_id"]) if data["node_id"] is not None else None except (TypeError, ValueError) as exc: raise ValidationError({"detail": "node_id must be an integer or null"}) from exc try: switch_active_node(request.user, ctx.tenant, node_id=node_id) except Exception as exc: # noqa: BLE001 raise ValidationError({"detail": str(exc)}) from exc if "workspace_slug" in data: try: switch_active_workspace( request.user, ctx.tenant, workspace_slug=data["workspace_slug"] or None, ) except Exception as exc: # noqa: BLE001 raise ValidationError({"detail": str(exc)}) from exc org_pref = get_user_org_context(request.user, ctx.tenant) ws_pref = get_user_workspace_context(request.user, ctx.tenant) return Response(_serialize_context(org_pref, ws_pref, ctx.tenant))