1593 lines
48 KiB
Python
1593 lines
48 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Literal
|
|
|
|
from sqlalchemy import and_, or_
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from govoplan_core.core.views import (
|
|
VIEW_SURFACE_CONTRACT_VERSION,
|
|
EffectiveView,
|
|
ViewSurface,
|
|
)
|
|
from govoplan_core.db.base import utcnow
|
|
from govoplan_views.backend.db.models import (
|
|
ViewAssignment,
|
|
ViewDefinition,
|
|
ViewPreference,
|
|
ViewRevision,
|
|
)
|
|
|
|
|
|
ASSIGNMENT_MODES = frozenset({"available", "default", "required"})
|
|
ASSIGNMENT_SCOPES = frozenset({"system", "tenant", "group", "user"})
|
|
DEFINITION_SCOPES = frozenset({"system", "tenant", "group", "user"})
|
|
MAX_EFFECTIVE_VIEW_ASSIGNMENTS = 1000
|
|
MAX_EFFECTIVE_VIEW_GROUPS = 500
|
|
LOCKOUT_BASE_SURFACE_IDS = (
|
|
"access.module",
|
|
"access.nav.admin",
|
|
"access.route.admin",
|
|
"views.module",
|
|
"views.selector",
|
|
)
|
|
LOCKOUT_ADMIN_SURFACE_IDS = {
|
|
"system": "views.admin.system",
|
|
"tenant": "views.admin.tenant",
|
|
"group": "views.admin.tenant",
|
|
"user": "views.admin.tenant",
|
|
}
|
|
_KEY_RE = re.compile(r"[^a-z0-9]+")
|
|
|
|
|
|
class ViewsError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class ViewsNotFoundError(ViewsError):
|
|
pass
|
|
|
|
|
|
class ViewsConflictError(ViewsError):
|
|
pass
|
|
|
|
|
|
class ViewsValidationError(ViewsError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class EffectiveViewOption:
|
|
id: str
|
|
name: str
|
|
description: str | None
|
|
revision_id: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ViewDiagnostic:
|
|
severity: Literal["warning", "error"]
|
|
code: str
|
|
message: str
|
|
surface_ids: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class EffectiveViewState:
|
|
effective: EffectiveView
|
|
available_views: tuple[EffectiveViewOption, ...]
|
|
diagnostics: tuple[ViewDiagnostic, ...] = ()
|
|
invalidation_token: str | None = None
|
|
|
|
|
|
ResolvedAssignment = tuple[ViewAssignment, ViewRevision]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ViewSelection:
|
|
selected: ResolvedAssignment | None
|
|
locked: bool
|
|
provenance: tuple[dict[str, object], ...]
|
|
diagnostics: tuple[ViewDiagnostic, ...] = ()
|
|
|
|
|
|
def _now() -> datetime:
|
|
return utcnow()
|
|
|
|
|
|
def _scope_values(
|
|
scope_type: str,
|
|
*,
|
|
tenant_id: str,
|
|
scope_id: str | None = None,
|
|
) -> tuple[str | None, str | None, str]:
|
|
if scope_type == "system":
|
|
if scope_id:
|
|
raise ViewsValidationError(
|
|
"System View definitions cannot declare a target id"
|
|
)
|
|
return None, None, "system"
|
|
if scope_type == "tenant":
|
|
if scope_id and scope_id != tenant_id:
|
|
raise ViewsValidationError(
|
|
"Tenant View definitions must target the active tenant"
|
|
)
|
|
return tenant_id, tenant_id, f"tenant:{tenant_id}"
|
|
if scope_type in {"group", "user"}:
|
|
target_id = (scope_id or "").strip()
|
|
if not target_id:
|
|
raise ViewsValidationError(
|
|
f"{scope_type.title()} View definitions require a target id"
|
|
)
|
|
return tenant_id, target_id, f"{scope_type}:{tenant_id}:{target_id}"
|
|
raise ViewsValidationError(f"Unsupported View definition scope: {scope_type}")
|
|
|
|
|
|
def _assignment_target(
|
|
scope_type: str,
|
|
scope_id: str | None,
|
|
*,
|
|
tenant_id: str,
|
|
) -> tuple[str | None, str | None, str]:
|
|
normalized_scope = scope_type.strip().lower()
|
|
if normalized_scope not in ASSIGNMENT_SCOPES:
|
|
raise ViewsValidationError(f"Unsupported View assignment scope: {scope_type}")
|
|
if normalized_scope == "system":
|
|
if scope_id:
|
|
raise ViewsValidationError(
|
|
"System View assignments cannot declare a target id"
|
|
)
|
|
return None, None, "system"
|
|
if normalized_scope == "tenant":
|
|
if scope_id and scope_id != tenant_id:
|
|
raise ViewsValidationError(
|
|
"Tenant View assignments must target the active tenant"
|
|
)
|
|
return tenant_id, tenant_id, f"tenant:{tenant_id}"
|
|
target_id = (scope_id or "").strip()
|
|
if not target_id:
|
|
raise ViewsValidationError(
|
|
f"{normalized_scope.title()} View assignments require a target id"
|
|
)
|
|
return tenant_id, target_id, f"{normalized_scope}:{tenant_id}:{target_id}"
|
|
|
|
|
|
def _definition_key(value: str | None, name: str) -> str:
|
|
source = (value or name).strip().lower()
|
|
normalized = _KEY_RE.sub("-", source).strip("-")
|
|
if not normalized:
|
|
raise ViewsValidationError("View key must contain letters or numbers")
|
|
return normalized[:120]
|
|
|
|
|
|
def _definition_query_for_actor(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
include_system: bool,
|
|
):
|
|
query = session.query(ViewDefinition).filter(ViewDefinition.deleted_at.is_(None))
|
|
if include_system:
|
|
return query.filter(
|
|
or_(
|
|
ViewDefinition.tenant_id == tenant_id,
|
|
ViewDefinition.tenant_id.is_(None),
|
|
)
|
|
)
|
|
return query.filter(ViewDefinition.tenant_id == tenant_id)
|
|
|
|
|
|
def list_definitions(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
scope_type: str,
|
|
scope_id: str | None = None,
|
|
include_inherited: bool = True,
|
|
) -> list[ViewDefinition]:
|
|
if scope_type not in DEFINITION_SCOPES:
|
|
raise ViewsValidationError(f"Unsupported View definition scope: {scope_type}")
|
|
if scope_type == "system":
|
|
query = session.query(ViewDefinition).filter(
|
|
ViewDefinition.tenant_id.is_(None),
|
|
ViewDefinition.scope_type == "system",
|
|
ViewDefinition.deleted_at.is_(None),
|
|
)
|
|
elif scope_type == "tenant":
|
|
if not include_inherited:
|
|
query = session.query(ViewDefinition).filter(
|
|
ViewDefinition.tenant_id == tenant_id,
|
|
ViewDefinition.scope_type == "tenant",
|
|
ViewDefinition.deleted_at.is_(None),
|
|
)
|
|
else:
|
|
query = _definition_query_for_actor(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
include_system=True,
|
|
)
|
|
else:
|
|
target_id = (scope_id or "").strip()
|
|
if not target_id:
|
|
raise ViewsValidationError(
|
|
f"{scope_type.title()} View definition listings require a target id"
|
|
)
|
|
exact_target = (
|
|
(ViewDefinition.tenant_id == tenant_id)
|
|
& (ViewDefinition.scope_type == scope_type)
|
|
& (ViewDefinition.scope_id == target_id)
|
|
)
|
|
query = session.query(ViewDefinition).filter(
|
|
ViewDefinition.deleted_at.is_(None)
|
|
)
|
|
if include_inherited:
|
|
query = query.filter(
|
|
or_(
|
|
exact_target,
|
|
(ViewDefinition.tenant_id == tenant_id)
|
|
& (ViewDefinition.scope_type == "tenant"),
|
|
(ViewDefinition.tenant_id.is_(None))
|
|
& (ViewDefinition.scope_type == "system"),
|
|
)
|
|
)
|
|
else:
|
|
query = query.filter(exact_target)
|
|
return query.order_by(
|
|
ViewDefinition.scope_type.asc(),
|
|
ViewDefinition.name.asc(),
|
|
ViewDefinition.id.asc(),
|
|
).all()
|
|
|
|
|
|
def get_definition(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
definition_id: str,
|
|
include_system: bool = True,
|
|
) -> ViewDefinition:
|
|
definition = (
|
|
_definition_query_for_actor(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
include_system=include_system,
|
|
)
|
|
.filter(ViewDefinition.id == definition_id)
|
|
.first()
|
|
)
|
|
if definition is None:
|
|
raise ViewsNotFoundError("View definition not found")
|
|
return definition
|
|
|
|
|
|
def get_revision(
|
|
session: Session,
|
|
*,
|
|
definition_id: str,
|
|
revision_id: str | None = None,
|
|
revision: int | None = None,
|
|
) -> ViewRevision:
|
|
query = session.query(ViewRevision).filter(
|
|
ViewRevision.definition_id == definition_id
|
|
)
|
|
if revision_id is not None:
|
|
query = query.filter(ViewRevision.id == revision_id)
|
|
elif revision is not None:
|
|
query = query.filter(ViewRevision.revision == revision)
|
|
else:
|
|
query = query.order_by(ViewRevision.revision.desc())
|
|
result = query.first()
|
|
if result is None:
|
|
raise ViewsNotFoundError("View revision not found")
|
|
return result
|
|
|
|
|
|
def definition_revisions(
|
|
session: Session,
|
|
*,
|
|
definition_id: str,
|
|
) -> list[ViewRevision]:
|
|
return (
|
|
session.query(ViewRevision)
|
|
.filter(ViewRevision.definition_id == definition_id)
|
|
.order_by(ViewRevision.revision.desc())
|
|
.all()
|
|
)
|
|
|
|
|
|
def normalize_visible_surface_ids(
|
|
values: Iterable[str],
|
|
*,
|
|
catalogue: Iterable[ViewSurface],
|
|
) -> list[str]:
|
|
surfaces = tuple(catalogue)
|
|
by_id = {surface.id: surface for surface in surfaces}
|
|
selected = {str(value).strip() for value in values if str(value).strip()}
|
|
unknown = sorted(selected - set(by_id))
|
|
if unknown:
|
|
preview = ", ".join(unknown[:5])
|
|
suffix = "" if len(unknown) <= 5 else f" and {len(unknown) - 5} more"
|
|
raise ViewsValidationError(f"Unknown View surfaces: {preview}{suffix}")
|
|
|
|
selected.update(surface.id for surface in surfaces if surface.required)
|
|
pending = list(selected)
|
|
while pending:
|
|
surface_id = pending.pop()
|
|
surface = by_id[surface_id]
|
|
if surface.parent_id and surface.parent_id not in selected:
|
|
if surface.parent_id not in by_id:
|
|
raise ViewsValidationError(
|
|
f"View surface {surface.id} has unavailable parent {surface.parent_id}"
|
|
)
|
|
selected.add(surface.parent_id)
|
|
pending.append(surface.parent_id)
|
|
|
|
selected_surfaces = [by_id[surface_id] for surface_id in selected]
|
|
if not any(surface.kind == "navigation" for surface in selected_surfaces):
|
|
raise ViewsValidationError("A View must keep at least one navigation entry")
|
|
if not any(surface.kind == "route" for surface in selected_surfaces):
|
|
raise ViewsValidationError("A View must keep at least one route")
|
|
|
|
order = {surface.id: index for index, surface in enumerate(surfaces)}
|
|
return sorted(
|
|
selected, key=lambda surface_id: (order.get(surface_id, 10_000), surface_id)
|
|
)
|
|
|
|
|
|
def _revision_hash(surface_ids: list[str]) -> str:
|
|
payload = {
|
|
"surface_contract_version": VIEW_SURFACE_CONTRACT_VERSION,
|
|
"visible_surface_ids": surface_ids,
|
|
}
|
|
return hashlib.sha256(
|
|
json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def create_definition(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
scope_type: str,
|
|
scope_id: str | None = None,
|
|
definition_key: str | None,
|
|
name: str,
|
|
description: str | None,
|
|
visible_surface_ids: Iterable[str],
|
|
catalogue: Iterable[ViewSurface],
|
|
actor_id: str | None,
|
|
) -> ViewDefinition:
|
|
if scope_type not in DEFINITION_SCOPES:
|
|
raise ViewsValidationError(f"Unsupported View definition scope: {scope_type}")
|
|
row_tenant_id, scope_id, scope_key = _scope_values(
|
|
scope_type,
|
|
tenant_id=tenant_id,
|
|
scope_id=scope_id,
|
|
)
|
|
clean_name = name.strip()
|
|
if not clean_name:
|
|
raise ViewsValidationError("View name is required")
|
|
clean_key = _definition_key(definition_key, clean_name)
|
|
duplicate = (
|
|
session.query(ViewDefinition.id)
|
|
.filter(
|
|
ViewDefinition.scope_key == scope_key,
|
|
ViewDefinition.definition_key == clean_key,
|
|
ViewDefinition.deleted_at.is_(None),
|
|
)
|
|
.first()
|
|
)
|
|
if duplicate is not None:
|
|
raise ViewsConflictError(
|
|
f"A View with key {clean_key!r} already exists in this scope"
|
|
)
|
|
normalized_surfaces = normalize_visible_surface_ids(
|
|
visible_surface_ids,
|
|
catalogue=catalogue,
|
|
)
|
|
definition = ViewDefinition(
|
|
tenant_id=row_tenant_id,
|
|
scope_type=scope_type,
|
|
scope_id=scope_id,
|
|
scope_key=scope_key,
|
|
definition_key=clean_key,
|
|
name=clean_name,
|
|
description=(description or "").strip() or None,
|
|
status="draft",
|
|
current_revision=1,
|
|
created_by=actor_id,
|
|
updated_by=actor_id,
|
|
)
|
|
session.add(definition)
|
|
session.flush()
|
|
revision = ViewRevision(
|
|
tenant_id=row_tenant_id,
|
|
definition_id=definition.id,
|
|
revision=1,
|
|
surface_contract_version=VIEW_SURFACE_CONTRACT_VERSION,
|
|
visible_surface_ids=normalized_surfaces,
|
|
content_hash=_revision_hash(normalized_surfaces),
|
|
created_by=actor_id,
|
|
)
|
|
session.add(revision)
|
|
session.flush()
|
|
return definition
|
|
|
|
|
|
def update_definition(
|
|
session: Session,
|
|
definition: ViewDefinition,
|
|
*,
|
|
name: str | None,
|
|
description: str | None,
|
|
actor_id: str | None,
|
|
fields_set: set[str],
|
|
) -> ViewDefinition:
|
|
if definition.status == "archived":
|
|
raise ViewsConflictError("Archived Views cannot be edited")
|
|
if "name" in fields_set:
|
|
clean_name = (name or "").strip()
|
|
if not clean_name:
|
|
raise ViewsValidationError("View name is required")
|
|
definition.name = clean_name
|
|
if "description" in fields_set:
|
|
definition.description = (description or "").strip() or None
|
|
definition.updated_by = actor_id
|
|
definition.updated_at = _now()
|
|
session.flush()
|
|
return definition
|
|
|
|
|
|
def create_revision(
|
|
session: Session,
|
|
definition: ViewDefinition,
|
|
*,
|
|
visible_surface_ids: Iterable[str],
|
|
catalogue: Iterable[ViewSurface],
|
|
actor_id: str | None,
|
|
) -> ViewRevision:
|
|
if definition.status == "archived":
|
|
raise ViewsConflictError("Archived Views cannot be revised")
|
|
normalized_surfaces = normalize_visible_surface_ids(
|
|
visible_surface_ids,
|
|
catalogue=catalogue,
|
|
)
|
|
content_hash = _revision_hash(normalized_surfaces)
|
|
latest = get_revision(session, definition_id=definition.id)
|
|
if latest.content_hash == content_hash:
|
|
return latest
|
|
next_number = definition.current_revision + 1
|
|
revision = ViewRevision(
|
|
tenant_id=definition.tenant_id,
|
|
definition_id=definition.id,
|
|
revision=next_number,
|
|
surface_contract_version=VIEW_SURFACE_CONTRACT_VERSION,
|
|
visible_surface_ids=normalized_surfaces,
|
|
content_hash=content_hash,
|
|
created_by=actor_id,
|
|
)
|
|
session.add(revision)
|
|
definition.current_revision = next_number
|
|
definition.updated_by = actor_id
|
|
definition.updated_at = _now()
|
|
session.flush()
|
|
return revision
|
|
|
|
|
|
def lockout_required_surface_ids(
|
|
catalogue: Iterable[ViewSurface],
|
|
*,
|
|
assignment_scope_type: str,
|
|
) -> frozenset[str]:
|
|
known = {surface.id for surface in catalogue}
|
|
required = {surface.id for surface in catalogue if surface.required}
|
|
admin_surface = LOCKOUT_ADMIN_SURFACE_IDS.get(assignment_scope_type)
|
|
escape_ids = set(LOCKOUT_BASE_SURFACE_IDS)
|
|
if admin_surface:
|
|
escape_ids.add(admin_surface)
|
|
missing_contract = {
|
|
"access.nav.admin",
|
|
"access.route.admin",
|
|
admin_surface,
|
|
} - known
|
|
missing_contract.discard(None)
|
|
if missing_contract:
|
|
raise ViewsValidationError(
|
|
"Required View assignments are unavailable because the active module "
|
|
"graph does not expose all administration escape surfaces: "
|
|
+ ", ".join(sorted(missing_contract))
|
|
)
|
|
return frozenset(required | (escape_ids & known))
|
|
|
|
|
|
def validate_required_assignment_revision(
|
|
revision: ViewRevision,
|
|
*,
|
|
assignment_scope_type: str,
|
|
catalogue: Iterable[ViewSurface],
|
|
) -> None:
|
|
required_ids = lockout_required_surface_ids(
|
|
catalogue,
|
|
assignment_scope_type=assignment_scope_type,
|
|
)
|
|
missing = sorted(required_ids - set(revision.visible_surface_ids))
|
|
if missing:
|
|
raise ViewsValidationError(
|
|
"A required View must retain its selector and administration escape "
|
|
f"surfaces: {', '.join(missing)}"
|
|
)
|
|
|
|
|
|
def publish_revision(
|
|
session: Session,
|
|
definition: ViewDefinition,
|
|
revision: ViewRevision,
|
|
*,
|
|
catalogue: Iterable[ViewSurface],
|
|
actor_id: str | None,
|
|
) -> ViewDefinition:
|
|
if definition.status == "archived":
|
|
raise ViewsConflictError("Archived Views cannot be published")
|
|
tracking_required_assignments = (
|
|
session.query(ViewAssignment)
|
|
.filter(
|
|
ViewAssignment.definition_id == definition.id,
|
|
ViewAssignment.mode == "required",
|
|
ViewAssignment.is_active.is_(True),
|
|
ViewAssignment.revision_id.is_(None),
|
|
)
|
|
.all()
|
|
)
|
|
for assignment in tracking_required_assignments:
|
|
validate_required_assignment_revision(
|
|
revision,
|
|
assignment_scope_type=assignment.scope_type,
|
|
catalogue=catalogue,
|
|
)
|
|
definition.published_revision_id = revision.id
|
|
definition.status = "published"
|
|
definition.updated_by = actor_id
|
|
definition.updated_at = _now()
|
|
session.flush()
|
|
ensure_owner_available_assignment(
|
|
session,
|
|
definition,
|
|
catalogue=catalogue,
|
|
actor_id=actor_id,
|
|
)
|
|
return definition
|
|
|
|
|
|
def ensure_owner_available_assignment(
|
|
session: Session,
|
|
definition: ViewDefinition,
|
|
*,
|
|
catalogue: Iterable[ViewSurface],
|
|
actor_id: str | None,
|
|
) -> ViewAssignment | None:
|
|
if definition.scope_type not in {"group", "user"}:
|
|
return None
|
|
if not definition.tenant_id or not definition.scope_id:
|
|
raise ViewsValidationError(
|
|
"Owned View definitions require tenant and target identifiers"
|
|
)
|
|
target_key = f"{definition.scope_type}:{definition.tenant_id}:{definition.scope_id}"
|
|
assignment = (
|
|
session.query(ViewAssignment)
|
|
.filter(
|
|
ViewAssignment.target_key == target_key,
|
|
ViewAssignment.definition_id == definition.id,
|
|
ViewAssignment.mode == "available",
|
|
)
|
|
.first()
|
|
)
|
|
if assignment is None:
|
|
return create_assignment(
|
|
session,
|
|
tenant_id=definition.tenant_id,
|
|
scope_type=definition.scope_type,
|
|
scope_id=definition.scope_id,
|
|
definition=definition,
|
|
revision_id=None,
|
|
mode="available",
|
|
priority=0,
|
|
is_active=True,
|
|
metadata={"source": "definition_owner"},
|
|
catalogue=catalogue,
|
|
actor_id=actor_id,
|
|
)
|
|
assignment.revision_id = None
|
|
assignment.is_active = True
|
|
assignment.metadata_ = {
|
|
**dict(assignment.metadata_ or {}),
|
|
"source": "definition_owner",
|
|
}
|
|
assignment.updated_by = actor_id
|
|
assignment.updated_at = _now()
|
|
session.flush()
|
|
return assignment
|
|
|
|
|
|
def archive_definition(
|
|
session: Session,
|
|
definition: ViewDefinition,
|
|
*,
|
|
actor_id: str | None,
|
|
) -> ViewDefinition:
|
|
required_count = (
|
|
session.query(ViewAssignment)
|
|
.filter(
|
|
ViewAssignment.definition_id == definition.id,
|
|
ViewAssignment.mode == "required",
|
|
ViewAssignment.is_active.is_(True),
|
|
)
|
|
.count()
|
|
)
|
|
if required_count:
|
|
raise ViewsConflictError(
|
|
"Remove or replace required assignments before archiving this View"
|
|
)
|
|
session.query(ViewAssignment).filter(
|
|
ViewAssignment.definition_id == definition.id
|
|
).update({"is_active": False}, synchronize_session=False)
|
|
definition.status = "archived"
|
|
definition.updated_by = actor_id
|
|
definition.updated_at = _now()
|
|
session.flush()
|
|
return definition
|
|
|
|
|
|
def list_assignments(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
scope_type: str,
|
|
include_inherited: bool = True,
|
|
) -> list[ViewAssignment]:
|
|
if scope_type == "system":
|
|
query = session.query(ViewAssignment).filter(ViewAssignment.tenant_id.is_(None))
|
|
elif scope_type == "tenant":
|
|
query = session.query(ViewAssignment).filter(
|
|
ViewAssignment.tenant_id == tenant_id
|
|
)
|
|
if include_inherited:
|
|
query = session.query(ViewAssignment).filter(
|
|
or_(
|
|
ViewAssignment.tenant_id == tenant_id,
|
|
ViewAssignment.tenant_id.is_(None),
|
|
)
|
|
)
|
|
else:
|
|
raise ViewsValidationError(
|
|
f"Unsupported assignment listing scope: {scope_type}"
|
|
)
|
|
return query.order_by(
|
|
ViewAssignment.scope_type.asc(),
|
|
ViewAssignment.priority.desc(),
|
|
ViewAssignment.created_at.asc(),
|
|
).all()
|
|
|
|
|
|
def _validate_assignment_revision(
|
|
session: Session,
|
|
definition: ViewDefinition,
|
|
revision_id: str | None,
|
|
*,
|
|
allow_historical_pin: bool = False,
|
|
) -> ViewRevision:
|
|
if definition.status != "published" or not definition.published_revision_id:
|
|
raise ViewsConflictError("Publish the View before assigning it")
|
|
selected_revision_id = revision_id or definition.published_revision_id
|
|
revision = session.get(ViewRevision, selected_revision_id)
|
|
if revision is None or revision.definition_id != definition.id:
|
|
raise ViewsValidationError("The selected revision does not belong to this View")
|
|
if (
|
|
revision_id
|
|
and revision_id != definition.published_revision_id
|
|
and not allow_historical_pin
|
|
):
|
|
raise ViewsValidationError(
|
|
"Only the currently published revision can be pinned to a new assignment"
|
|
)
|
|
return revision
|
|
|
|
|
|
def _deactivate_competing_assignment(
|
|
session: Session,
|
|
*,
|
|
assignment: ViewAssignment,
|
|
) -> None:
|
|
if assignment.mode not in {"default", "required"} or not assignment.is_active:
|
|
return
|
|
(
|
|
session.query(ViewAssignment)
|
|
.filter(
|
|
ViewAssignment.target_key == assignment.target_key,
|
|
ViewAssignment.mode == assignment.mode,
|
|
ViewAssignment.is_active.is_(True),
|
|
ViewAssignment.id != assignment.id,
|
|
)
|
|
.update({"is_active": False}, synchronize_session=False)
|
|
)
|
|
|
|
|
|
def create_assignment(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
scope_type: str,
|
|
scope_id: str | None,
|
|
definition: ViewDefinition,
|
|
revision_id: str | None,
|
|
mode: str,
|
|
priority: int,
|
|
is_active: bool,
|
|
metadata: dict[str, object],
|
|
catalogue: Iterable[ViewSurface],
|
|
actor_id: str | None,
|
|
) -> ViewAssignment:
|
|
if mode not in ASSIGNMENT_MODES:
|
|
raise ViewsValidationError(f"Unsupported View assignment mode: {mode}")
|
|
row_tenant_id, target_id, target_key = _assignment_target(
|
|
scope_type,
|
|
scope_id,
|
|
tenant_id=tenant_id,
|
|
)
|
|
if scope_type == "system" and definition.scope_type != "system":
|
|
raise ViewsValidationError("System assignments can only use system Views")
|
|
if definition.tenant_id not in {None, tenant_id}:
|
|
raise ViewsNotFoundError("View definition not found")
|
|
revision = _validate_assignment_revision(session, definition, revision_id)
|
|
if mode == "required" and is_active:
|
|
validate_required_assignment_revision(
|
|
revision,
|
|
assignment_scope_type=scope_type,
|
|
catalogue=catalogue,
|
|
)
|
|
duplicate = (
|
|
session.query(ViewAssignment.id)
|
|
.filter(
|
|
ViewAssignment.target_key == target_key,
|
|
ViewAssignment.definition_id == definition.id,
|
|
ViewAssignment.mode == mode,
|
|
)
|
|
.first()
|
|
)
|
|
if duplicate is not None:
|
|
raise ViewsConflictError(
|
|
"This View already has an assignment with the same target and mode"
|
|
)
|
|
assignment = ViewAssignment(
|
|
tenant_id=row_tenant_id,
|
|
scope_type=scope_type,
|
|
scope_id=target_id,
|
|
target_key=target_key,
|
|
definition_id=definition.id,
|
|
revision_id=revision_id,
|
|
mode=mode,
|
|
priority=priority,
|
|
is_active=is_active,
|
|
metadata_=metadata,
|
|
created_by=actor_id,
|
|
updated_by=actor_id,
|
|
)
|
|
session.add(assignment)
|
|
session.flush()
|
|
_deactivate_competing_assignment(session, assignment=assignment)
|
|
session.flush()
|
|
return assignment
|
|
|
|
|
|
def get_assignment(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
assignment_id: str,
|
|
include_system: bool = True,
|
|
) -> ViewAssignment:
|
|
query = session.query(ViewAssignment).filter(ViewAssignment.id == assignment_id)
|
|
if include_system:
|
|
query = query.filter(
|
|
or_(
|
|
ViewAssignment.tenant_id == tenant_id,
|
|
ViewAssignment.tenant_id.is_(None),
|
|
)
|
|
)
|
|
else:
|
|
query = query.filter(ViewAssignment.tenant_id == tenant_id)
|
|
assignment = query.first()
|
|
if assignment is None:
|
|
raise ViewsNotFoundError("View assignment not found")
|
|
return assignment
|
|
|
|
|
|
def update_assignment(
|
|
session: Session,
|
|
assignment: ViewAssignment,
|
|
*,
|
|
updates: dict[str, object],
|
|
catalogue: Iterable[ViewSurface],
|
|
actor_id: str | None,
|
|
) -> ViewAssignment:
|
|
next_mode = str(updates.get("mode", assignment.mode))
|
|
next_revision_id = (
|
|
updates["revision_id"] if "revision_id" in updates else assignment.revision_id
|
|
)
|
|
next_active = bool(updates.get("is_active", assignment.is_active))
|
|
if next_mode not in ASSIGNMENT_MODES:
|
|
raise ViewsValidationError(f"Unsupported View assignment mode: {next_mode}")
|
|
revision = _validate_assignment_revision(
|
|
session,
|
|
assignment.definition,
|
|
next_revision_id if isinstance(next_revision_id, str) else None,
|
|
allow_historical_pin=(
|
|
isinstance(next_revision_id, str)
|
|
and next_revision_id == assignment.revision_id
|
|
),
|
|
)
|
|
if next_mode == "required" and next_active:
|
|
validate_required_assignment_revision(
|
|
revision,
|
|
assignment_scope_type=assignment.scope_type,
|
|
catalogue=catalogue,
|
|
)
|
|
duplicate = (
|
|
session.query(ViewAssignment.id)
|
|
.filter(
|
|
ViewAssignment.target_key == assignment.target_key,
|
|
ViewAssignment.definition_id == assignment.definition_id,
|
|
ViewAssignment.mode == next_mode,
|
|
ViewAssignment.id != assignment.id,
|
|
)
|
|
.first()
|
|
)
|
|
if duplicate is not None:
|
|
raise ViewsConflictError(
|
|
"This View already has an assignment with the same target and mode"
|
|
)
|
|
assignment.mode = next_mode
|
|
if "revision_id" in updates:
|
|
assignment.revision_id = (
|
|
str(next_revision_id) if next_revision_id is not None else None
|
|
)
|
|
if "priority" in updates:
|
|
assignment.priority = int(updates["priority"])
|
|
if "is_active" in updates:
|
|
assignment.is_active = next_active
|
|
if "metadata" in updates:
|
|
assignment.metadata_ = dict(updates["metadata"] or {})
|
|
assignment.updated_by = actor_id
|
|
assignment.updated_at = _now()
|
|
session.flush()
|
|
_deactivate_competing_assignment(session, assignment=assignment)
|
|
session.flush()
|
|
return assignment
|
|
|
|
|
|
def delete_assignment(session: Session, assignment: ViewAssignment) -> None:
|
|
session.delete(assignment)
|
|
session.flush()
|
|
|
|
|
|
def _assignment_order(assignment: ViewAssignment) -> tuple[int, int, float, str]:
|
|
specificity = {
|
|
"system": 0,
|
|
"tenant": 1,
|
|
"group": 2,
|
|
"user": 3,
|
|
}.get(assignment.scope_type, -1)
|
|
updated_at = assignment.updated_at
|
|
if updated_at.tzinfo is None:
|
|
updated_at = updated_at.replace(tzinfo=timezone.utc)
|
|
return specificity, assignment.priority, updated_at.timestamp(), assignment.id
|
|
|
|
|
|
def _preference(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
account_id: str,
|
|
) -> ViewPreference | None:
|
|
return (
|
|
session.query(ViewPreference)
|
|
.filter(
|
|
ViewPreference.tenant_id == tenant_id,
|
|
ViewPreference.account_id == account_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
|
|
def _collect_effective_view_candidates(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
account_id: str,
|
|
group_ids: frozenset[str],
|
|
) -> tuple[tuple[ViewAssignment, ...], bool]:
|
|
scope_filters = [
|
|
and_(
|
|
ViewAssignment.scope_type == "system",
|
|
ViewAssignment.tenant_id.is_(None),
|
|
),
|
|
and_(
|
|
ViewAssignment.scope_type == "tenant",
|
|
ViewAssignment.tenant_id == tenant_id,
|
|
ViewAssignment.scope_id == tenant_id,
|
|
),
|
|
and_(
|
|
ViewAssignment.scope_type == "user",
|
|
ViewAssignment.tenant_id == tenant_id,
|
|
ViewAssignment.scope_id == account_id,
|
|
),
|
|
]
|
|
if group_ids:
|
|
scope_filters.append(
|
|
and_(
|
|
ViewAssignment.scope_type == "group",
|
|
ViewAssignment.tenant_id == tenant_id,
|
|
ViewAssignment.scope_id.in_(group_ids),
|
|
)
|
|
)
|
|
rows = (
|
|
session.query(ViewAssignment)
|
|
.join(ViewDefinition, ViewAssignment.definition_id == ViewDefinition.id)
|
|
.options(joinedload(ViewAssignment.definition))
|
|
.filter(
|
|
ViewAssignment.is_active.is_(True),
|
|
ViewDefinition.status == "published",
|
|
ViewDefinition.deleted_at.is_(None),
|
|
or_(*scope_filters),
|
|
)
|
|
.order_by(ViewAssignment.id.asc())
|
|
.limit(MAX_EFFECTIVE_VIEW_ASSIGNMENTS + 1)
|
|
.all()
|
|
)
|
|
return (
|
|
tuple(rows[:MAX_EFFECTIVE_VIEW_ASSIGNMENTS]),
|
|
len(rows) > MAX_EFFECTIVE_VIEW_ASSIGNMENTS,
|
|
)
|
|
|
|
|
|
def _resolve_assignment_revisions(
|
|
session: Session,
|
|
assignments: Iterable[ViewAssignment],
|
|
) -> tuple[ResolvedAssignment, ...]:
|
|
candidates = tuple(assignments)
|
|
revision_ids = {
|
|
revision_id
|
|
for assignment in candidates
|
|
if (
|
|
revision_id := (
|
|
assignment.revision_id
|
|
or assignment.definition.published_revision_id
|
|
)
|
|
)
|
|
}
|
|
revisions = (
|
|
session.query(ViewRevision)
|
|
.filter(ViewRevision.id.in_(revision_ids))
|
|
.all()
|
|
if revision_ids
|
|
else ()
|
|
)
|
|
by_id = {revision.id: revision for revision in revisions}
|
|
resolved: list[ResolvedAssignment] = []
|
|
for assignment in candidates:
|
|
revision_id = (
|
|
assignment.revision_id
|
|
or assignment.definition.published_revision_id
|
|
)
|
|
revision = by_id.get(revision_id)
|
|
if (
|
|
revision is not None
|
|
and revision.definition_id == assignment.definition_id
|
|
):
|
|
resolved.append((assignment, revision))
|
|
return tuple(resolved)
|
|
|
|
|
|
def _option_assignments(
|
|
assignments: Iterable[ResolvedAssignment],
|
|
) -> dict[str, ResolvedAssignment]:
|
|
options: dict[str, ResolvedAssignment] = {}
|
|
for assignment, revision in assignments:
|
|
existing = options.get(assignment.definition_id)
|
|
if existing is None or _assignment_order(assignment) > _assignment_order(
|
|
existing[0]
|
|
):
|
|
options[assignment.definition_id] = (assignment, revision)
|
|
return options
|
|
|
|
|
|
def _available_view_options(
|
|
options: dict[str, ResolvedAssignment],
|
|
) -> tuple[EffectiveViewOption, ...]:
|
|
ordered = sorted(
|
|
options.values(),
|
|
key=lambda item: (
|
|
-_assignment_order(item[0])[0],
|
|
-_assignment_order(item[0])[1],
|
|
item[0].definition.name.lower(),
|
|
item[0].definition_id,
|
|
),
|
|
)
|
|
return tuple(
|
|
EffectiveViewOption(
|
|
id=assignment.definition.id,
|
|
name=assignment.definition.name,
|
|
description=assignment.definition.description,
|
|
revision_id=revision.id,
|
|
)
|
|
for assignment, revision in ordered
|
|
)
|
|
|
|
|
|
def _assignment_provenance(
|
|
source: str,
|
|
assignment: ViewAssignment,
|
|
detail: str,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"source": source,
|
|
"scope_type": assignment.scope_type,
|
|
"scope_id": assignment.scope_id,
|
|
"detail": detail,
|
|
}
|
|
|
|
|
|
def _select_effective_assignment(
|
|
assignments: tuple[ResolvedAssignment, ...],
|
|
options: dict[str, ResolvedAssignment],
|
|
*,
|
|
preference: ViewPreference | None,
|
|
account_id: str,
|
|
workflow_view_id: str | None,
|
|
) -> ViewSelection:
|
|
required = [
|
|
pair for pair in assignments if pair[0].mode == "required"
|
|
]
|
|
if required:
|
|
selected = max(
|
|
required,
|
|
key=lambda pair: _assignment_order(pair[0]),
|
|
)
|
|
return ViewSelection(
|
|
selected=selected,
|
|
locked=True,
|
|
provenance=(
|
|
_assignment_provenance(
|
|
"required_assignment",
|
|
selected[0],
|
|
f"Required View assignment {selected[0].id}",
|
|
),
|
|
),
|
|
)
|
|
|
|
diagnostics: list[ViewDiagnostic] = []
|
|
if workflow_view_id is not None:
|
|
selected = options.get(workflow_view_id)
|
|
if selected is not None:
|
|
return ViewSelection(
|
|
selected=selected,
|
|
locked=False,
|
|
provenance=(
|
|
_assignment_provenance(
|
|
"workflow_selection",
|
|
selected[0],
|
|
f"Workflow selected View {workflow_view_id}",
|
|
),
|
|
),
|
|
)
|
|
diagnostics.append(
|
|
ViewDiagnostic(
|
|
severity="warning",
|
|
code="view.workflow_selection_unavailable",
|
|
message=(
|
|
"The workflow-selected View is not available to this "
|
|
"account. Normal View selection was used."
|
|
),
|
|
)
|
|
)
|
|
|
|
if preference and preference.selection_kind == "none":
|
|
return ViewSelection(
|
|
selected=None,
|
|
locked=False,
|
|
provenance=(
|
|
{
|
|
"source": "user_selection",
|
|
"scope_type": "user",
|
|
"scope_id": account_id,
|
|
"detail": "Full interface selected",
|
|
},
|
|
),
|
|
diagnostics=tuple(diagnostics),
|
|
)
|
|
if (
|
|
preference
|
|
and preference.selection_kind == "view"
|
|
and preference.view_id in options
|
|
):
|
|
selected = options[preference.view_id]
|
|
return ViewSelection(
|
|
selected=selected,
|
|
locked=False,
|
|
provenance=(
|
|
{
|
|
"source": "user_selection",
|
|
"scope_type": "user",
|
|
"scope_id": account_id,
|
|
"detail": f"Selected View {preference.view_id}",
|
|
},
|
|
),
|
|
diagnostics=tuple(diagnostics),
|
|
)
|
|
|
|
defaults = [
|
|
pair for pair in assignments if pair[0].mode == "default"
|
|
]
|
|
if defaults:
|
|
selected = max(
|
|
defaults,
|
|
key=lambda pair: _assignment_order(pair[0]),
|
|
)
|
|
return ViewSelection(
|
|
selected=selected,
|
|
locked=False,
|
|
provenance=(
|
|
_assignment_provenance(
|
|
"default_assignment",
|
|
selected[0],
|
|
f"Default View assignment {selected[0].id}",
|
|
),
|
|
),
|
|
diagnostics=tuple(diagnostics),
|
|
)
|
|
return ViewSelection(
|
|
selected=None,
|
|
locked=False,
|
|
provenance=(),
|
|
diagnostics=tuple(diagnostics),
|
|
)
|
|
|
|
|
|
def _evaluate_selected_surfaces(
|
|
selection: ViewSelection,
|
|
*,
|
|
catalogue: tuple[ViewSurface, ...] | None,
|
|
) -> ViewSelection:
|
|
if selection.selected is None or catalogue is None:
|
|
return selection
|
|
assignment, revision = selection.selected
|
|
known_surfaces = {surface.id: surface for surface in catalogue}
|
|
stale_ids = sorted(
|
|
set(revision.visible_surface_ids) - set(known_surfaces)
|
|
)
|
|
diagnostics = list(selection.diagnostics)
|
|
provenance = list(selection.provenance)
|
|
if stale_ids:
|
|
diagnostics.append(
|
|
ViewDiagnostic(
|
|
severity="warning",
|
|
code="view.stale_surfaces",
|
|
message=(
|
|
"The active View references surfaces that are no longer "
|
|
"announced by the active module graph."
|
|
),
|
|
surface_ids=tuple(stale_ids),
|
|
)
|
|
)
|
|
|
|
if selection.locked:
|
|
lockout_diagnostic = _required_lockout_diagnostic(
|
|
assignment,
|
|
revision,
|
|
catalogue=catalogue,
|
|
)
|
|
if lockout_diagnostic is not None:
|
|
diagnostics.append(lockout_diagnostic)
|
|
provenance.append(
|
|
_fallback_provenance(
|
|
assignment,
|
|
detail=(
|
|
"Required View administration escape is unavailable "
|
|
f"for revision {revision.id}"
|
|
),
|
|
)
|
|
)
|
|
return ViewSelection(
|
|
selected=None,
|
|
locked=False,
|
|
provenance=tuple(provenance),
|
|
diagnostics=tuple(diagnostics),
|
|
)
|
|
|
|
known_visible = [
|
|
known_surfaces[surface_id]
|
|
for surface_id in revision.visible_surface_ids
|
|
if surface_id in known_surfaces
|
|
]
|
|
if (
|
|
not any(surface.kind == "navigation" for surface in known_visible)
|
|
or not any(surface.kind == "route" for surface in known_visible)
|
|
):
|
|
diagnostics.append(
|
|
ViewDiagnostic(
|
|
severity="error",
|
|
code="view.no_reachable_surface",
|
|
message=(
|
|
"The active View no longer contains a reachable navigation "
|
|
"entry and route. The full authorized interface is shown."
|
|
),
|
|
)
|
|
)
|
|
provenance.append(
|
|
_fallback_provenance(
|
|
assignment,
|
|
detail=f"Invalid View revision {revision.id}",
|
|
)
|
|
)
|
|
return ViewSelection(
|
|
selected=None,
|
|
locked=False,
|
|
provenance=tuple(provenance),
|
|
diagnostics=tuple(diagnostics),
|
|
)
|
|
return ViewSelection(
|
|
selected=selection.selected,
|
|
locked=selection.locked,
|
|
provenance=tuple(provenance),
|
|
diagnostics=tuple(diagnostics),
|
|
)
|
|
|
|
|
|
def _required_lockout_diagnostic(
|
|
assignment: ViewAssignment,
|
|
revision: ViewRevision,
|
|
*,
|
|
catalogue: tuple[ViewSurface, ...],
|
|
) -> ViewDiagnostic | None:
|
|
try:
|
|
required_ids = lockout_required_surface_ids(
|
|
catalogue,
|
|
assignment_scope_type=assignment.scope_type,
|
|
)
|
|
missing_ids = tuple(
|
|
sorted(required_ids - set(revision.visible_surface_ids))
|
|
)
|
|
message = (
|
|
"The required View no longer retains all administration escape surfaces."
|
|
if missing_ids
|
|
else None
|
|
)
|
|
except ViewsValidationError as exc:
|
|
missing_ids = ()
|
|
message = str(exc)
|
|
if message is None:
|
|
return None
|
|
return ViewDiagnostic(
|
|
severity="error",
|
|
code="view.lockout_escape_unavailable",
|
|
message=(
|
|
f"{message} The full authorized interface is shown until the "
|
|
"assignment is repaired."
|
|
),
|
|
surface_ids=missing_ids,
|
|
)
|
|
|
|
|
|
def _fallback_provenance(
|
|
assignment: ViewAssignment,
|
|
*,
|
|
detail: str,
|
|
) -> dict[str, object]:
|
|
return _assignment_provenance(
|
|
"invalid_view_fallback",
|
|
assignment,
|
|
detail,
|
|
)
|
|
|
|
|
|
def _effective_view_from_selection(selection: ViewSelection) -> EffectiveView:
|
|
if selection.selected is None:
|
|
return EffectiveView(
|
|
view_id=None,
|
|
revision_id=None,
|
|
name=None,
|
|
visible_surface_ids=frozenset(),
|
|
locked=False,
|
|
provenance=selection.provenance,
|
|
)
|
|
assignment, revision = selection.selected
|
|
return EffectiveView(
|
|
view_id=assignment.definition.id,
|
|
revision_id=revision.id,
|
|
name=assignment.definition.name,
|
|
visible_surface_ids=frozenset(revision.visible_surface_ids),
|
|
locked=selection.locked,
|
|
provenance=selection.provenance,
|
|
)
|
|
|
|
|
|
def _resolution_invalidation_token(
|
|
assignments: Iterable[ResolvedAssignment],
|
|
*,
|
|
preference: ViewPreference | None,
|
|
catalogue: tuple[ViewSurface, ...] | None,
|
|
workflow_view_id: str | None,
|
|
) -> str:
|
|
payload = {
|
|
"assignments": [
|
|
{
|
|
"id": assignment.id,
|
|
"revision_id": revision.id,
|
|
"mode": assignment.mode,
|
|
"priority": assignment.priority,
|
|
"updated_at": assignment.updated_at.isoformat(),
|
|
}
|
|
for assignment, revision in sorted(
|
|
assignments,
|
|
key=lambda pair: pair[0].id,
|
|
)
|
|
],
|
|
"preference": (
|
|
{
|
|
"selection_kind": preference.selection_kind,
|
|
"view_id": preference.view_id,
|
|
"updated_at": preference.updated_at.isoformat(),
|
|
}
|
|
if preference is not None
|
|
else None
|
|
),
|
|
"surfaces": (
|
|
sorted(
|
|
(
|
|
surface.id,
|
|
surface.module_id,
|
|
surface.kind,
|
|
surface.parent_id,
|
|
surface.required,
|
|
)
|
|
for surface in catalogue
|
|
)
|
|
if catalogue is not None
|
|
else None
|
|
),
|
|
"workflow_view_id": workflow_view_id,
|
|
}
|
|
encoded = json.dumps(
|
|
payload,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def resolve_effective_view(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
account_id: str,
|
|
group_ids: Iterable[str] = (),
|
|
catalogue: Iterable[ViewSurface] | None = None,
|
|
workflow_view_id: str | None = None,
|
|
) -> EffectiveViewState:
|
|
groups = frozenset(str(group_id) for group_id in group_ids)
|
|
if len(groups) > MAX_EFFECTIVE_VIEW_GROUPS:
|
|
raise ViewsValidationError(
|
|
"Too many group memberships were supplied for View resolution"
|
|
)
|
|
current_catalogue = tuple(catalogue) if catalogue is not None else None
|
|
candidates, overflow = _collect_effective_view_candidates(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
account_id=account_id,
|
|
group_ids=groups,
|
|
)
|
|
if overflow:
|
|
diagnostic = ViewDiagnostic(
|
|
severity="error",
|
|
code="view.assignment_limit_exceeded",
|
|
message=(
|
|
"View resolution exceeded its assignment safety limit. The "
|
|
"full authorized interface is shown."
|
|
),
|
|
)
|
|
return EffectiveViewState(
|
|
effective=_effective_view_from_selection(
|
|
ViewSelection(
|
|
selected=None,
|
|
locked=False,
|
|
provenance=(
|
|
{
|
|
"source": "invalid_view_fallback",
|
|
"scope_type": "user",
|
|
"scope_id": account_id,
|
|
"detail": "View assignment safety limit exceeded",
|
|
},
|
|
),
|
|
diagnostics=(diagnostic,),
|
|
)
|
|
),
|
|
available_views=(),
|
|
diagnostics=(diagnostic,),
|
|
)
|
|
resolved_assignments = _resolve_assignment_revisions(
|
|
session,
|
|
candidates,
|
|
)
|
|
options = _option_assignments(resolved_assignments)
|
|
preference = _preference(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
account_id=account_id,
|
|
)
|
|
selection = _select_effective_assignment(
|
|
resolved_assignments,
|
|
options,
|
|
preference=preference,
|
|
account_id=account_id,
|
|
workflow_view_id=workflow_view_id,
|
|
)
|
|
evaluated = _evaluate_selected_surfaces(
|
|
selection,
|
|
catalogue=current_catalogue,
|
|
)
|
|
return EffectiveViewState(
|
|
effective=_effective_view_from_selection(evaluated),
|
|
available_views=_available_view_options(options),
|
|
diagnostics=evaluated.diagnostics,
|
|
invalidation_token=_resolution_invalidation_token(
|
|
resolved_assignments,
|
|
preference=preference,
|
|
catalogue=current_catalogue,
|
|
workflow_view_id=workflow_view_id,
|
|
),
|
|
)
|
|
|
|
|
|
def select_view(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
account_id: str,
|
|
group_ids: Iterable[str],
|
|
view_id: str | None,
|
|
catalogue: Iterable[ViewSurface] | None = None,
|
|
) -> EffectiveViewState:
|
|
current = resolve_effective_view(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
account_id=account_id,
|
|
group_ids=group_ids,
|
|
catalogue=catalogue,
|
|
)
|
|
if current.effective.locked and view_id != current.effective.view_id:
|
|
raise ViewsConflictError(
|
|
"The effective View is required by an administrator and cannot be changed"
|
|
)
|
|
available_ids = {option.id for option in current.available_views}
|
|
if view_id is not None and view_id not in available_ids:
|
|
raise ViewsValidationError("The selected View is not available to this account")
|
|
|
|
preference = _preference(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
account_id=account_id,
|
|
)
|
|
if preference is None:
|
|
preference = ViewPreference(
|
|
tenant_id=tenant_id,
|
|
account_id=account_id,
|
|
)
|
|
session.add(preference)
|
|
preference.selection_kind = "view" if view_id is not None else "none"
|
|
preference.view_id = view_id
|
|
preference.updated_at = _now()
|
|
session.flush()
|
|
return resolve_effective_view(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
account_id=account_id,
|
|
group_ids=group_ids,
|
|
catalogue=catalogue,
|
|
)
|
|
|
|
|
|
def definition_payload(
|
|
session: Session,
|
|
definition: ViewDefinition,
|
|
*,
|
|
readonly: bool,
|
|
catalogue: Iterable[ViewSurface] | None = None,
|
|
) -> dict[str, object]:
|
|
latest = get_revision(session, definition_id=definition.id)
|
|
published = (
|
|
session.get(ViewRevision, definition.published_revision_id)
|
|
if definition.published_revision_id
|
|
else None
|
|
)
|
|
stale_surface_ids = (
|
|
sorted(set(latest.visible_surface_ids) - {surface.id for surface in catalogue})
|
|
if catalogue is not None
|
|
else []
|
|
)
|
|
return {
|
|
"id": definition.id,
|
|
"tenant_id": definition.tenant_id,
|
|
"scope_type": definition.scope_type,
|
|
"scope_id": definition.scope_id,
|
|
"definition_key": definition.definition_key,
|
|
"name": definition.name,
|
|
"description": definition.description,
|
|
"status": definition.status,
|
|
"current_revision": definition.current_revision,
|
|
"latest_revision": latest,
|
|
"published_revision": published,
|
|
"readonly": readonly,
|
|
"stale_surface_ids": stale_surface_ids,
|
|
"created_by": definition.created_by,
|
|
"updated_by": definition.updated_by,
|
|
"created_at": definition.created_at,
|
|
"updated_at": definition.updated_at,
|
|
}
|
|
|
|
|
|
def assignment_payload(assignment: ViewAssignment) -> dict[str, object]:
|
|
return {
|
|
"id": assignment.id,
|
|
"tenant_id": assignment.tenant_id,
|
|
"scope_type": assignment.scope_type,
|
|
"scope_id": assignment.scope_id,
|
|
"definition_id": assignment.definition_id,
|
|
"revision_id": assignment.revision_id,
|
|
"mode": assignment.mode,
|
|
"priority": assignment.priority,
|
|
"is_active": assignment.is_active,
|
|
"metadata": assignment.metadata_,
|
|
"created_by": assignment.created_by,
|
|
"updated_by": assignment.updated_by,
|
|
"created_at": assignment.created_at,
|
|
"updated_at": assignment.updated_at,
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"EffectiveViewOption",
|
|
"EffectiveViewState",
|
|
"ViewDiagnostic",
|
|
"ViewsConflictError",
|
|
"ViewsError",
|
|
"ViewsNotFoundError",
|
|
"ViewsValidationError",
|
|
"archive_definition",
|
|
"assignment_payload",
|
|
"create_assignment",
|
|
"create_definition",
|
|
"create_revision",
|
|
"definition_payload",
|
|
"definition_revisions",
|
|
"delete_assignment",
|
|
"get_assignment",
|
|
"get_definition",
|
|
"get_revision",
|
|
"ensure_owner_available_assignment",
|
|
"list_assignments",
|
|
"list_definitions",
|
|
"lockout_required_surface_ids",
|
|
"normalize_visible_surface_ids",
|
|
"publish_revision",
|
|
"resolve_effective_view",
|
|
"select_view",
|
|
"update_assignment",
|
|
"update_definition",
|
|
"validate_required_assignment_revision",
|
|
]
|