from __future__ import annotations import os from pathlib import Path from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from sqlalchemy.orm import Session from govoplan_admin.backend.governance import create_template, delete_template, update_template from govoplan_core.admin.settings import get_system_settings from govoplan_core.auth import ApiPrincipal, has_scope, require_any_scope, require_scope from govoplan_core.audit.logging import audit_from_principal, audit_operation_context from govoplan_admin.backend.db.models import GovernanceTemplate, GovernanceTemplateAssignment from govoplan_core.admin.common import AdminConflictError, AdminValidationError from govoplan_core.core.access import CAPABILITY_ACCESS_ADMINISTRATION, AccessAdministration from govoplan_core.core.change_sequence import ( decode_sequence_watermark, encode_sequence_watermark, max_sequence_id, record_change, sequence_entries_since, sequence_watermark_is_expired, ) from govoplan_core.core.pagination import KeysetCursorError, decode_keyset_cursor, encode_keyset_cursor, keyset_query_fingerprint from govoplan_core.core.configuration_control import ( ConfigurationChangeApproval, ConfigurationControlError, ensure_configuration_change_allowed, record_configuration_change_applied, ) from govoplan_core.core.module_management import ( PROTECTED_MODULES, ModuleInstallPlan, ModuleInstallPlanItem, ModuleManagementError, configured_enabled_modules, module_install_plan_commands, module_dependents, module_uninstall_plan_item, plan_desired_enabled_modules, save_desired_enabled_modules, save_module_install_plan, saved_desired_enabled_modules, saved_module_install_plan, ) from govoplan_core.core.module_installer import ( ModuleInstallerError, cancel_module_installer_request, default_installer_runtime_dir, list_module_installer_runs, list_module_installer_requests, module_install_catalog_companion_module_ids, module_installer_daemon_status, module_install_preflight, module_installer_lock_status, queue_module_installer_request, read_module_installer_run, read_module_installer_request, retry_module_installer_request, ) from govoplan_core.core.module_license import module_license_decision, module_license_diagnostics from govoplan_core.core.module_package_catalog import record_module_package_catalog_acceptance, validate_module_package_catalog from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, MaintenanceMode, saved_maintenance_mode, save_maintenance_mode from govoplan_core.core.lifecycle import ModuleLifecycleManager from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.runtime import get_registry from govoplan_core.db.session import get_session from govoplan_core.i18n import ( i18n_settings, normalize_enabled_language_codes, normalize_language_packages, system_i18n_payload, update_i18n_settings, ) from govoplan_core.privacy.retention import privacy_policy_from_settings, set_privacy_policy from govoplan_core.security.permissions import ALL_PERMISSIONS, effective_permission_count from govoplan_core.server.registry import available_module_manifests from govoplan_core.settings import settings as core_settings from govoplan_core.tenancy.scope import Tenant from govoplan_core.tenancy.service import tenant_counts from .schemas import ( AdminOverviewResponse, GovernanceTemplateCreateRequest, GovernanceTemplateItem, GovernanceTemplateListDeltaResponse, GovernanceTemplateListResponse, GovernanceTemplateUpdateRequest, MaintenanceModeItem, ModuleCatalogItem, ModuleCatalogResponse, ModuleInstallPreflightResponse, ModuleInstallerLockStatus, ModuleInstallerDaemonStatus, ModuleInstallerRequestCreateRequest, ModuleInstallerRequestItem, ModuleInstallerRequestListResponse, ModuleInstallerRunListResponse, ModuleInstallerRunSummary, ModuleInstallPlanResponse, ModuleInstallPlanUpdateRequest, ModulePackageCatalogItem, ModulePackageCatalogResponse, ModuleLicenseDiagnostics, ModuleStateUpdateRequest, PrivacyRetentionPolicyItem, SystemSettingsDeltaResponse, SystemSettingsItem, SystemSettingsUpdateRequest, ) router = APIRouter(prefix="/admin", tags=["admin"]) ADMIN_MODULE_ID = "admin" SYSTEM_SETTINGS_COLLECTION = "admin.system_settings" SYSTEM_SETTINGS_RESOURCE = "system_settings_section" GOVERNANCE_TEMPLATES_COLLECTION = "admin.governance_templates" GOVERNANCE_TEMPLATE_RESOURCE = "governance_template" INSTALLER_RUNS_CURSOR_SCOPE = "admin.installer.runs.v1" INSTALLER_REQUESTS_CURSOR_SCOPE = "admin.installer.requests.v1" DEFAULT_INSTALLER_HISTORY_PAGE_SIZE = 25 SYSTEM_SETTINGS_SECTIONS = ( "defaults", "tenant_capabilities", "languages", "privacy_retention_policy", "maintenance_mode", "settings", ) def _access_administration() -> AccessAdministration: registry = get_registry() if registry is None or not registry.has_capability(CAPABILITY_ACCESS_ADMINISTRATION): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Access administration capability is not configured") capability = registry.require_capability(CAPABILITY_ACCESS_ADMINISTRATION) if not isinstance(capability, AccessAdministration): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Access administration capability is invalid") return capability def _request_registry(request: Request) -> PlatformRegistry: registry = getattr(request.app.state, "govoplan_registry", None) if not isinstance(registry, PlatformRegistry): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="GovOPlaN module registry is not configured") return registry def _request_lifecycle(request: Request) -> ModuleLifecycleManager: lifecycle = getattr(request.app.state, "govoplan_lifecycle", None) if not isinstance(lifecycle, ModuleLifecycleManager): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="GovOPlaN module lifecycle manager is not configured") return lifecycle def _http_admin_error(exc: Exception) -> HTTPException: if isinstance(exc, AdminConflictError): return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) if isinstance(exc, AdminValidationError): return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) def _configuration_control_http_error(exc: ConfigurationControlError) -> HTTPException: return HTTPException( status_code=status.HTTP_409_CONFLICT, detail={"code": exc.code, "message": str(exc), "plan": exc.plan}, ) def _admin_delta_watermark(session: Session, collections: tuple[str, ...]) -> str: return encode_sequence_watermark(max_sequence_id(session, module_id=ADMIN_MODULE_ID, collections=collections)) def _admin_delta_entries(session: Session, *, collections: tuple[str, ...], since: str, limit: int): try: since_sequence = decode_sequence_watermark(since) except ValueError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc if sequence_watermark_is_expired( session, since=since_sequence, module_id=ADMIN_MODULE_ID, collections=collections, ): return None, False entries_plus_one = sequence_entries_since( session, since=since_sequence, module_id=ADMIN_MODULE_ID, collections=collections, limit=limit + 1, ) has_more = len(entries_plus_one) > limit return entries_plus_one[:limit], has_more def _admin_delta_response_watermark(session: Session, *, collections: tuple[str, ...], entries, has_more: bool) -> str: return encode_sequence_watermark(entries[-1].id) if has_more and entries else _admin_delta_watermark(session, collections) def _history_cursor_fingerprint(scope: str, *, page_size: int) -> str: return keyset_query_fingerprint(scope, {"page_size": page_size, "sort": "id.desc"}) def _window_history_records(records: list[dict[str, object]], *, id_key: str, scope: str, page_size: int, cursor: str | None): start = 0 full = False fingerprint = _history_cursor_fingerprint(scope, page_size=page_size) if cursor: try: values = decode_keyset_cursor(scope, cursor, fingerprint=fingerprint) except KeysetCursorError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc after_id = values.get(id_key) if values else None if not isinstance(after_id, str): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor") matching_index = next((index for index, item in enumerate(records) if str(item.get(id_key) or "") == after_id), None) if matching_index is None: full = True else: start = matching_index + 1 page = records[start:start + page_size] next_cursor = None if start + page_size < len(records) and page: next_cursor = encode_keyset_cursor(scope, fingerprint=fingerprint, values={id_key: str(page[-1].get(id_key) or ""), "page_size": page_size}) return page, next_cursor, full def _system_settings_item(session: Session) -> SystemSettingsItem: item = get_system_settings(session) policy = privacy_policy_from_settings(item) maintenance_mode = saved_maintenance_mode(session) i18n_payload = system_i18n_payload(item) return SystemSettingsItem( default_locale=item.default_locale, allow_tenant_custom_groups=item.allow_tenant_custom_groups, allow_tenant_custom_roles=item.allow_tenant_custom_roles, allow_tenant_api_keys=item.allow_tenant_api_keys, privacy_retention_policy=PrivacyRetentionPolicyItem.model_validate(policy.model_dump(mode="json")), maintenance_mode=MaintenanceModeItem.model_validate(maintenance_mode.as_dict()), available_languages=i18n_payload["available_languages"], enabled_language_codes=i18n_payload["enabled_languages"], settings=item.settings or {}, ) def _system_settings_sections(item: SystemSettingsItem) -> dict[str, Any]: payload = item.model_dump(mode="json") return { "defaults": {"default_locale": payload["default_locale"]}, "tenant_capabilities": { "allow_tenant_custom_groups": payload["allow_tenant_custom_groups"], "allow_tenant_custom_roles": payload["allow_tenant_custom_roles"], "allow_tenant_api_keys": payload["allow_tenant_api_keys"], }, "languages": { "available_languages": payload["available_languages"], "enabled_language_codes": payload["enabled_language_codes"], }, "privacy_retention_policy": payload["privacy_retention_policy"], "maintenance_mode": payload["maintenance_mode"], "settings": payload["settings"], } def _record_system_settings_section_changes( session: Session, *, before: dict[str, Any], after: dict[str, Any], principal: ApiPrincipal, ) -> None: for section in SYSTEM_SETTINGS_SECTIONS: if before.get(section) == after.get(section): continue record_change( session, module_id=ADMIN_MODULE_ID, collection=SYSTEM_SETTINGS_COLLECTION, resource_type=SYSTEM_SETTINGS_RESOURCE, resource_id=section, operation="updated", actor_type="user", actor_id=principal.user.id, payload={"section": section}, ) def _governance_template_item(session: Session, item: GovernanceTemplate) -> GovernanceTemplateItem: assignments = session.query(GovernanceTemplateAssignment).filter( GovernanceTemplateAssignment.template_id == item.id ).order_by(GovernanceTemplateAssignment.tenant_id.asc()).all() return GovernanceTemplateItem( id=item.id, kind=item.kind, slug=item.slug, name=item.name, description=item.description, permissions=item.permissions or [], effective_permission_count=effective_permission_count(item.permissions or [], level="tenant") if item.kind == "role" else 0, is_active=item.is_active, assignments=[{"tenant_id": row.tenant_id, "mode": row.mode} for row in assignments], created_at=item.created_at, updated_at=item.updated_at, ) def _record_governance_template_change(session: Session, *, item: GovernanceTemplate, operation: str, principal: ApiPrincipal) -> None: record_change( session, module_id=ADMIN_MODULE_ID, collection=GOVERNANCE_TEMPLATES_COLLECTION, resource_type=GOVERNANCE_TEMPLATE_RESOURCE, resource_id=item.id, operation=operation, actor_type="user", actor_id=principal.user.id, payload={"kind": item.kind, "slug": item.slug, "name": item.name, "is_active": item.is_active}, ) def _governance_template_deleted_entries(entries, visible_template_ids: set[str]): return [ {"id": entry.resource_id, "resource_type": entry.resource_type or GOVERNANCE_TEMPLATE_RESOURCE} for entry in entries if entry.resource_id and (entry.operation == "deleted" or entry.resource_id not in visible_template_ids) ] def _module_catalog_response(session: Session, request: Request, *, notes: list[str] | None = None) -> ModuleCatalogResponse: registry = _request_registry(request) current_enabled = [manifest.id for manifest in registry.manifests()] configured_enabled = list(configured_enabled_modules(core_settings.enabled_modules)) desired_enabled = list(saved_desired_enabled_modules(session, configured_enabled)) lifecycle = getattr(request.app.state, "govoplan_lifecycle", None) available = dict(lifecycle.available_modules) if isinstance(lifecycle, ModuleLifecycleManager) else available_module_manifests(ignore_load_errors=True) dependents = module_dependents(available) current_set = set(current_enabled) desired_set = set(desired_enabled) protected_set = set(PROTECTED_MODULES) items: list[ModuleCatalogItem] = [] for module_id, manifest in sorted(available.items(), key=lambda item: item[0]): migration = manifest.migration_spec frontend = manifest.frontend items.append(ModuleCatalogItem( id=manifest.id, name=manifest.name, version=manifest.version, installed=True, current_enabled=manifest.id in current_set, desired_enabled=manifest.id in desired_set, protected=manifest.id in protected_set, restart_required=(manifest.id in current_set) != (manifest.id in desired_set), dependencies=list(manifest.dependencies), optional_dependencies=list(manifest.optional_dependencies), dependents=list(dependents.get(manifest.id, ())), permissions_count=len(manifest.permissions), role_templates_count=len(manifest.role_templates), route_contributed=manifest.route_factory is not None, nav_count=len(manifest.nav_items), frontend_package=frontend.package_name if frontend else None, migration_module_id=migration.module_id if migration else None, migration_script_location=migration.script_location if migration else None, runtime_toggle_supported=True, install_uninstall_supported=manifest.id not in protected_set, )) restart_required = current_set != desired_set maintenance_mode = saved_maintenance_mode(session) return ModuleCatalogResponse( modules=items, current_enabled=current_enabled, desired_enabled=desired_enabled, configured_enabled=configured_enabled, protected_modules=list(PROTECTED_MODULES), restart_required=restart_required, runtime_toggle_supported=True, install_uninstall_supported=True, install_plan_supported=True, package_mutation_supported=False, maintenance_mode=MaintenanceModeItem.model_validate(maintenance_mode.as_dict()), notes=notes or [ "Enable/disable changes are applied to the running server and saved as startup state.", "Installing or uninstalling Python/WebUI packages is planned here and applied by the separate installer daemon during maintenance mode.", ], ) def _module_install_plan_response(session: Session, request: Request, *, notes: list[str] | None = None) -> ModuleInstallPlanResponse: plan = saved_module_install_plan(session) maintenance_mode = saved_maintenance_mode(session) registry = _request_registry(request) current_enabled = [manifest.id for manifest in registry.manifests()] configured_enabled = list(configured_enabled_modules(core_settings.enabled_modules)) desired_enabled = list(saved_desired_enabled_modules(session, configured_enabled)) lifecycle = getattr(request.app.state, "govoplan_lifecycle", None) available = dict(lifecycle.available_modules) if isinstance(lifecycle, ModuleLifecycleManager) else available_module_manifests(ignore_load_errors=True) preflight = module_install_preflight( plan=plan, available=available, current_enabled=current_enabled, desired_enabled=desired_enabled, maintenance_mode=maintenance_mode.enabled, session=session, webui_root=_webui_root(), runtime_dir=default_installer_runtime_dir(core_settings.database_url), ) default_notes = [ "The running server does not install or remove packages inside the FastAPI request.", "Use govoplan-module-installer from an operator shell while maintenance mode is active.", ] if not maintenance_mode.enabled: default_notes.append("Maintenance mode is currently off.") return ModuleInstallPlanResponse( items=[item.as_dict() for item in plan.items], updated_at=plan.updated_at, commands=list(module_install_plan_commands(plan)), maintenance_mode=MaintenanceModeItem.model_validate(maintenance_mode.as_dict()), preflight=ModuleInstallPreflightResponse.model_validate(preflight.as_dict()), notes=notes or default_notes, ) def _webui_root() -> Path | None: configured = os.getenv("GOVOPLAN_WEBUI_ROOT") if configured: return Path(configured).expanduser().resolve() candidate = Path.cwd() / "webui" return candidate if candidate.exists() else None def _upsert_install_plan_item(session: Session, item: ModuleInstallPlanItem): return _upsert_install_plan_items(session, [item]) def _upsert_install_plan_items(session: Session, items: list[ModuleInstallPlanItem]): existing = saved_module_install_plan(session) replacement_ids = {item.module_id for item in items if item.status == "planned"} retained = [ current for current in existing.items if not (current.status == "planned" and current.module_id in replacement_ids) ] return save_module_install_plan(session, [*retained, *items]) def _catalog_plan_item( module_id: str, available_module_ids: set[str], *, validation: dict[str, object] | None = None, ) -> tuple[ModuleInstallPlanItem, dict[str, object]]: result = validation or validate_module_package_catalog() if not result.get("valid"): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(result.get("error") or "Module package catalog is invalid."), ) for raw_item in result.get("modules", []): if not isinstance(raw_item, dict): continue if raw_item.get("module_id") != module_id or raw_item.get("action") not in {"install", "update"}: continue raw_action = raw_item.get("action") action = "update" if raw_action == "update" or module_id in available_module_ids else "install" license_decision = module_license_decision(_catalog_license_features(raw_item)) if not license_decision.get("allowed"): missing = license_decision.get("missing_features") missing_text = ", ".join(str(item) for item in missing) if isinstance(missing, list) else "required feature" raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"License does not allow {action} for {module_id}: {missing_text}.", ) notes = raw_item.get("notes") if isinstance(raw_item.get("notes"), str) else None if license_decision.get("reason") and license_decision.get("missing_features"): prefix = f"{notes}\n" if notes else "" notes = f"{prefix}License warning: {license_decision['reason']}" return ModuleInstallPlanItem( module_id=str(raw_item["module_id"]), action=action, source="catalog", catalog=_catalog_plan_metadata(result), python_package=raw_item.get("python_package") if isinstance(raw_item.get("python_package"), str) else None, python_ref=raw_item.get("python_ref") if isinstance(raw_item.get("python_ref"), str) else None, webui_package=raw_item.get("webui_package") if isinstance(raw_item.get("webui_package"), str) else None, webui_ref=raw_item.get("webui_ref") if isinstance(raw_item.get("webui_ref"), str) else None, notes=notes, ), result raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Catalog install/update entry not found: {module_id}") def _catalog_plan_metadata(validation: dict[str, object]) -> dict[str, object]: metadata: dict[str, object] = { "source": validation.get("source") or validation.get("path"), "source_type": validation.get("source_type"), "cache_used": bool(validation.get("cache_used")), "cache_path": validation.get("cache_path"), "channel": validation.get("channel"), "sequence": validation.get("sequence"), "generated_at": validation.get("generated_at"), "not_before": validation.get("not_before"), "expires_at": validation.get("expires_at"), "signed": bool(validation.get("signed")), "trusted": bool(validation.get("trusted")), "key_id": validation.get("key_id"), } return { key: value for key, value in metadata.items() if value is not None and value != "" } def _catalog_license_features(raw_item: dict[str, object]) -> list[str]: value = raw_item.get("license_features") if not isinstance(value, list): return [] return [str(item).strip() for item in value if str(item).strip()] def _module_package_catalog_item(raw_item: dict[str, object]) -> ModulePackageCatalogItem: payload = dict(raw_item) decision = module_license_decision(_catalog_license_features(raw_item)) payload["license_allowed"] = bool(decision.get("allowed")) payload["license_enforced"] = bool(decision.get("enforced")) missing = decision.get("missing_features") payload["license_missing_features"] = [str(item) for item in missing] if isinstance(missing, list) else [] reason = decision.get("reason") payload["license_reason"] = reason if isinstance(reason, str) else None return ModulePackageCatalogItem.model_validate(payload) def _catalog_required_license_features(result: dict[str, object]) -> list[str]: required: list[str] = [] for raw_item in result.get("modules", []): if isinstance(raw_item, dict): required.extend(_catalog_license_features(raw_item)) return list(dict.fromkeys(required)) @router.get("/overview", response_model=AdminOverviewResponse) def admin_overview( session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_any_scope( "admin:users:read", "admin:groups:read", "admin:roles:read", "admin:settings:read", "admin:api_keys:read", "system:tenants:read", "system:accounts:read", "system:roles:read", "system:access:read", "system:settings:read", "system:governance:read", "system:audit:read", )), ): tenant = session.get(Tenant, principal.tenant_id) if tenant is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found") counts = tenant_counts(session, tenant.id) access_admin = _access_administration() capabilities = [item.scope for item in ALL_PERMISSIONS if has_scope(principal, item.scope)] return AdminOverviewResponse( active_tenant_id=tenant.id, active_tenant_name=tenant.name, tenant_count=session.query(Tenant).count() if has_scope(principal, "system:tenants:read") else None, system_account_count=access_admin.system_account_count(session) if has_scope(principal, "system:accounts:read") or has_scope(principal, "system:access:read") else None, system_group_template_count=session.query(GovernanceTemplate).filter(GovernanceTemplate.kind == "group").count() if has_scope(principal, "system:governance:read") else None, system_role_template_count=session.query(GovernanceTemplate).filter(GovernanceTemplate.kind == "role").count() if has_scope(principal, "system:governance:read") else None, user_count=counts["users"], active_user_count=counts["active_users"], group_count=counts["groups"], role_count=access_admin.role_count_for_tenant(session, tenant.id), active_api_key_count=access_admin.active_api_key_count_for_tenant(session, tenant.id), capabilities=capabilities, ) @router.get("/system/modules", response_model=ModuleCatalogResponse) def list_system_modules( request: Request, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): del principal return _module_catalog_response(session, request) @router.get("/system/modules/install-plan", response_model=ModuleInstallPlanResponse) def read_module_install_plan( request: Request, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): del principal return _module_install_plan_response(session, request) @router.get("/system/modules/install-runs", response_model=ModuleInstallerRunListResponse) def list_module_install_runs( page_size: int = Query(default=DEFAULT_INSTALLER_HISTORY_PAGE_SIZE, ge=1, le=100), cursor: str | None = None, principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): del principal runtime_dir = default_installer_runtime_dir(core_settings.database_url) records = list(list_module_installer_runs(runtime_dir=runtime_dir, limit=1000)) page, next_cursor, full = _window_history_records(records, id_key="run_id", scope=INSTALLER_RUNS_CURSOR_SCOPE, page_size=page_size, cursor=cursor) return ModuleInstallerRunListResponse( runs=[ ModuleInstallerRunSummary.model_validate(item) for item in page ], lock=ModuleInstallerLockStatus.model_validate(module_installer_lock_status(runtime_dir=runtime_dir)), cursor=cursor, next_cursor=next_cursor, full=full, ) @router.get("/system/modules/install-runs/{run_id}") def read_module_install_run_detail( run_id: str, principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): del principal runtime_dir = default_installer_runtime_dir(core_settings.database_url) try: return read_module_installer_run(runtime_dir=runtime_dir, run_id=run_id) except ModuleInstallerError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.get("/system/modules/install-requests", response_model=ModuleInstallerRequestListResponse) def list_module_install_requests( page_size: int = Query(default=DEFAULT_INSTALLER_HISTORY_PAGE_SIZE, ge=1, le=100), cursor: str | None = None, principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): del principal runtime_dir = default_installer_runtime_dir(core_settings.database_url) records = list(list_module_installer_requests(runtime_dir=runtime_dir, limit=1000)) page, next_cursor, full = _window_history_records(records, id_key="request_id", scope=INSTALLER_REQUESTS_CURSOR_SCOPE, page_size=page_size, cursor=cursor) return ModuleInstallerRequestListResponse( requests=[ ModuleInstallerRequestItem.model_validate(item) for item in page ], daemon=ModuleInstallerDaemonStatus.model_validate(module_installer_daemon_status(runtime_dir=runtime_dir)), cursor=cursor, next_cursor=next_cursor, full=full, ) @router.get("/system/modules/install-requests/{request_id}") def read_module_install_request_detail( request_id: str, principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): del principal runtime_dir = default_installer_runtime_dir(core_settings.database_url) try: return read_module_installer_request(runtime_dir=runtime_dir, request_id=request_id) except ModuleInstallerError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.post("/system/modules/install-requests", response_model=ModuleInstallerRequestItem) def create_module_install_request( payload: ModuleInstallerRequestCreateRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): maintenance_mode = saved_maintenance_mode(session) if not maintenance_mode.enabled: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Maintenance mode must be enabled before queuing installer requests.") if not has_scope(principal, MAINTENANCE_ACCESS_SCOPE): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Queuing installer requests requires maintenance access.") runtime_dir = default_installer_runtime_dir(core_settings.database_url) request = queue_module_installer_request( runtime_dir=runtime_dir, requested_by=principal.user.id, options=payload.options.model_dump(exclude_none=True), ) audit_from_principal( session, principal, action="system_modules.install_request_queued", scope="system", object_type="module_install_request", object_id=str(request["request_id"]), details=audit_operation_context( request_id=request.get("request_id"), outcome=request.get("status"), trace=request.get("trace") if isinstance(request.get("trace"), dict) else None, options=payload.options.model_dump(exclude_none=True), ), ) session.commit() return ModuleInstallerRequestItem.model_validate(request) @router.post("/system/modules/install-requests/{request_id}/cancel", response_model=ModuleInstallerRequestItem) def cancel_module_install_request( request_id: str, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): maintenance_mode = saved_maintenance_mode(session) if not maintenance_mode.enabled: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Maintenance mode must be enabled before changing installer requests.") if not has_scope(principal, MAINTENANCE_ACCESS_SCOPE): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Changing installer requests requires maintenance access.") runtime_dir = default_installer_runtime_dir(core_settings.database_url) try: request = cancel_module_installer_request( runtime_dir=runtime_dir, request_id=request_id, cancelled_by=principal.user.id, ) except ModuleInstallerError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc audit_from_principal( session, principal, action="system_modules.install_request_cancelled", scope="system", object_type="module_install_request", object_id=request_id, details=audit_operation_context( request_id=request_id, outcome=request.get("status"), trace=request.get("trace") if isinstance(request.get("trace"), dict) else None, cancelled_by=request.get("cancelled_by"), ), ) session.commit() return ModuleInstallerRequestItem.model_validate(request) @router.post("/system/modules/install-requests/{request_id}/retry", response_model=ModuleInstallerRequestItem) def retry_module_install_request( request_id: str, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): maintenance_mode = saved_maintenance_mode(session) if not maintenance_mode.enabled: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Maintenance mode must be enabled before changing installer requests.") if not has_scope(principal, MAINTENANCE_ACCESS_SCOPE): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Changing installer requests requires maintenance access.") runtime_dir = default_installer_runtime_dir(core_settings.database_url) try: request = retry_module_installer_request( runtime_dir=runtime_dir, request_id=request_id, requested_by=principal.user.id, ) except ModuleInstallerError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc audit_from_principal( session, principal, action="system_modules.install_request_retried", scope="system", object_type="module_install_request", object_id=request_id, details=audit_operation_context( request_id=request_id, outcome=request.get("status"), trace=request.get("trace") if isinstance(request.get("trace"), dict) else None, retry_of=request.get("retry_of"), new_request_id=request.get("request_id"), ), ) session.commit() return ModuleInstallerRequestItem.model_validate(request) @router.get("/system/modules/package-catalog", response_model=ModulePackageCatalogResponse) def read_module_package_catalog( principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): del principal result = validate_module_package_catalog() license_diagnostics = module_license_diagnostics(required_features=_catalog_required_license_features(result)) return ModulePackageCatalogResponse( modules=[_module_package_catalog_item(item) for item in result["modules"] if isinstance(item, dict)], configured=bool(result["configured"]), valid=bool(result["valid"]), path=result["path"] if isinstance(result["path"], str) else None, source=result["source"] if isinstance(result.get("source"), str) else None, source_type=result["source_type"] if isinstance(result.get("source_type"), str) else None, cache_used=bool(result.get("cache_used")), cache_path=result["cache_path"] if isinstance(result.get("cache_path"), str) else None, channel=result["channel"] if isinstance(result["channel"], str) else None, sequence=result["sequence"] if isinstance(result.get("sequence"), int) else None, generated_at=result["generated_at"] if isinstance(result.get("generated_at"), str) else None, not_before=result["not_before"] if isinstance(result.get("not_before"), str) else None, expires_at=result["expires_at"] if isinstance(result.get("expires_at"), str) else None, signed=bool(result["signed"]), trusted=bool(result["trusted"]), key_id=result["key_id"] if isinstance(result["key_id"], str) else None, license=ModuleLicenseDiagnostics.model_validate(license_diagnostics), warnings=[str(item) for item in result["warnings"]] if isinstance(result["warnings"], list) else [], error=result["error"] if isinstance(result["error"], str) else None, ) @router.post("/system/modules/install-plan/catalog/{module_id}", response_model=ModuleInstallPlanResponse) def plan_module_install_from_catalog( module_id: str, request: Request, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): try: lifecycle = getattr(request.app.state, "govoplan_lifecycle", None) available = dict(lifecycle.available_modules) if isinstance(lifecycle, ModuleLifecycleManager) else available_module_manifests(ignore_load_errors=True) item, validation = _catalog_plan_item(module_id, set(available)) existing = saved_module_install_plan(session) retained = [ current for current in existing.items if not (current.status == "planned" and current.module_id == item.module_id) ] candidate_plan = ModuleInstallPlan(items=tuple([*retained, item])) companion_ids = module_install_catalog_companion_module_ids(candidate_plan, available, validation=validation) companion_items = [ _catalog_plan_item(companion_id, set(available), validation=validation)[0] for companion_id in companion_ids ] plan = _upsert_install_plan_items(session, [item, *companion_items]) record_module_package_catalog_acceptance(validation) except ModuleManagementError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc audit_from_principal( session, principal, action="system_modules.install_plan_catalog_added", scope="system", object_type="module_install_plan", object_id=module_id, details=audit_operation_context( module_id=module_id, outcome="planned", items=[item.as_dict() for item in plan.items], companion_module_ids=list(companion_ids), catalog_validation={ "valid": validation.get("valid"), "channel": validation.get("channel"), "key_id": validation.get("key_id"), }, ), ) session.commit() planned_action = "update" if item.action == "update" else "install" notes = [f"Catalog {planned_action} entry planned for {module_id}."] if companion_ids: notes.append("Required companion catalog entries added: " + ", ".join(companion_ids)) return _module_install_plan_response(session, request, notes=notes) @router.post("/system/modules/{module_id}/uninstall-plan", response_model=ModuleInstallPlanResponse) def plan_module_uninstall( module_id: str, request: Request, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): registry = _request_registry(request) lifecycle = _request_lifecycle(request) current_enabled = {manifest.id for manifest in registry.manifests()} configured_enabled = configured_enabled_modules(core_settings.enabled_modules) desired_enabled = set(saved_desired_enabled_modules(session, configured_enabled)) if module_id in PROTECTED_MODULES: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Protected platform modules cannot be uninstalled.") if module_id in current_enabled or module_id in desired_enabled: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Disable the module before planning package uninstall.") try: item = module_uninstall_plan_item(module_id, lifecycle.available_modules) plan = _upsert_install_plan_item(session, item) except ModuleManagementError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc audit_from_principal( session, principal, action="system_modules.uninstall_plan_added", scope="system", object_type="module_install_plan", object_id=module_id, details=audit_operation_context( module_id=module_id, outcome="planned", items=[item.as_dict() for item in plan.items], ), ) session.commit() return _module_install_plan_response(session, request, notes=[f"Package uninstall planned for {module_id}."]) @router.put("/system/modules", response_model=ModuleCatalogResponse) def update_system_modules( request: Request, payload: ModuleStateUpdateRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): lifecycle = _request_lifecycle(request) available = lifecycle.available_modules before_value = {"enabled_modules": list(saved_desired_enabled_modules(session, configured_enabled_modules(core_settings.enabled_modules)))} apply_value = {"enabled_modules": list(payload.enabled_modules)} try: approval = ensure_configuration_change_allowed( session, key="module_management.desired_enabled", value=apply_value, actor_user_id=principal.user.id, actor_scopes=tuple(principal.scopes), change_request_id=payload.change_request_id, target={"scope": "system"}, ) except ConfigurationControlError as exc: raise _configuration_control_http_error(exc) from exc try: plan = plan_desired_enabled_modules(payload.enabled_modules, available) except ModuleManagementError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc try: result = lifecycle.apply_enabled_modules(plan.enabled_modules, protected_modules=PROTECTED_MODULES) except Exception as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc desired = save_desired_enabled_modules(session, result.enabled_modules) audit_from_principal( session, principal, action="system_modules.desired_state_updated", scope="system", object_type="module_state", object_id="global", details=audit_operation_context( outcome="applied", desired_enabled=list(desired), added_dependencies=list(plan.added_dependencies), activated=list(result.activated_modules), deactivated=list(result.deactivated_modules), mounted=list(result.mounted_modules), ), ) record_configuration_change_applied( session, key="module_management.desired_enabled", before_value=before_value, after_value={"enabled_modules": list(desired)}, actor_user_id=principal.user.id, approval=approval, target={"scope": "system"}, audit_event="module_management.updated", ) session.commit() notes = ["Module state saved and applied to the running server."] if plan.added_dependencies: notes.append("Required dependencies added automatically: " + ", ".join(plan.added_dependencies)) if result.activated_modules: notes.append("Activated: " + ", ".join(result.activated_modules)) if result.deactivated_modules: notes.append("Deactivated: " + ", ".join(result.deactivated_modules)) return _module_catalog_response(session, request, notes=notes) @router.put("/system/modules/install-plan", response_model=ModuleInstallPlanResponse) def update_module_install_plan( request: Request, payload: ModuleInstallPlanUpdateRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): before_value = {"items": [item.as_dict() for item in saved_module_install_plan(session).items]} apply_value = {"items": [item.model_dump() for item in payload.items]} try: approval = ensure_configuration_change_allowed( session, key="module_management.install_plan", value=apply_value, actor_user_id=principal.user.id, actor_scopes=tuple(principal.scopes), change_request_id=payload.change_request_id, target={"scope": "system"}, ) except ConfigurationControlError as exc: raise _configuration_control_http_error(exc) from exc try: plan = save_module_install_plan(session, [item.model_dump() for item in payload.items]) except ModuleManagementError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc audit_from_principal( session, principal, action="system_modules.install_plan_updated", scope="system", object_type="module_install_plan", object_id="global", details=audit_operation_context( outcome="saved", items=[item.as_dict() for item in plan.items], item_count=len(plan.items), ), ) record_configuration_change_applied( session, key="module_management.install_plan", before_value=before_value, after_value={"items": [item.as_dict() for item in plan.items]}, actor_user_id=principal.user.id, approval=approval, target={"scope": "system"}, audit_event="module_install.requested", ) session.commit() return _module_install_plan_response(session, request, notes=["Module package install plan saved."]) @router.delete("/system/modules/install-plan", response_model=ModuleInstallPlanResponse) def clear_module_install_plan( request: Request, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): plan = save_module_install_plan(session, ()) audit_from_principal( session, principal, action="system_modules.install_plan_cleared", scope="system", object_type="module_install_plan", object_id="global", details=audit_operation_context( outcome="cleared", items=[item.as_dict() for item in plan.items], item_count=len(plan.items), ), ) session.commit() return _module_install_plan_response(session, request, notes=["Module package install plan cleared."]) @router.get("/system/settings", response_model=SystemSettingsItem) def read_system_settings( session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): get_system_settings(session) session.commit() return _system_settings_item(session) def _full_system_settings_delta_response(session: Session) -> SystemSettingsDeltaResponse: item = _system_settings_item(session) return SystemSettingsDeltaResponse( item=item, sections=_system_settings_sections(item), changed_sections=list(SYSTEM_SETTINGS_SECTIONS), deleted=[], watermark=_admin_delta_watermark(session, (SYSTEM_SETTINGS_COLLECTION,)), has_more=False, full=True, ) @router.get("/system/settings/delta", response_model=SystemSettingsDeltaResponse) def read_system_settings_delta( since: str | None = None, limit: int = Query(default=100, ge=1, le=500), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:read")), ): del principal get_system_settings(session) session.commit() if since is None: return _full_system_settings_delta_response(session) entries, has_more = _admin_delta_entries(session, collections=(SYSTEM_SETTINGS_COLLECTION,), since=since, limit=limit) if entries is None: return _full_system_settings_delta_response(session) changed_sections = [ section for section in SYSTEM_SETTINGS_SECTIONS if section in {entry.resource_id for entry in entries if entry.resource_type == SYSTEM_SETTINGS_RESOURCE} ] item = _system_settings_item(session) sections = _system_settings_sections(item) return SystemSettingsDeltaResponse( item=None, sections={section: sections[section] for section in changed_sections}, changed_sections=changed_sections, deleted=[], watermark=_admin_delta_response_watermark( session, collections=(SYSTEM_SETTINGS_COLLECTION,), entries=entries, has_more=has_more, ), has_more=has_more, full=False, ) @router.patch("/system/settings", response_model=SystemSettingsItem) def write_system_settings( payload: SystemSettingsUpdateRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:settings:write")), ): item = get_system_settings(session) privacy_approval: ConfigurationChangeApproval | None = None maintenance_approval: ConfigurationChangeApproval | None = None before_sections = _system_settings_sections(_system_settings_item(session)) before_privacy = privacy_policy_from_settings(item).model_dump(mode="json") before_maintenance = saved_maintenance_mode(session).as_dict() if payload.privacy_retention_policy is not None: privacy_value = payload.privacy_retention_policy.model_dump(mode="json") try: privacy_approval = ensure_configuration_change_allowed( session, key="privacy_retention_policy", value=privacy_value, actor_user_id=principal.user.id, actor_scopes=tuple(principal.scopes), change_request_id=payload.change_request_id, target={"scope": "system"}, ) except ConfigurationControlError as exc: raise _configuration_control_http_error(exc) from exc current_i18n = i18n_settings(item.settings) raw_available = payload.available_languages if "available_languages" in payload.model_fields_set else current_i18n.get("available_languages") raw_enabled = payload.enabled_language_codes if "enabled_language_codes" in payload.model_fields_set else current_i18n.get("enabled_language_codes", current_i18n.get("enabled_languages")) available_languages = normalize_language_packages( [item.model_dump() for item in raw_available] if payload.available_languages is not None else raw_available ) enabled_language_codes = normalize_enabled_language_codes( raw_enabled, available_languages, default_locale=payload.default_locale, ) item.default_locale = payload.default_locale.strip() or "en" item.allow_tenant_custom_groups = payload.allow_tenant_custom_groups item.allow_tenant_custom_roles = payload.allow_tenant_custom_roles item.allow_tenant_api_keys = payload.allow_tenant_api_keys item.settings = update_i18n_settings( item.settings, available_languages=available_languages, enabled_language_codes=enabled_language_codes, ) if payload.privacy_retention_policy is not None: set_privacy_policy(item, payload.privacy_retention_policy.model_dump(mode="json")) if payload.maintenance_mode is not None: current = saved_maintenance_mode(session) next_mode = MaintenanceMode( enabled=payload.maintenance_mode.enabled, message=payload.maintenance_mode.message.strip() if payload.maintenance_mode.message else None, ) try: maintenance_approval = ensure_configuration_change_allowed( session, key="maintenance_mode", value=next_mode.as_dict(), actor_user_id=principal.user.id, actor_scopes=tuple(principal.scopes), change_request_id=None, target={"scope": "system"}, ) except ConfigurationControlError as exc: raise _configuration_control_http_error(exc) from exc if current.enabled != next_mode.enabled and not has_scope(principal, MAINTENANCE_ACCESS_SCOPE): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=f"Changing maintenance mode requires {MAINTENANCE_ACCESS_SCOPE}.", ) save_maintenance_mode(session, next_mode) session.add(item) audit_from_principal( session, principal, action="system_settings.updated", scope="system", object_type="system_settings", object_id=item.id, details=payload.model_dump(), ) if payload.privacy_retention_policy is not None: record_configuration_change_applied( session, key="privacy_retention_policy", before_value=before_privacy, after_value=privacy_policy_from_settings(item).model_dump(mode="json"), actor_user_id=principal.user.id, approval=privacy_approval, target={"scope": "system"}, audit_event="privacy_retention.policy_updated", ) if payload.maintenance_mode is not None: record_configuration_change_applied( session, key="maintenance_mode", before_value=before_maintenance, after_value=saved_maintenance_mode(session).as_dict(), actor_user_id=principal.user.id, approval=maintenance_approval, target={"scope": "system"}, audit_event="system_settings.updated", ) after_sections = _system_settings_sections(_system_settings_item(session)) _record_system_settings_section_changes(session, before=before_sections, after=after_sections, principal=principal) session.commit() return _system_settings_item(session) @router.get("/system/governance-templates", response_model=GovernanceTemplateListResponse) def list_governance_templates( kind: str | None = Query(default=None), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:governance:read")), ): del principal query = session.query(GovernanceTemplate) if kind: query = query.filter(GovernanceTemplate.kind == kind) items = query.order_by(GovernanceTemplate.kind.asc(), GovernanceTemplate.name.asc()).all() return GovernanceTemplateListResponse(templates=[_governance_template_item(session, item) for item in items]) def _full_governance_template_delta_response(session: Session) -> GovernanceTemplateListDeltaResponse: items = session.query(GovernanceTemplate).order_by(GovernanceTemplate.kind.asc(), GovernanceTemplate.name.asc()).all() return GovernanceTemplateListDeltaResponse( templates=[_governance_template_item(session, item) for item in items], deleted=[], watermark=_admin_delta_watermark(session, (GOVERNANCE_TEMPLATES_COLLECTION,)), has_more=False, full=True, ) @router.get("/system/governance-templates/delta", response_model=GovernanceTemplateListDeltaResponse) def list_governance_templates_delta( since: str | None = None, limit: int = Query(default=100, ge=1, le=500), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:governance:read")), ): del principal if since is None: return _full_governance_template_delta_response(session) entries, has_more = _admin_delta_entries(session, collections=(GOVERNANCE_TEMPLATES_COLLECTION,), since=since, limit=limit) if entries is None: return _full_governance_template_delta_response(session) changed_ids = [entry.resource_id for entry in entries if entry.resource_id and entry.operation != "deleted"] items = [] if changed_ids: items = session.query(GovernanceTemplate).filter(GovernanceTemplate.id.in_(changed_ids)).order_by(GovernanceTemplate.kind.asc(), GovernanceTemplate.name.asc()).all() visible_template_ids = {item.id for item in items} return GovernanceTemplateListDeltaResponse( templates=[_governance_template_item(session, item) for item in items], deleted=_governance_template_deleted_entries(entries, visible_template_ids), watermark=_admin_delta_response_watermark(session, collections=(GOVERNANCE_TEMPLATES_COLLECTION,), entries=entries, has_more=has_more), has_more=has_more, full=False, ) @router.post("/system/governance-templates", response_model=GovernanceTemplateItem, status_code=status.HTTP_201_CREATED) def create_governance_template( payload: GovernanceTemplateCreateRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:governance:write")), ): apply_value = payload.model_dump(exclude={"change_request_id"}) try: approval = ensure_configuration_change_allowed( session, key="governance_templates", value=apply_value, actor_user_id=principal.user.id, actor_scopes=tuple(principal.scopes), change_request_id=payload.change_request_id, target={"kind": payload.kind, "slug": payload.slug}, ) except ConfigurationControlError as exc: raise _configuration_control_http_error(exc) from exc try: item = create_template( session, kind=payload.kind, slug=payload.slug, name=payload.name, description=payload.description, permissions=payload.permissions, is_active=payload.is_active, assignments=[assignment.model_dump() for assignment in payload.assignments], ) except (AdminConflictError, AdminValidationError) as exc: raise _http_admin_error(exc) from exc audit_from_principal( session, principal, action="governance_template.created", scope="system", object_type=f"{payload.kind}_template", object_id=item.id, details={"slug": item.slug, "tenant_ids": [assignment.tenant_id for assignment in payload.assignments]}, ) record_configuration_change_applied( session, key="governance_templates", before_value=None, after_value={**apply_value, "id": item.id}, actor_user_id=principal.user.id, approval=approval, target={"kind": item.kind, "id": item.id}, audit_event="governance_template.updated", ) _record_governance_template_change(session, item=item, operation="created", principal=principal) session.commit() return _governance_template_item(session, item) @router.patch("/system/governance-templates/{template_id}", response_model=GovernanceTemplateItem) def update_governance_template( template_id: str, payload: GovernanceTemplateUpdateRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:governance:write")), ): item = session.get(GovernanceTemplate, template_id) if item is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Governance template not found") before_value = _governance_template_item(session, item).model_dump(mode="json") apply_value = payload.model_dump(exclude={"change_request_id"}) try: approval = ensure_configuration_change_allowed( session, key="governance_templates", value=apply_value, actor_user_id=principal.user.id, actor_scopes=tuple(principal.scopes), change_request_id=payload.change_request_id, target={"kind": item.kind, "id": item.id}, ) except ConfigurationControlError as exc: raise _configuration_control_http_error(exc) from exc try: update_template( session, item, name=payload.name, description=payload.description, permissions=payload.permissions, is_active=payload.is_active, assignments=[assignment.model_dump() for assignment in payload.assignments], ) except (AdminConflictError, AdminValidationError) as exc: raise _http_admin_error(exc) from exc audit_from_principal( session, principal, action="governance_template.updated", scope="system", object_type=f"{item.kind}_template", object_id=item.id, details={"tenant_ids": [assignment.tenant_id for assignment in payload.assignments]}, ) record_configuration_change_applied( session, key="governance_templates", before_value=before_value, after_value=_governance_template_item(session, item).model_dump(mode="json"), actor_user_id=principal.user.id, approval=approval, target={"kind": item.kind, "id": item.id}, audit_event="governance_template.updated", ) _record_governance_template_change(session, item=item, operation="updated", principal=principal) session.commit() return _governance_template_item(session, item) @router.delete("/system/governance-templates/{template_id}", status_code=status.HTTP_204_NO_CONTENT) def remove_governance_template( template_id: str, change_request_id: str | None = Query(default=None), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("system:governance:write")), ): item = session.get(GovernanceTemplate, template_id) if item is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Governance template not found") kind = item.kind before_value = _governance_template_item(session, item).model_dump(mode="json") apply_value = {"id": item.id, "kind": item.kind, "delete": True} try: approval = ensure_configuration_change_allowed( session, key="governance_templates", value=apply_value, actor_user_id=principal.user.id, actor_scopes=tuple(principal.scopes), change_request_id=change_request_id, target={"kind": item.kind, "id": item.id}, ) except ConfigurationControlError as exc: raise _configuration_control_http_error(exc) from exc try: delete_template(session, item) except (AdminConflictError, AdminValidationError) as exc: raise _http_admin_error(exc) from exc audit_from_principal( session, principal, action="governance_template.deleted", scope="system", object_type=f"{kind}_template", object_id=template_id, details={}, ) record_configuration_change_applied( session, key="governance_templates", before_value=before_value, after_value=None, actor_user_id=principal.user.id, approval=approval, target={"kind": kind, "id": template_id}, audit_event="governance_template.updated", ) _record_governance_template_change(session, item=item, operation="deleted", principal=principal) session.commit() return None