GovOPlaN/govoplan-templates
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(templates): add governed DSAR coverage

Browse Source
This commit is contained in:
zemion
2026-08-21 13:24:13 +02:00
parent 5de54ccd2e
commit 80001bbe98
3 changed files with 705 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_templates/backend/dsar_provider.py
+377
View File
@@ -0,0 +1,377 @@
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from sqlalchemy import or_
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarExecutionResultRef,
DsarRecordRef,
DsarSubjectRef,
dsar_capability_name,
)
from govoplan_templates.backend.db.models import (
TemplateDefinition,
TemplateRender,
TemplateRevision,
)
TEMPLATES_DSAR_CAPABILITY = dsar_capability_name("templates")
_MAX_RECORDS = 5_000
_CONFLICT = object()
@dataclass(frozen=True, slots=True)
class _Selectors:
account_id: str
template_id: str | None
revision_id: str | None
render_id: str | None
@property
def narrowed(self) -> bool:
return bool(self.template_id or self.revision_id or self.render_id)
class TemplatesDsarProvider:
provider_id = "templates"
module_id = "templates"
def search_subject(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
) -> Sequence[DsarRecordRef]:
db = _session(session)
selectors = _selectors(subject)
if selectors is None:
return ()
records: list[DsarRecordRef] = []
if not selectors.narrowed or selectors.template_id:
query = db.query(TemplateDefinition).filter(
TemplateDefinition.tenant_id == tenant_id,
or_(
TemplateDefinition.created_by_account_id == selectors.account_id,
TemplateDefinition.updated_by_account_id == selectors.account_id,
),
)
if selectors.template_id:
query = query.filter(TemplateDefinition.id == selectors.template_id)
records.extend(
_definition_record(row, selectors.account_id)
for row in _limited(
query,
TemplateDefinition.created_at,
TemplateDefinition.id,
label="definition attribution",
)
)
if not selectors.narrowed or selectors.template_id or selectors.revision_id:
query = db.query(TemplateRevision).filter(
TemplateRevision.tenant_id == tenant_id,
or_(
TemplateRevision.created_by_account_id == selectors.account_id,
TemplateRevision.published_by_account_id == selectors.account_id,
),
)
if selectors.template_id:
query = query.filter(
TemplateRevision.template_id == selectors.template_id
)
if selectors.revision_id:
query = query.filter(TemplateRevision.id == selectors.revision_id)
records.extend(
_revision_record(row, selectors.account_id)
for row in _limited(
query,
TemplateRevision.created_at,
TemplateRevision.id,
label="revision attribution",
)
)
if not selectors.narrowed or selectors.template_id or selectors.render_id:
query = db.query(TemplateRender).filter(
TemplateRender.tenant_id == tenant_id,
TemplateRender.created_by_account_id == selectors.account_id,
)
if selectors.template_id:
query = query.filter(
TemplateRender.template_id == selectors.template_id
)
if selectors.render_id:
query = query.filter(TemplateRender.id == selectors.render_id)
records.extend(
_render_record(row)
for row in _limited(
query,
TemplateRender.created_at,
TemplateRender.id,
label="render attribution",
)
)
if len(records) > _MAX_RECORDS:
raise ValueError("Templates DSAR result limit exceeded; narrow selectors.")
return tuple(
sorted(records, key=lambda item: (item.resource_type, item.resource_id))
)
def plan_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
records: Sequence[DsarRecordRef],
) -> Sequence[DsarErasureActionRef]:
del tenant_id
_session(session)
if _selectors(subject) is None:
raise ValueError("Templates DSAR subject selectors conflict.")
actions = []
for record in records:
_validate_record(record)
actions.append(
DsarErasureActionRef(
action_id=(
f"templates:retain:{record.resource_type}:{record.resource_id}"
),
provider_id=self.provider_id,
module_id=self.module_id,
kind="retain",
resource_type=record.resource_type,
resource_id=record.resource_id,
title=f"Retain {record.title}",
rationale=(
record.retention_reason
or "Template lifecycle attribution remains evidence."
),
executable=False,
)
)
return tuple(actions)
def execute_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
actions: Sequence[DsarErasureActionRef],
request_id: str,
) -> Sequence[DsarExecutionResultRef]:
del tenant_id
_session(session)
if _selectors(subject) is None:
raise ValueError("Templates DSAR subject selectors conflict.")
results = []
for action in actions:
_validate_action(action)
if action.executable or action.kind != "retain":
raise ValueError("Templates DSAR publishes retain actions only.")
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary="Template lifecycle attribution remains evidence.",
evidence={"request_id": request_id},
)
)
return tuple(results)
def _selectors(subject: DsarSubjectRef) -> _Selectors | None:
references = subject.external_references
values = {
"account_id": _coalesce(
subject.account_id,
references.get("templates.account"),
references.get("access.account"),
),
"template_id": _coalesce(
references.get("templates.template"),
references.get("templates.template_id"),
),
"revision_id": _coalesce(
references.get("templates.revision"),
references.get("templates.revision_id"),
),
"render_id": _coalesce(
references.get("templates.render"), references.get("templates.render_id")
),
}
if any(value is _CONFLICT for value in values.values()):
return None
account_id = _optional(values["account_id"])
if not account_id:
return None
return _Selectors(
account_id=account_id,
template_id=_optional(values["template_id"]),
revision_id=_optional(values["revision_id"]),
render_id=_optional(values["render_id"]),
)
def _definition_record(row: TemplateDefinition, account_id: str) -> DsarRecordRef:
activities = []
if row.created_by_account_id == account_id:
activities.append("created_template")
if row.updated_by_account_id == account_id:
activities.append("updated_template")
return _record(
resource_type="template_definition_actor_attribution",
resource_id=row.id,
title="Template-definition actor attribution",
data={
"template_id": row.id,
"template_type": row.template_type,
"scope_type": row.scope_type,
"status": row.status,
"current_revision": row.current_revision,
"resource_revision": row.resource_revision,
"activities": activities,
"created_at": _iso(row.created_at),
"updated_at": _iso(row.updated_at),
"retired_at": _iso(row.deleted_at),
},
observed_at=row.updated_at,
)
def _revision_record(row: TemplateRevision, account_id: str) -> DsarRecordRef:
activities = []
if row.created_by_account_id == account_id:
activities.append("created_template_revision")
if row.published_by_account_id == account_id:
activities.append("published_template_revision")
return _record(
resource_type="template_revision_actor_attribution",
resource_id=row.id,
title="Template-revision actor attribution",
data={
"template_id": row.template_id,
"revision_id": row.id,
"revision": row.revision,
"template_type": row.template_type,
"locale": row.locale,
"activities": activities,
"created_at": _iso(row.created_at),
"published_at": _iso(row.published_at),
},
observed_at=row.published_at or row.created_at,
)
def _render_record(row: TemplateRender) -> DsarRecordRef:
return _record(
resource_type="template_render_actor_attribution",
resource_id=row.id,
title="Template-render actor attribution",
data={
"template_id": row.template_id,
"revision_id": row.revision_id,
"render_id": row.id,
"revision": row.revision_number,
"mode": row.mode,
"usage": row.usage,
"output_format": row.output_format,
"content_type": row.content_type,
"item_count": row.item_count,
"page_count": row.page_count,
"output_size_bytes": row.output_size_bytes,
"activity": "rendered_template",
"created_at": _iso(row.created_at),
},
observed_at=row.created_at,
)
def _record(
*,
resource_type: str,
resource_id: str,
title: str,
data: dict[str, object],
observed_at: datetime | None,
) -> DsarRecordRef:
return DsarRecordRef(
provider_id="templates",
module_id="templates",
resource_type=resource_type,
resource_id=resource_id,
category="template_governance_attribution",
title=title,
data=data,
observed_at=_aware(observed_at),
immutable_evidence=True,
retention_reason=(
"Template definition, publication, and render attribution is retained "
"with immutable lifecycle evidence."
),
)
def _limited(query, first, second, *, label: str):
rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all()
if len(rows) > _MAX_RECORDS:
raise ValueError(f"Templates DSAR {label} limit exceeded; narrow selectors.")
return rows
def _coalesce(*values: str | None) -> str | None | object:
normalized = {str(value).strip() for value in values if str(value or "").strip()}
if len(normalized) > 1:
return _CONFLICT
return next(iter(normalized), None)
def _optional(value: object) -> str | None:
return value if isinstance(value, str) and value else None
def _iso(value: datetime | None) -> str | None:
aware = _aware(value)
return aware.isoformat() if aware else None
def _aware(value: datetime | None) -> datetime | None:
if value is None or value.tzinfo is not None:
return value
return value.replace(tzinfo=timezone.utc)
def _session(value: object) -> Session:
if not isinstance(value, Session):
raise TypeError("Templates DSAR requires a SQLAlchemy Session.")
return value
_RESOURCE_TYPES = {
"template_definition_actor_attribution",
"template_revision_actor_attribution",
"template_render_actor_attribution",
}
def _validate_record(record: DsarRecordRef) -> None:
if record.provider_id != "templates" or record.module_id != "templates":
raise ValueError("Templates DSAR cannot plan a foreign provider record.")
if record.resource_type not in _RESOURCE_TYPES or not record.resource_id:
raise ValueError("Templates DSAR record identity is invalid.")
def _validate_action(action: DsarErasureActionRef) -> None:
if action.provider_id != "templates" or action.module_id != "templates":
raise ValueError("Templates DSAR cannot execute a foreign provider action.")
if not action.action_id.startswith("templates:retain:"):
raise ValueError("Templates DSAR action identity is invalid.")
__all__ = ["TEMPLATES_DSAR_CAPABILITY", "TemplatesDsarProvider"]
src/govoplan_templates/backend/manifest.py
+145 -20
View File
@@ -8,6 +8,7 @@ from govoplan_core.core.module_guards import (
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
CapabilityDocumentation,
DocumentationTopic,
FrontendModule,
FrontendRoute,
@@ -30,6 +31,10 @@ from govoplan_core.core.templates import (
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_templates.backend.db import models as template_models
from govoplan_templates.backend.dsar_provider import (
TEMPLATES_DSAR_CAPABILITY,
TemplatesDsarProvider,
)
MODULE_ID = "templates"
@@ -58,11 +63,29 @@ def _permission(scope: str, label: str, description: str) -> PermissionDefinitio
PERMISSIONS = (
_permission(READ_SCOPE, "View templates", "Read template definitions, revisions, and render evidence."),
_permission(WRITE_SCOPE, "Manage templates", "Create and revise reusable templates."),
_permission(PUBLISH_SCOPE, "Publish templates", "Publish immutable template revisions for final output."),
_permission(RENDER_SCOPE, "Render templates", "Preview and render governed output from supplied snapshots."),
_permission(ADMIN_SCOPE, "Administer templates", "Manage all tenant, group, and user templates."),
_permission(
READ_SCOPE,
"View templates",
"Read template definitions, revisions, and render evidence.",
),
_permission(
WRITE_SCOPE, "Manage templates", "Create and revise reusable templates."
),
_permission(
PUBLISH_SCOPE,
"Publish templates",
"Publish immutable template revisions for final output.",
),
_permission(
RENDER_SCOPE,
"Render templates",
"Preview and render governed output from supplied snapshots.",
),
_permission(
ADMIN_SCOPE,
"Administer templates",
"Manage all tenant, group, and user templates.",
),
)
ROLE_TEMPLATES = (
@@ -221,15 +244,23 @@ def _content_library(context: ModuleContext):
return content_library_capability(context)
def _dsar_provider(_context: ModuleContext) -> TemplatesDsarProvider:
return TemplatesDsarProvider()
def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
return {
"templates": session.query(template_models.TemplateDefinition).filter(
"templates": session.query(template_models.TemplateDefinition)
.filter(
template_models.TemplateDefinition.tenant_id == tenant_id,
template_models.TemplateDefinition.deleted_at.is_(None),
).count(),
"template_renders": session.query(template_models.TemplateRender).filter(
)
.count(),
"template_renders": session.query(template_models.TemplateRender)
.filter(
template_models.TemplateRender.tenant_id == tenant_id,
).count(),
)
.count(),
}
@@ -241,12 +272,17 @@ manifest = ModuleManifest(
optional_dependencies=("files", "dist_lists", "campaigns", "audit"),
optional_capabilities=(CAPABILITY_FILES_ARTIFACT_STORE,),
provides_interfaces=(
ModuleInterfaceProvider(name=CAPABILITY_TEMPLATE_CATALOG, version=MODULE_VERSION),
ModuleInterfaceProvider(
name=CAPABILITY_TEMPLATE_CATALOG, version=MODULE_VERSION
),
ModuleInterfaceProvider(
name=CAPABILITY_TEMPLATE_CONTENT_LIBRARY,
version=MODULE_VERSION,
),
ModuleInterfaceProvider(name=CAPABILITY_TEMPLATE_RENDERER, version=MODULE_VERSION),
ModuleInterfaceProvider(
name=CAPABILITY_TEMPLATE_RENDERER, version=MODULE_VERSION
),
ModuleInterfaceProvider(name=TEMPLATES_DSAR_CAPABILITY, version="0.1.0"),
),
requires_interfaces=(
ModuleInterfaceRequirement(
@@ -263,7 +299,13 @@ manifest = ModuleManifest(
path="/templates",
label=MODULE_NAME,
icon="layout-template",
required_any=(READ_SCOPE, WRITE_SCOPE, PUBLISH_SCOPE, RENDER_SCOPE, ADMIN_SCOPE),
required_any=(
READ_SCOPE,
WRITE_SCOPE,
PUBLISH_SCOPE,
RENDER_SCOPE,
ADMIN_SCOPE,
),
order=75,
),
),
@@ -274,7 +316,13 @@ manifest = ModuleManifest(
FrontendRoute(
path="/templates",
component="TemplatesPage",
required_any=(READ_SCOPE, WRITE_SCOPE, PUBLISH_SCOPE, RENDER_SCOPE, ADMIN_SCOPE),
required_any=(
READ_SCOPE,
WRITE_SCOPE,
PUBLISH_SCOPE,
RENDER_SCOPE,
ADMIN_SCOPE,
),
order=75,
),
),
@@ -283,7 +331,13 @@ manifest = ModuleManifest(
path="/templates",
label=MODULE_NAME,
icon="layout-template",
required_any=(READ_SCOPE, WRITE_SCOPE, PUBLISH_SCOPE, RENDER_SCOPE, ADMIN_SCOPE),
required_any=(
READ_SCOPE,
WRITE_SCOPE,
PUBLISH_SCOPE,
RENDER_SCOPE,
ADMIN_SCOPE,
),
order=75,
),
),
@@ -299,10 +353,34 @@ manifest = ModuleManifest(
),
),
view_surfaces=(
ViewSurface(id="templates.page", module_id=MODULE_ID, kind="route", label="Templates", order=75),
ViewSurface(id="templates.library", module_id=MODULE_ID, kind="section", label="Template library", order=10),
ViewSurface(id="templates.editor", module_id=MODULE_ID, kind="section", label="Template editor", order=20),
ViewSurface(id="templates.preview", module_id=MODULE_ID, kind="section", label="Template preview and output", order=30),
ViewSurface(
id="templates.page",
module_id=MODULE_ID,
kind="route",
label="Templates",
order=75,
),
ViewSurface(
id="templates.library",
module_id=MODULE_ID,
kind="section",
label="Template library",
order=10,
),
ViewSurface(
id="templates.editor",
module_id=MODULE_ID,
kind="section",
label="Template editor",
order=20,
),
ViewSurface(
id="templates.preview",
module_id=MODULE_ID,
kind="section",
label="Template preview and output",
order=30,
),
),
),
route_factory=_router,
@@ -310,6 +388,17 @@ manifest = ModuleManifest(
CAPABILITY_TEMPLATE_CATALOG: _catalog,
CAPABILITY_TEMPLATE_CONTENT_LIBRARY: _content_library,
CAPABILITY_TEMPLATE_RENDERER: _renderer,
TEMPLATES_DSAR_CAPABILITY: _dsar_provider,
},
capability_documentation={
TEMPLATES_DSAR_CAPABILITY: CapabilityDocumentation(
label="Templates data-subject request provider",
summary=(
"Exports minimized template-author and render attribution without "
"template, input, or output payloads."
),
contract_version="0.1.0",
),
},
tenant_summary_providers=(_tenant_summary,),
migration_spec=MigrationSpec(
@@ -336,7 +425,39 @@ manifest = ModuleManifest(
label=MODULE_NAME,
),
),
documentation=DOCUMENTATION,
documentation=(
DocumentationTopic(
id="templates.data-subject-requests",
title="Template data-subject requests",
summary=(
"Export template-author and render activity without content or supplied data."
),
body=(
"Templates correlates only an exact tenant account identifier and can "
"narrow an already verified search to one template, revision, or render. "
"It returns minimized definition, publication, and render lifecycle "
"metadata. Template text and HTML, required fields, layouts, metadata, "
"render input snapshots, filenames, diagnostics, artifact references, "
"output bytes, hashes, and idempotency keys are excluded. Rendered "
"business data belongs to the supplying module and is not inferred from "
"opaque Template payloads. Attribution remains retained with immutable "
"definition and render evidence."
),
layer="configured",
documentation_types=("admin", "user"),
audience=("user", "operator", "module_admin", "auditor"),
related_modules=("core", "files", "campaigns", "audit"),
order=90,
metadata={
"help_contexts": ["templates.page", "privacy.data-subject-requests"],
"consequence_classes": {
"export_template_attribution": "Returns minimized author, publication, and render activity.",
"exclude_template_payloads": "Does not return template content or render inputs and outputs.",
},
},
),
*DOCUMENTATION,
),
architecture=declared_module_architecture(
layer="content_records_evidence",
kind="domain",
@@ -347,7 +468,11 @@ manifest = ModuleManifest(
"The baseline emits safe deterministic HTML/text for browser or OS printing; PDF and printer delivery remain connector concerns.",
),
supported_authority_modes=("native_authoritative",),
owned_concepts=("template definition", "template revision", "template render evidence"),
owned_concepts=(
"template definition",
"template revision",
"template render evidence",
),
non_owned_concepts=("recipient", "campaign", "file asset", "printer endpoint"),
recovery_docs=("docs/TEMPLATE_BOUNDARY.md",),
security_docs=("docs/TEMPLATE_BOUNDARY.md",),