# retoor <retoor@molodetz.nl>
from typing import Annotated
from fastapi import APIRouter, Depends, Request
from fastapi.responses import JSONResponse
from devplacepy.database import db, get_table, get_users_by_uids
from devplacepy.dependencies import json_or_form
from devplacepy.models import (
EditorPrefsForm,
WorkspaceFlagForm,
WorkspaceQuotaForm,
WorkspaceSuspendForm,
)
from devplacepy.responses import action_result, json_error, respond
from devplacepy.routers.admin._shared import deny_senior, is_senior_admin
from devplacepy.schemas import AdminWorkspacesOut
from devplacepy.seo import base_seo_context
from devplacepy.services.audit import record as audit
from devplacepy.services.containers import store
from devplacepy.services.containers.workspace import (
editor,
flags,
provision,
quota,
tunnels,
)
from devplacepy.utils import create_notification, generate_uid, not_found, require_admin
router = APIRouter()
def _decorate(rows: list[dict]) -> list[dict]:
owner_uids = {row.get("workspace_owner_uid") for row in rows if row.get("workspace_owner_uid")}
owners = get_users_by_uids(list(owner_uids)) if owner_uids else {}
projects = {}
if "projects" in db.tables:
project_uids = {row.get("project_uid") for row in rows if row.get("project_uid")}
for uid in project_uids:
found = get_table("projects").find_one(uid=uid)
if found:
projects[uid] = found
decorated = []
for row in rows:
view = provision.view(row)
owner = owners.get(row.get("workspace_owner_uid", "")) or {}
project = projects.get(row.get("project_uid", "")) or {}
view["owner_username"] = owner.get("username", "")
view["project_title"] = project.get("title", "")
view["project_slug"] = project.get("slug", "") or project.get("uid", "")
decorated.append(view)
return decorated
def _all_workspaces() -> list[dict]:
return list(get_table("instances").find(is_workspace=1, deleted_at=None))
def _instance_or_404(uid: str) -> dict:
instance = store.get_instance(uid)
if not instance or not instance.get("is_workspace"):
raise not_found("Workspace not found")
return instance
def _audit(request: Request, admin: dict, event_key: str, instance: dict, **extra):
audit.record(
request,
event_key,
user=admin,
target_type="instance",
target_uid=instance["uid"],
target_label=instance.get("name"),
summary=f"{admin['username']} {event_key} workspace {instance.get('name')}",
links=[audit.instance(instance["uid"], instance.get("name"))],
**extra,
)
@router.get("/workspaces")
async def admin_workspaces(request: Request):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
context = {
"workspaces": _decorate(_all_workspaces()),
"flags": flags.list_flags(),
"admin_section": "workspaces",
"user": admin,
**base_seo_context(
request,
title="Workspaces - Admin",
description="Administer dev workspaces.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Workspaces", "url": "/admin/workspaces"},
],
),
}
return respond(request, "admin_workspaces.html", context, model=AdminWorkspacesOut)
@router.get("/workspaces/data")
async def admin_workspaces_data(request: Request):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
return JSONResponse(
{"workspaces": _decorate(_all_workspaces()), "flags": flags.list_flags()}
)
@router.post("/workspaces/{uid}/suspend")
async def admin_workspace_suspend(
request: Request,
uid: str,
data: Annotated[WorkspaceSuspendForm, Depends(json_or_form(WorkspaceSuspendForm))],
):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
instance = _instance_or_404(uid)
reason = (data.reason or "").strip()
if not reason:
return json_error(400, "a reason is required and is shown to the owner")
provision.suspend(instance, admin["uid"], reason)
_audit(request, admin, "container.workspace.suspend", instance, metadata={"reason": reason})
owner = instance.get("workspace_owner_uid", "")
if owner:
create_notification(
owner,
"workspace",
f"Workspace {instance.get('name', '')} was suspended: {reason}",
instance["uid"],
"/projects",
)
return action_result(request, "/admin/workspaces")
@router.post("/workspaces/{uid}/unsuspend")
async def admin_workspace_unsuspend(request: Request, uid: str):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
instance = _instance_or_404(uid)
provision.unsuspend(instance)
_audit(request, admin, "container.workspace.unsuspend", instance)
owner = instance.get("workspace_owner_uid", "")
if owner:
create_notification(
owner,
"workspace",
f"Workspace {instance.get('name', '')} is available again.",
instance["uid"],
"/projects",
)
return action_result(request, "/admin/workspaces")
@router.post("/workspaces/{uid}/stop")
async def admin_workspace_stop(request: Request, uid: str):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
instance = _instance_or_404(uid)
provision.stop(instance)
_audit(request, admin, "container.workspace.stop", instance)
return action_result(request, "/admin/workspaces")
@router.post("/workspaces/{uid}/start")
async def admin_workspace_start(request: Request, uid: str):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
instance = _instance_or_404(uid)
store.update_instance(instance["uid"], {"desired_state": "running"})
_audit(request, admin, "container.workspace.resume", instance)
return action_result(request, "/admin/workspaces")
@router.post("/workspaces/{uid}/delete")
async def admin_workspace_delete(request: Request, uid: str):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
instance = _instance_or_404(uid)
for row in tunnels.list_for_instance(instance["uid"]):
tunnels.soft_delete(row["uid"], admin["uid"])
store.delete_instance(instance["uid"], admin["uid"])
_audit(request, admin, "container.workspace.delete", instance)
return action_result(request, "/admin/workspaces")
@router.post("/workspaces/{uid}/flag")
async def admin_workspace_flag(
request: Request,
uid: str,
data: Annotated[WorkspaceFlagForm, Depends(json_or_form(WorkspaceFlagForm))],
):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
instance = _instance_or_404(uid)
row = flags.raise_flag(
instance, data.kind or flags.KIND_MANUAL, data.severity, data.detail
)
_audit(request, admin, "container.workspace.flag.raise", instance,
metadata={"kind": data.kind, "severity": data.severity})
owner = instance.get("workspace_owner_uid", "")
if owner:
create_notification(
owner,
"workspace",
f"Workspace {instance.get('name', '')} was flagged: {data.detail or data.kind}",
instance["uid"],
"/projects",
)
return action_result(request, "/admin/workspaces", data=row)
@router.post("/workspaces/flags/{flag_uid}/resolve")
async def admin_flag_resolve(request: Request, flag_uid: str, status: str = "resolved"):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
if not flags.set_status(flag_uid, status, admin["uid"]):
raise not_found("Flag not found")
event = (
"container.workspace.flag.dismiss"
if status == "dismissed"
else "container.workspace.flag.resolve"
)
audit.record(
request,
event,
user=admin,
target_type="workspace_flag",
target_uid=flag_uid,
summary=f"{admin['username']} set flag {flag_uid} to {status}",
)
return action_result(request, "/admin/workspaces")
@router.post("/workspaces/{uid}/editor")
async def admin_workspace_editor(
request: Request,
uid: str,
data: Annotated[EditorPrefsForm, Depends(json_or_form(EditorPrefsForm))],
):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
instance = _instance_or_404(uid)
owner_uid = instance.get("workspace_owner_uid", "")
owner = get_users_by_uids([owner_uid]).get(owner_uid) if owner_uid else None
if is_senior_admin(admin, owner):
return deny_senior(
request,
admin,
owner_uid,
owner,
"container.workspace.editor.update",
"/admin/workspaces",
)
if not owner_uid:
return json_error(400, "this workspace has no owner")
if data.reset:
editor.reset_prefs(owner_uid, admin["uid"])
else:
editor.save_prefs(
owner_uid, data.model_dump(exclude={"reset"}, exclude_unset=True)
)
_audit(request, admin, "container.workspace.editor.update", instance)
return action_result(
request,
"/admin/workspaces",
data={"editor": editor.view(owner_uid, instance)},
)
@router.post("/workspaces/quota")
async def admin_workspace_quota(
request: Request,
data: Annotated[WorkspaceQuotaForm, Depends(json_or_form(WorkspaceQuotaForm))],
):
admin = require_admin(request)
if not isinstance(admin, dict):
return admin
if not data.owner_id:
return json_error(400, "owner_id is required")
table = get_table(quota.RULES_TABLE)
existing = table.find_one(
owner_kind="user", owner_id=data.owner_id, deleted_at=None
)
payload = {key: getattr(data, key) for key in quota.RULE_COLUMNS}
if existing:
table.update({"uid": existing["uid"], "label": data.label, **payload}, ["uid"])
uid = existing["uid"]
else:
uid = generate_uid()
table.insert(
{
"uid": uid,
"owner_kind": "user",
"owner_id": data.owner_id,
"label": data.label,
"created_at": "",
"updated_at": "",
"deleted_at": None,
"deleted_by": None,
**payload,
}
)
audit.record(
request,
"container.workspace.settings.update",
user=admin,
target_type="user",
target_uid=data.owner_id,
summary=f"{admin['username']} updated workspace quota",
metadata=payload,
)
return action_result(request, "/admin/workspaces", data={"uid": uid, **payload})
Reference in New Issue View Git Blame Copy Permalink