Initialize configurable Quick Access module
Module Package Release / publish-packages (push) Successful in 12s

This commit is contained in:
2026-08-06 19:02:54 +02:00
parent f3cf91b898
commit 3fbbb84267
34 changed files with 3086 additions and 1 deletions
+3
View File
@@ -0,0 +1,3 @@
"""GovOPlaN Quick Access module."""
__version__ = "0.1.18"
@@ -0,0 +1 @@
"""Quick Access backend."""
@@ -0,0 +1,3 @@
from govoplan_quick_access.backend.db.models import QuickAccessProfile
__all__ = ["QuickAccessProfile"]
@@ -0,0 +1,35 @@
from __future__ import annotations
import uuid
from typing import Any
from sqlalchemy import Index, Integer, JSON, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from govoplan_core.db.base import Base, TimestampMixin
def new_uuid() -> str:
return str(uuid.uuid4())
class QuickAccessProfile(Base, TimestampMixin):
__tablename__ = "quick_access_profiles"
__table_args__ = (
UniqueConstraint("scope_key", name="uq_quick_access_profile_scope"),
Index("ix_quick_access_profiles_scope", "scope_type", "tenant_id", "scope_id"),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
scope_type: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
tenant_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
scope_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
scope_key: Mapped[str] = mapped_column(String(340), nullable=False, index=True)
category_preferences: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
tool_preferences: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
created_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
__all__ = ["QuickAccessProfile", "new_uuid"]
@@ -0,0 +1,317 @@
from __future__ import annotations
from pathlib import Path
from govoplan_core.core.access import (
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
)
from govoplan_core.core.module_guards import (
drop_table_retirement_provider,
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
DocumentationLink,
DocumentationTopic,
FrontendModule,
MigrationSpec,
ModuleContext,
ModuleInterfaceProvider,
ModuleManifest,
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.provider_governance import declared_module_architecture
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_quick_access.backend.db import models as quick_access_models
MODULE_ID = "quick_access"
MODULE_NAME = "Quick Access"
MODULE_VERSION = "0.1.18"
READ_SCOPE = "quick_access:profile:read"
WRITE_SCOPE = "quick_access:profile:write"
TENANT_ADMIN_SCOPE = "quick_access:profile:admin"
SYSTEM_ADMIN_SCOPE = "quick_access:system:admin"
def _permission(
scope: str,
label: str,
description: str,
*,
level: str = "tenant",
) -> PermissionDefinition:
module_id, resource, action = scope.split(":", 2)
return PermissionDefinition(
scope=scope,
label=label,
description=description,
category="Quick Access",
level=level,
module_id=module_id,
resource=resource,
action=action,
)
PERMISSIONS = (
_permission(
READ_SCOPE,
"Use Quick Access",
"Read the effective Quick Access catalogue and preferences.",
),
_permission(
WRITE_SCOPE,
"Configure personal Quick Access",
"Enable, disable, and order available Quick Access categories and tools.",
),
_permission(
TENANT_ADMIN_SCOPE,
"Manage tenant Quick Access",
"Set tenant availability, forced items, and default ordering.",
),
_permission(
SYSTEM_ADMIN_SCOPE,
"Manage system Quick Access",
"Set system-wide availability, forced items, and default ordering.",
level="system",
),
)
ROLE_TEMPLATES = (
RoleTemplate(
slug="quick_access_user",
name="Quick Access user",
description="Use and arrange the Quick Access rail.",
permissions=(READ_SCOPE, WRITE_SCOPE),
default_authenticated=True,
),
RoleTemplate(
slug="quick_access_manager",
name="Quick Access manager",
description="Manage tenant Quick Access policy and defaults.",
permissions=(READ_SCOPE, WRITE_SCOPE, TENANT_ADMIN_SCOPE),
),
RoleTemplate(
slug="quick_access_system_manager",
name="Quick Access system manager",
description="Manage system-wide Quick Access policy and defaults.",
permissions=(SYSTEM_ADMIN_SCOPE,),
level="system",
),
)
def _router(context: ModuleContext):
from govoplan_quick_access.backend.router import create_router
return create_router(context.registry)
DOCUMENTATION = (
DocumentationTopic(
id="quick-access.user",
title="Quick Access rail",
summary="Keep selected work, calendar, message, and file tools available beside the current page.",
body=(
"Open a category on the right rail to use compact tools without leaving the current task. "
"Messages combines enabled Mail, Postbox, and future chat contributions in one overlay. "
"Personal settings can reorder or hide items that remain available under system, tenant, "
"permission, and View policy. Every item retains a link to its complete owning page."
),
layer="configured",
documentation_types=("user",),
audience=("user",),
links=(
DocumentationLink(
label="Quick Access architecture",
href="govoplan-quick-access/docs/QUICK_ACCESS.md",
kind="repository",
),
),
translations={
"de": {
"title": "Schnellzugriffsleiste",
"summary": "Ausgewaehlte Werkzeuge fuer Arbeit, Kalender, Nachrichten und Dateien neben der aktuellen Seite verwenden.",
"body": (
"Eine Kategorie in der rechten Leiste oeffnet kompakte Werkzeuge, ohne die aktuelle Aufgabe zu verlassen. "
"Nachrichten fuehrt Beitraege aus Mail, Postfach und kuenftigen Chat-Modulen in einer Einblendung zusammen. "
"Persoenliche Einstellungen koennen alle durch System, Mandant, Berechtigungen und Ansicht zugelassenen Eintraege ordnen oder ausblenden."
),
}
},
metadata={
"help_contexts": [
"quick_access.rail",
"quick_access.drawer",
"quick_access.settings.personal",
]
},
),
DocumentationTopic(
id="quick-access.admin",
title="Quick Access policy",
summary="Govern which registered compact tools lower scopes may use and how they are ordered by default.",
body=(
"The catalogue follows installed module registrations. System settings constrain tenants; "
"tenant settings constrain users. An item may remain available, be blocked, or be forced. "
"Views and permissions form additional ceilings and Quick Access never grants access to domain data."
),
layer="configured",
documentation_types=("admin",),
audience=("system_admin", "tenant_admin", "module_admin"),
translations={
"de": {
"title": "Richtlinien fuer den Schnellzugriff",
"summary": "Verfuegbarkeit und Standardreihenfolge registrierter kompakter Werkzeuge steuern.",
"body": (
"Der Katalog folgt den Registrierungen installierter Module. Systemeinstellungen begrenzen Mandanten, "
"Mandanteneinstellungen begrenzen Benutzer. Ein Eintrag kann verfuegbar, gesperrt oder erzwungen sein. "
"Ansichten und Berechtigungen bilden weitere Grenzen; Schnellzugriff erteilt selbst keinen Datenzugriff."
),
}
},
metadata={
"help_contexts": [
"quick_access.admin.system",
"quick_access.admin.tenant",
"quick_access.field.availability",
"quick_access.field.order",
]
},
),
)
manifest = ModuleManifest(
id=MODULE_ID,
name=MODULE_NAME,
version=MODULE_VERSION,
dependencies=("access",),
optional_dependencies=("views", "policy"),
required_capabilities=(
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
),
provides_interfaces=(
ModuleInterfaceProvider(name="quick_access.runtime", version="1.0.0"),
ModuleInterfaceProvider(name="quick_access.preferences", version="1.0.0"),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
route_factory=_router,
frontend=FrontendModule(
module_id=MODULE_ID,
package_name="@govoplan/quick-access-webui",
view_surfaces=(
ViewSurface(
id="quick_access.rail",
module_id=MODULE_ID,
kind="quick_access",
label="Quick Access rail",
description="Optional right-side rail and overlay host.",
order=5,
required=True,
),
ViewSurface(
id="quick_access.drawer",
module_id=MODULE_ID,
kind="quick_access",
label="Quick Access drawer",
parent_id="quick_access.rail",
order=10,
required=True,
),
ViewSurface(
id="quick_access.settings.personal",
module_id=MODULE_ID,
kind="section",
label="Personal Quick Access settings",
order=20,
),
ViewSurface(
id="quick_access.admin.tenant",
module_id=MODULE_ID,
kind="section",
label="Tenant Quick Access policy",
order=30,
),
ViewSurface(
id="quick_access.admin.system",
module_id=MODULE_ID,
kind="section",
label="System Quick Access policy",
order=40,
),
),
),
migration_spec=MigrationSpec(
module_id=MODULE_ID,
metadata=Base.metadata,
script_location=str(Path(__file__).with_name("migrations") / "versions"),
retirement_supported=True,
retirement_provider=drop_table_retirement_provider(
quick_access_models.QuickAccessProfile,
label="Quick Access preferences",
),
retirement_notes=(
"Destructive retirement removes presentation preferences only; "
"contributing module data and full-page routes remain unchanged."
),
),
uninstall_guard_providers=(
persistent_table_uninstall_guard(
quick_access_models.QuickAccessProfile,
label="Quick Access preferences",
),
),
documentation=DOCUMENTATION,
architecture=declared_module_architecture(
layer="runtime_meta",
kind="presentation",
maturity="vertical_slice",
documentation_ref="docs/QUICK_ACCESS.md",
test_ref="tests/test_quick_access.py",
known_limits=(
"The first slice provides four stable categories; administrators cannot yet define additional category identities.",
),
supported_authority_modes=("native_authoritative",),
owned_concepts=(
"Quick Access profile",
"Quick Access rail",
"Quick Access category ordering",
),
non_owned_concepts=(
"task",
"calendar event",
"mail message",
"postbox message",
"file",
"authorization decision",
),
reference_packages=("product.task-focused-workspace",),
migration_docs=("docs/QUICK_ACCESS.md",),
recovery_docs=("docs/QUICK_ACCESS.md",),
security_docs=("docs/QUICK_ACCESS.md",),
operations_docs=("docs/QUICK_ACCESS.md",),
),
)
def get_manifest() -> ModuleManifest:
return manifest
__all__ = [
"MODULE_ID",
"MODULE_VERSION",
"READ_SCOPE",
"SYSTEM_ADMIN_SCOPE",
"TENANT_ADMIN_SCOPE",
"WRITE_SCOPE",
"get_manifest",
"manifest",
]
@@ -0,0 +1 @@
"""Quick Access migrations."""
@@ -0,0 +1,49 @@
"""v0.1.18 Quick Access profiles.
Revision ID: 9a4e6c2d8f10
Revises: None
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "9a4e6c2d8f10"
down_revision = None
branch_labels = None
depends_on = "4f2a9c8e7b6d"
def upgrade() -> None:
op.create_table(
"quick_access_profiles",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("scope_type", sa.String(length=20), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=True),
sa.Column("scope_id", sa.String(length=255), nullable=True),
sa.Column("scope_key", sa.String(length=340), nullable=False),
sa.Column("category_preferences", sa.JSON(), nullable=False),
sa.Column("tool_preferences", sa.JSON(), nullable=False),
sa.Column("revision", sa.Integer(), nullable=False),
sa.Column("created_by", sa.String(length=255), nullable=True),
sa.Column("updated_by", sa.String(length=255), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("scope_key", name="uq_quick_access_profile_scope"),
)
op.create_index("ix_quick_access_profiles_scope_type", "quick_access_profiles", ["scope_type"])
op.create_index("ix_quick_access_profiles_tenant_id", "quick_access_profiles", ["tenant_id"])
op.create_index("ix_quick_access_profiles_scope_id", "quick_access_profiles", ["scope_id"])
op.create_index("ix_quick_access_profiles_scope_key", "quick_access_profiles", ["scope_key"])
op.create_index(
"ix_quick_access_profiles_scope",
"quick_access_profiles",
["scope_type", "tenant_id", "scope_id"],
)
def downgrade() -> None:
op.drop_table("quick_access_profiles")
@@ -0,0 +1 @@
"""Quick Access migration revisions."""
+313
View File
@@ -0,0 +1,313 @@
from __future__ import annotations
from fastapi import APIRouter, Depends, Header, HTTPException, Response, status
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.core.concurrency import (
ConcurrencyError,
MissingPreconditionError,
RevisionConflictError,
)
from govoplan_core.core.module_entitlements import tenant_module_entitlement_state
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.db.session import get_session
from govoplan_core.tenancy.scope import Tenant
from govoplan_quick_access.backend.manifest import (
READ_SCOPE,
SYSTEM_ADMIN_SCOPE,
TENANT_ADMIN_SCOPE,
WRITE_SCOPE,
)
from govoplan_quick_access.backend.schemas import (
CatalogueResponse,
EffectiveQuickAccessResponse,
ProfileResponse,
ProfileUpdateRequest,
)
from govoplan_quick_access.backend.service import (
build_catalogue,
get_profile,
profile_response,
resolve_effective,
update_profile,
)
def create_router(registry: PlatformRegistry) -> APIRouter:
router = APIRouter(prefix="/quick-access", tags=["quick-access"])
@router.get("/catalogue", response_model=CatalogueResponse)
def api_catalogue(
include_all: bool = False,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> CatalogueResponse:
_require(principal, READ_SCOPE)
return _catalogue_for_principal(
session, registry, principal, include_all=include_all
)
@router.get("/effective", response_model=EffectiveQuickAccessResponse)
def api_effective(
include_all: bool = False,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> EffectiveQuickAccessResponse:
_require(principal, READ_SCOPE)
catalogue = _catalogue_for_principal(
session, registry, principal, include_all=include_all
)
return resolve_effective(
session,
registry=registry,
tenant_id=principal.tenant_id,
account_id=principal.account_id,
catalogue=catalogue,
)
@router.get("/profiles/me", response_model=ProfileResponse)
def api_my_profile(
response: Response,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ProfileResponse:
_require(principal, READ_SCOPE)
return _read_profile(
response,
session,
registry,
scope_type="user",
tenant_id=principal.tenant_id,
scope_id=principal.account_id,
principal=principal,
)
@router.put("/profiles/me", response_model=ProfileResponse)
def api_update_my_profile(
payload: ProfileUpdateRequest,
response: Response,
if_match: str | None = Header(default=None, alias="If-Match"),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ProfileResponse:
_require(principal, WRITE_SCOPE)
return _write_profile(
payload,
response,
session,
registry,
principal,
if_match=if_match,
scope_type="user",
tenant_id=principal.tenant_id,
scope_id=principal.account_id,
)
@router.get("/profiles/tenant", response_model=ProfileResponse)
def api_tenant_profile(
response: Response,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ProfileResponse:
_require(principal, TENANT_ADMIN_SCOPE)
return _read_profile(
response,
session,
registry,
scope_type="tenant",
tenant_id=principal.tenant_id,
scope_id=principal.tenant_id,
principal=principal,
)
@router.put("/profiles/tenant", response_model=ProfileResponse)
def api_update_tenant_profile(
payload: ProfileUpdateRequest,
response: Response,
if_match: str | None = Header(default=None, alias="If-Match"),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ProfileResponse:
_require(principal, TENANT_ADMIN_SCOPE)
return _write_profile(
payload,
response,
session,
registry,
principal,
if_match=if_match,
scope_type="tenant",
tenant_id=principal.tenant_id,
scope_id=principal.tenant_id,
)
@router.get("/profiles/system", response_model=ProfileResponse)
def api_system_profile(
response: Response,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ProfileResponse:
_require(principal, SYSTEM_ADMIN_SCOPE)
return _read_profile(
response,
session,
registry,
scope_type="system",
tenant_id=None,
scope_id=None,
principal=principal,
)
@router.put("/profiles/system", response_model=ProfileResponse)
def api_update_system_profile(
payload: ProfileUpdateRequest,
response: Response,
if_match: str | None = Header(default=None, alias="If-Match"),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ProfileResponse:
_require(principal, SYSTEM_ADMIN_SCOPE)
return _write_profile(
payload,
response,
session,
registry,
principal,
if_match=if_match,
scope_type="system",
tenant_id=None,
scope_id=None,
)
return router
def _read_profile(
response: Response,
session: Session,
registry: PlatformRegistry,
*,
scope_type: str,
tenant_id: str | None,
scope_id: str | None,
principal: ApiPrincipal,
) -> ProfileResponse:
catalogue = _catalogue_for_principal(
session,
registry,
principal,
include_all=scope_type != "user",
)
result = profile_response(
get_profile(
session,
scope_type=scope_type,
tenant_id=tenant_id,
scope_id=scope_id,
),
scope_type=scope_type,
tenant_id=tenant_id,
scope_id=scope_id,
catalogue=catalogue,
)
response.headers["ETag"] = result.etag
return result
def _write_profile(
payload: ProfileUpdateRequest,
response: Response,
session: Session,
registry: PlatformRegistry,
principal: ApiPrincipal,
*,
if_match: str | None,
scope_type: str,
tenant_id: str | None,
scope_id: str | None,
) -> ProfileResponse:
try:
row = update_profile(
session,
scope_type=scope_type,
tenant_id=tenant_id,
scope_id=scope_id,
actor_id=principal.account_id,
payload=payload,
if_match=if_match,
)
session.commit()
except MissingPreconditionError as exc:
session.rollback()
raise HTTPException(status_code=428, detail=exc.as_dict()) from exc
except RevisionConflictError as exc:
session.rollback()
raise HTTPException(status_code=409, detail=exc.as_dict()) from exc
except ConcurrencyError as exc:
session.rollback()
raise HTTPException(status_code=412, detail=str(exc)) from exc
except ValueError as exc:
session.rollback()
raise HTTPException(status_code=400, detail=str(exc)) from exc
result = profile_response(
row,
scope_type=scope_type,
tenant_id=tenant_id,
scope_id=scope_id,
catalogue=_catalogue_for_principal(
session,
registry,
principal,
include_all=scope_type != "user",
),
)
response.headers["ETag"] = result.etag
return result
def _require(principal: ApiPrincipal, scope: str) -> None:
if not has_scope(principal, scope):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Missing scope: {scope}",
)
def _catalogue_for_principal(
session: Session,
registry: PlatformRegistry,
principal: ApiPrincipal,
*,
include_all: bool,
) -> CatalogueResponse:
system_admin = has_scope(principal, SYSTEM_ADMIN_SCOPE)
tenant_admin = has_scope(principal, TENANT_ADMIN_SCOPE)
if include_all and not (system_admin or tenant_admin):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Complete Quick Access catalogue requires administration permission.",
)
if include_all and system_admin:
return build_catalogue(registry)
tenant = session.get(Tenant, principal.tenant_id)
if tenant is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="The active tenant is unavailable.",
)
manifests = {manifest.id: manifest for manifest in registry.manifests()}
entitlement = tenant_module_entitlement_state(
tenant.settings or {},
manifests,
runtime_active_modules=manifests,
)
return build_catalogue(
registry,
permission_checker=None if include_all else principal.has,
allowed_module_ids=entitlement.effective_modules,
)
__all__ = ["create_router"]
@@ -0,0 +1,89 @@
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class PreferenceEntry(BaseModel):
model_config = ConfigDict(extra="forbid")
enabled: bool | None = None
forced: bool = False
order: int | None = Field(default=None, ge=0, le=100_000)
class ProfileUpdateRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
base_revision: int = Field(ge=1)
category_preferences: dict[str, PreferenceEntry] = Field(default_factory=dict)
tool_preferences: dict[str, PreferenceEntry] = Field(default_factory=dict)
class ProfileResponse(BaseModel):
scope_type: Literal["system", "tenant", "user"]
tenant_id: str | None = None
scope_id: str | None = None
revision: int
etag: str
category_preferences: dict[str, PreferenceEntry]
tool_preferences: dict[str, PreferenceEntry]
stale_category_ids: list[str] = Field(default_factory=list)
stale_tool_ids: list[str] = Field(default_factory=list)
class CatalogueCategoryResponse(BaseModel):
id: str
label: str
description: str
icon: str
order: int
class CatalogueToolResponse(BaseModel):
id: str
module_id: str
category_id: str
label: str
description: str | None = None
icon: str
surface_id: str
full_page_path: str | None = None
required_all: list[str]
required_any: list[str]
order: int
default_enabled: bool
modes: list[str]
class CatalogueResponse(BaseModel):
categories: list[CatalogueCategoryResponse]
tools: list[CatalogueToolResponse]
class EffectiveToolResponse(CatalogueToolResponse):
enabled: bool
forced: bool
locked_by: str | None = None
class EffectiveCategoryResponse(CatalogueCategoryResponse):
enabled: bool
forced: bool
locked_by: str | None = None
tools: list[EffectiveToolResponse]
class EffectiveQuickAccessResponse(BaseModel):
categories: list[EffectiveCategoryResponse]
diagnostics: list[str] = Field(default_factory=list)
__all__ = [
"CatalogueResponse",
"EffectiveQuickAccessResponse",
"PreferenceEntry",
"ProfileResponse",
"ProfileUpdateRequest",
]
@@ -0,0 +1,381 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable, Iterable, Mapping
from sqlalchemy.orm import Session
from govoplan_core.core.concurrency import (
RevisionConflictError,
assert_revision_precondition,
strong_resource_etag,
)
from govoplan_core.core.registry import PlatformRegistry
from govoplan_quick_access.backend.db.models import QuickAccessProfile
from govoplan_quick_access.backend.schemas import (
CatalogueCategoryResponse,
CatalogueResponse,
CatalogueToolResponse,
EffectiveCategoryResponse,
EffectiveQuickAccessResponse,
EffectiveToolResponse,
PreferenceEntry,
ProfileResponse,
ProfileUpdateRequest,
)
BASE_CATEGORIES = (
CatalogueCategoryResponse(
id="work",
label="i18n:govoplan-quick-access.category.work",
description="i18n:govoplan-quick-access.category.work_description",
icon="list-checks",
order=10,
),
CatalogueCategoryResponse(
id="calendar",
label="i18n:govoplan-quick-access.category.calendar",
description="i18n:govoplan-quick-access.category.calendar_description",
icon="calendar",
order=20,
),
CatalogueCategoryResponse(
id="messages",
label="i18n:govoplan-quick-access.category.messages",
description="i18n:govoplan-quick-access.category.messages_description",
icon="messages-square",
order=30,
),
CatalogueCategoryResponse(
id="files",
label="i18n:govoplan-quick-access.category.files",
description="i18n:govoplan-quick-access.category.files_description",
icon="files",
order=40,
),
)
def scope_key(scope_type: str, tenant_id: str | None, scope_id: str | None) -> str:
if scope_type == "system":
return "system:*"
if scope_type == "tenant" and tenant_id:
return f"tenant:{tenant_id}"
if scope_type == "user" and tenant_id and scope_id:
return f"user:{tenant_id}:{scope_id}"
raise ValueError("Invalid Quick Access profile scope")
def build_catalogue(
registry: PlatformRegistry,
*,
permission_checker: Callable[[str], bool] | None = None,
allowed_module_ids: Iterable[str] | None = None,
) -> CatalogueResponse:
allowed_modules = (
None if allowed_module_ids is None else frozenset(allowed_module_ids)
)
tools: list[CatalogueToolResponse] = []
for manifest in registry.manifests():
if allowed_modules is not None and manifest.id not in allowed_modules:
continue
frontend = manifest.frontend
if frontend is None:
continue
for tool in frontend.quick_access_tools:
if permission_checker is not None:
if tool.required_all and not all(permission_checker(scope) for scope in tool.required_all):
continue
if tool.required_any and not any(permission_checker(scope) for scope in tool.required_any):
continue
tools.append(
CatalogueToolResponse(
id=tool.id,
module_id=tool.module_id,
category_id=tool.category_id,
label=tool.label,
description=tool.description,
icon=tool.icon,
surface_id=tool.surface_id,
full_page_path=tool.full_page_path,
required_all=list(tool.required_all),
required_any=list(tool.required_any),
order=tool.order,
default_enabled=tool.default_enabled,
modes=list(tool.modes),
)
)
tools.sort(key=lambda item: (item.category_id, item.order, item.id))
return CatalogueResponse(categories=list(BASE_CATEGORIES), tools=tools)
def get_profile(
session: Session,
*,
scope_type: str,
tenant_id: str | None,
scope_id: str | None,
) -> QuickAccessProfile | None:
key = scope_key(scope_type, tenant_id, scope_id)
return session.query(QuickAccessProfile).filter(QuickAccessProfile.scope_key == key).one_or_none()
def profile_response(
row: QuickAccessProfile | None,
*,
scope_type: str,
tenant_id: str | None,
scope_id: str | None,
catalogue: CatalogueResponse,
) -> ProfileResponse:
revision = row.revision if row is not None else 1
resource_id = scope_key(scope_type, tenant_id, scope_id)
category_preferences = _preference_map(row.category_preferences if row else {})
tool_preferences = _preference_map(row.tool_preferences if row else {})
category_ids = {item.id for item in catalogue.categories}
tool_ids = {item.id for item in catalogue.tools}
return ProfileResponse(
scope_type=scope_type,
tenant_id=tenant_id,
scope_id=scope_id,
revision=revision,
etag=strong_resource_etag("quick_access_profile", resource_id, revision),
category_preferences=category_preferences,
tool_preferences=tool_preferences,
stale_category_ids=sorted(set(category_preferences) - category_ids),
stale_tool_ids=sorted(set(tool_preferences) - tool_ids),
)
def update_profile(
session: Session,
*,
scope_type: str,
tenant_id: str | None,
scope_id: str | None,
actor_id: str,
payload: ProfileUpdateRequest,
if_match: str | None,
) -> QuickAccessProfile:
key = scope_key(scope_type, tenant_id, scope_id)
assert_revision_precondition(
if_match,
resource_type="quick_access_profile",
resource_id=key,
submitted_base_revision=payload.base_revision,
)
row = get_profile(
session,
scope_type=scope_type,
tenant_id=tenant_id,
scope_id=scope_id,
)
current_revision = row.revision if row is not None else 1
if current_revision != payload.base_revision:
raise RevisionConflictError(
resource_type="quick_access_profile",
resource_id=key,
current_revision=current_revision,
submitted_base_revision=payload.base_revision,
refresh_path=_profile_path(scope_type),
current_etag=strong_resource_etag(
"quick_access_profile", key, current_revision
),
)
if scope_type == "user" and any(
entry.forced for entry in (
*payload.category_preferences.values(),
*payload.tool_preferences.values(),
)
):
raise ValueError("Personal Quick Access preferences cannot force items")
category_preferences = _serialized_preferences(payload.category_preferences)
tool_preferences = _serialized_preferences(payload.tool_preferences)
if row is None:
row = QuickAccessProfile(
scope_type=scope_type,
tenant_id=tenant_id,
scope_id=scope_id,
scope_key=key,
revision=2,
created_by=actor_id,
updated_by=actor_id,
category_preferences=category_preferences,
tool_preferences=tool_preferences,
)
session.add(row)
else:
row.revision += 1
row.updated_by = actor_id
row.category_preferences = category_preferences
row.tool_preferences = tool_preferences
session.flush()
return row
def resolve_effective(
session: Session,
*,
registry: PlatformRegistry,
tenant_id: str,
account_id: str,
permission_checker: Callable[[str], bool] | None = None,
allowed_module_ids: Iterable[str] | None = None,
catalogue: CatalogueResponse | None = None,
) -> EffectiveQuickAccessResponse:
if catalogue is None:
catalogue = build_catalogue(
registry,
permission_checker=permission_checker,
allowed_module_ids=allowed_module_ids,
)
profiles = (
("system", get_profile(session, scope_type="system", tenant_id=None, scope_id=None)),
("tenant", get_profile(session, scope_type="tenant", tenant_id=tenant_id, scope_id=tenant_id)),
("user", get_profile(session, scope_type="user", tenant_id=tenant_id, scope_id=account_id)),
)
diagnostics: list[str] = []
category_states: dict[str, _EffectiveState] = {}
for category in catalogue.categories:
state = _EffectiveState(enabled=True, order=category.order)
for source, profile in profiles:
entry = _profile_entry(profile, "category_preferences", category.id, diagnostics)
state.apply(entry, source=source)
category_states[category.id] = state
tools_by_category: dict[str, list[EffectiveToolResponse]] = {
category.id: [] for category in catalogue.categories
}
for tool in catalogue.tools:
state = _EffectiveState(enabled=tool.default_enabled, order=tool.order)
for source, profile in profiles:
entry = _profile_entry(profile, "tool_preferences", tool.id, diagnostics)
state.apply(entry, source=source)
category_state = category_states.get(tool.category_id)
enabled = state.enabled and bool(category_state and category_state.enabled)
tool_payload = tool.model_dump()
tool_payload["order"] = state.order
tools_by_category.setdefault(tool.category_id, []).append(
EffectiveToolResponse(
**tool_payload,
enabled=enabled,
forced=state.forced,
locked_by=state.locked_by,
)
)
categories: list[EffectiveCategoryResponse] = []
for category in catalogue.categories:
state = category_states[category.id]
tools = sorted(
tools_by_category.get(category.id, []),
key=lambda item: (item.order, item.id),
)
enabled = state.enabled and any(tool.enabled for tool in tools)
category_payload = category.model_dump()
category_payload["order"] = state.order
categories.append(
EffectiveCategoryResponse(
**category_payload,
enabled=enabled,
forced=state.forced,
locked_by=state.locked_by,
tools=tools,
)
)
categories.sort(key=lambda item: (item.order, item.id))
return EffectiveQuickAccessResponse(
categories=categories,
diagnostics=list(dict.fromkeys(diagnostics)),
)
@dataclass(slots=True)
class _EffectiveState:
enabled: bool
order: int
forced: bool = False
locked_by: str | None = None
def apply(self, entry: PreferenceEntry | None, *, source: str) -> None:
if entry is None:
return
if entry.order is not None:
self.order = entry.order
if self.locked_by is not None:
return
if entry.enabled is not None:
self.enabled = entry.enabled
if source != "user" and entry.enabled is False:
self.forced = False
self.locked_by = source
elif source != "user" and entry.forced:
self.enabled = True
self.forced = True
self.locked_by = source
def _profile_entry(
profile: QuickAccessProfile | None,
field: str,
item_id: str,
diagnostics: list[str],
) -> PreferenceEntry | None:
if profile is None:
return None
raw_map = getattr(profile, field, {})
if not isinstance(raw_map, Mapping):
diagnostics.append(f"Ignored malformed {field} in {profile.scope_key}.")
return None
raw = raw_map.get(item_id)
if raw is None:
return None
try:
return PreferenceEntry.model_validate(raw)
except Exception:
diagnostics.append(f"Ignored malformed preference for {item_id} in {profile.scope_key}.")
return None
def _preference_map(raw: object) -> dict[str, PreferenceEntry]:
if not isinstance(raw, Mapping):
return {}
result: dict[str, PreferenceEntry] = {}
for item_id, value in raw.items():
try:
result[str(item_id)] = PreferenceEntry.model_validate(value)
except Exception:
continue
return result
def _serialized_preferences(
preferences: Mapping[str, PreferenceEntry],
) -> dict[str, dict[str, object]]:
if len(preferences) > 500:
raise ValueError("Quick Access profiles may contain at most 500 preferences")
return {
str(item_id): entry.model_dump(mode="json", exclude_none=True)
for item_id, entry in preferences.items()
if str(item_id).strip()
}
def _profile_path(scope_type: str) -> str:
return {
"system": "/api/v1/quick-access/profiles/system",
"tenant": "/api/v1/quick-access/profiles/tenant",
"user": "/api/v1/quick-access/profiles/me",
}[scope_type]
__all__ = [
"BASE_CATEGORIES",
"build_catalogue",
"get_profile",
"profile_response",
"resolve_effective",
"scope_key",
"update_profile",
]
+1
View File
@@ -0,0 +1 @@