feat: add tenant semantic documentation lifecycle
This commit is contained in:
@@ -0,0 +1,980 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Mapping, Sequence
|
||||
from datetime import UTC, datetime
|
||||
from typing import Literal
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.core.semantic_documentation import (
|
||||
SemanticDocumentationSubjectReference,
|
||||
SemanticDocumentationSubjectResolution,
|
||||
resolve_semantic_documentation_subject,
|
||||
semantic_documentation_fingerprint,
|
||||
)
|
||||
from govoplan_core.tenancy.scope import Tenant
|
||||
|
||||
from govoplan_docs.backend.db.models import (
|
||||
SemanticDocumentationEntry,
|
||||
SemanticDocumentationRevision,
|
||||
)
|
||||
|
||||
|
||||
SEMANTIC_PUBLICATION_POLICY_KEY = "docs.semantic_publication_policy"
|
||||
SEMANTIC_PUBLICATION_MODES = frozenset({"direct", "reviewer_required"})
|
||||
SEMANTIC_LIFECYCLE_STATES = frozenset(
|
||||
{"draft", "published", "superseded", "retired"}
|
||||
)
|
||||
_LOCALE_RE = re.compile(r"^[A-Za-z]{2,3}(?:-[A-Za-z0-9]{2,8})*$")
|
||||
|
||||
|
||||
class SemanticDocumentationError(ValueError):
|
||||
"""Base semantic-documentation service error."""
|
||||
|
||||
|
||||
class SemanticDocumentationNotFoundError(SemanticDocumentationError):
|
||||
pass
|
||||
|
||||
|
||||
class SemanticDocumentationConflictError(SemanticDocumentationError):
|
||||
pass
|
||||
|
||||
|
||||
class SemanticDocumentationAuthorizationError(SemanticDocumentationError):
|
||||
pass
|
||||
|
||||
|
||||
class SemanticDocumentationSubjectError(SemanticDocumentationError):
|
||||
pass
|
||||
|
||||
|
||||
def publication_policy(session: Session, tenant_id: str) -> str:
|
||||
tenant = session.get(Tenant, tenant_id)
|
||||
settings = tenant.settings if tenant is not None else {}
|
||||
raw = settings.get(SEMANTIC_PUBLICATION_POLICY_KEY) if settings else None
|
||||
return str(raw) if raw in SEMANTIC_PUBLICATION_MODES else "reviewer_required"
|
||||
|
||||
|
||||
def set_publication_policy(
|
||||
session: Session,
|
||||
*,
|
||||
tenant_id: str,
|
||||
mode: str,
|
||||
) -> str:
|
||||
if mode not in SEMANTIC_PUBLICATION_MODES:
|
||||
raise SemanticDocumentationError(
|
||||
"Semantic publication policy must be direct or reviewer_required."
|
||||
)
|
||||
tenant = session.get(Tenant, tenant_id)
|
||||
if tenant is None:
|
||||
raise SemanticDocumentationNotFoundError("The active tenant is unavailable.")
|
||||
settings = dict(tenant.settings or {})
|
||||
settings[SEMANTIC_PUBLICATION_POLICY_KEY] = mode
|
||||
tenant.settings = settings
|
||||
session.flush()
|
||||
return mode
|
||||
|
||||
|
||||
def create_semantic_entry(
|
||||
session: Session,
|
||||
registry: object,
|
||||
principal: object,
|
||||
*,
|
||||
subject: SemanticDocumentationSubjectReference,
|
||||
locale: str,
|
||||
content: Mapping[str, object],
|
||||
change_reason: str,
|
||||
now: datetime | None = None,
|
||||
) -> SemanticDocumentationEntry:
|
||||
tenant_id = _principal_tenant_id(principal)
|
||||
_require_tenant(subject, tenant_id)
|
||||
clean_locale = _locale(locale)
|
||||
current_subject = _resolved_current_subject(
|
||||
session,
|
||||
registry,
|
||||
principal,
|
||||
subject,
|
||||
)
|
||||
stable_key = current_subject.stable_key
|
||||
existing = (
|
||||
session.query(SemanticDocumentationEntry)
|
||||
.filter(
|
||||
SemanticDocumentationEntry.tenant_id == tenant_id,
|
||||
SemanticDocumentationEntry.subject_stable_key == stable_key,
|
||||
SemanticDocumentationEntry.locale == clean_locale,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
if existing is not None:
|
||||
raise SemanticDocumentationConflictError(
|
||||
"Semantic documentation already exists for this subject and locale."
|
||||
)
|
||||
actor_id = _principal_account_id(principal)
|
||||
anchor = current_subject.anchor
|
||||
entry = SemanticDocumentationEntry(
|
||||
tenant_id=tenant_id,
|
||||
subject_stable_key=stable_key,
|
||||
subject_module_id=current_subject.module_id,
|
||||
subject_kind=current_subject.subject_kind,
|
||||
subject_id=current_subject.subject_id,
|
||||
anchor_kind=anchor.kind if anchor else None,
|
||||
anchor_id=anchor.id if anchor else None,
|
||||
locale=clean_locale,
|
||||
lifecycle_state="draft",
|
||||
current_revision=1,
|
||||
created_by=actor_id,
|
||||
updated_by=actor_id,
|
||||
)
|
||||
session.add(entry)
|
||||
session.flush()
|
||||
revision = _new_revision(
|
||||
entry,
|
||||
revision=1,
|
||||
lifecycle_state="draft",
|
||||
action="create",
|
||||
content=content,
|
||||
subject=current_subject,
|
||||
actor_id=actor_id,
|
||||
change_reason=change_reason,
|
||||
now=now,
|
||||
)
|
||||
session.add(revision)
|
||||
session.flush()
|
||||
entry.current_revision_id = revision.id
|
||||
session.flush()
|
||||
return entry
|
||||
|
||||
|
||||
def update_semantic_entry(
|
||||
session: Session,
|
||||
registry: object,
|
||||
principal: object,
|
||||
*,
|
||||
entry_id: str,
|
||||
expected_revision: int,
|
||||
content: Mapping[str, object],
|
||||
change_reason: str,
|
||||
now: datetime | None = None,
|
||||
) -> SemanticDocumentationEntry:
|
||||
entry = get_semantic_entry(session, principal, entry_id=entry_id)
|
||||
_require_editable(entry)
|
||||
_require_expected_revision(entry, expected_revision)
|
||||
subject = _resolved_current_subject(
|
||||
session,
|
||||
registry,
|
||||
principal,
|
||||
entry_subject_reference(entry),
|
||||
)
|
||||
actor_id = _principal_account_id(principal)
|
||||
revision = _new_revision(
|
||||
entry,
|
||||
revision=entry.current_revision + 1,
|
||||
lifecycle_state="draft",
|
||||
action="edit",
|
||||
content=content,
|
||||
subject=subject,
|
||||
actor_id=actor_id,
|
||||
change_reason=change_reason,
|
||||
now=now,
|
||||
)
|
||||
session.add(revision)
|
||||
session.flush()
|
||||
entry.current_revision += 1
|
||||
entry.current_revision_id = revision.id
|
||||
entry.lifecycle_state = "draft"
|
||||
entry.updated_by = actor_id
|
||||
session.flush()
|
||||
return entry
|
||||
|
||||
|
||||
def publish_semantic_entry(
|
||||
session: Session,
|
||||
registry: object,
|
||||
principal: object,
|
||||
*,
|
||||
entry_id: str,
|
||||
expected_revision: int,
|
||||
change_reason: str,
|
||||
now: datetime | None = None,
|
||||
) -> SemanticDocumentationEntry:
|
||||
entry = get_semantic_entry(session, principal, entry_id=entry_id)
|
||||
_require_editable(entry)
|
||||
_require_expected_revision(entry, expected_revision)
|
||||
current = current_revision(session, entry)
|
||||
actor_id = _principal_account_id(principal)
|
||||
mode = publication_policy(session, entry.tenant_id)
|
||||
if mode == "reviewer_required" and current.authored_by == actor_id:
|
||||
raise SemanticDocumentationAuthorizationError(
|
||||
"The configured publication policy requires another reviewer."
|
||||
)
|
||||
subject = _resolved_current_subject(
|
||||
session,
|
||||
registry,
|
||||
principal,
|
||||
entry_subject_reference(entry, revision=current),
|
||||
)
|
||||
timestamp = _utc(now)
|
||||
revision = _new_revision(
|
||||
entry,
|
||||
revision=entry.current_revision + 1,
|
||||
lifecycle_state="published",
|
||||
action="publish",
|
||||
content=current.content,
|
||||
subject=subject,
|
||||
actor_id=current.authored_by,
|
||||
reviewer_id=actor_id,
|
||||
change_reason=change_reason,
|
||||
now=timestamp,
|
||||
)
|
||||
revision.published_at = timestamp
|
||||
revision.provenance = {
|
||||
**revision.provenance,
|
||||
"publication_policy": mode,
|
||||
"published_by": actor_id,
|
||||
}
|
||||
session.add(revision)
|
||||
session.flush()
|
||||
entry.current_revision += 1
|
||||
entry.current_revision_id = revision.id
|
||||
entry.published_revision_id = revision.id
|
||||
entry.lifecycle_state = "published"
|
||||
entry.updated_by = actor_id
|
||||
entry.published_by = actor_id
|
||||
entry.published_at = timestamp
|
||||
session.flush()
|
||||
return entry
|
||||
|
||||
|
||||
def supersede_semantic_entry(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
entry_id: str,
|
||||
replacement_entry_id: str,
|
||||
expected_revision: int,
|
||||
change_reason: str,
|
||||
now: datetime | None = None,
|
||||
) -> SemanticDocumentationEntry:
|
||||
entry = get_semantic_entry(session, principal, entry_id=entry_id)
|
||||
_require_editable(entry)
|
||||
_require_expected_revision(entry, expected_revision)
|
||||
if entry.id == replacement_entry_id:
|
||||
raise SemanticDocumentationError("An entry cannot supersede itself.")
|
||||
replacement = get_semantic_entry(
|
||||
session,
|
||||
principal,
|
||||
entry_id=replacement_entry_id,
|
||||
)
|
||||
if replacement.published_revision_id is None:
|
||||
raise SemanticDocumentationError(
|
||||
"The replacement semantic documentation must be published."
|
||||
)
|
||||
current = current_revision(session, entry)
|
||||
actor_id = _principal_account_id(principal)
|
||||
revision = _lifecycle_revision(
|
||||
entry,
|
||||
current=current,
|
||||
lifecycle_state="superseded",
|
||||
action="supersede",
|
||||
actor_id=actor_id,
|
||||
change_reason=change_reason,
|
||||
now=now,
|
||||
provenance={"replacement_entry_id": replacement.id},
|
||||
)
|
||||
session.add(revision)
|
||||
session.flush()
|
||||
entry.current_revision += 1
|
||||
entry.current_revision_id = revision.id
|
||||
entry.lifecycle_state = "superseded"
|
||||
entry.superseded_by_entry_id = replacement.id
|
||||
entry.updated_by = actor_id
|
||||
session.flush()
|
||||
return entry
|
||||
|
||||
|
||||
def retire_semantic_entry(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
entry_id: str,
|
||||
expected_revision: int,
|
||||
change_reason: str,
|
||||
now: datetime | None = None,
|
||||
) -> SemanticDocumentationEntry:
|
||||
entry = get_semantic_entry(session, principal, entry_id=entry_id)
|
||||
_require_editable(entry)
|
||||
_require_expected_revision(entry, expected_revision)
|
||||
current = current_revision(session, entry)
|
||||
actor_id = _principal_account_id(principal)
|
||||
timestamp = _utc(now)
|
||||
revision = _lifecycle_revision(
|
||||
entry,
|
||||
current=current,
|
||||
lifecycle_state="retired",
|
||||
action="retire",
|
||||
actor_id=actor_id,
|
||||
change_reason=change_reason,
|
||||
now=timestamp,
|
||||
)
|
||||
session.add(revision)
|
||||
session.flush()
|
||||
entry.current_revision += 1
|
||||
entry.current_revision_id = revision.id
|
||||
entry.lifecycle_state = "retired"
|
||||
entry.retired_by = actor_id
|
||||
entry.retired_at = timestamp
|
||||
entry.updated_by = actor_id
|
||||
session.flush()
|
||||
return entry
|
||||
|
||||
|
||||
def get_semantic_entry(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
entry_id: str,
|
||||
) -> SemanticDocumentationEntry:
|
||||
tenant_id = _principal_tenant_id(principal)
|
||||
entry = (
|
||||
session.query(SemanticDocumentationEntry)
|
||||
.filter(
|
||||
SemanticDocumentationEntry.id == entry_id,
|
||||
SemanticDocumentationEntry.tenant_id == tenant_id,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
if entry is None:
|
||||
raise SemanticDocumentationNotFoundError(
|
||||
"Semantic documentation entry not found."
|
||||
)
|
||||
return entry
|
||||
|
||||
|
||||
def list_semantic_entries(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
module_id: str | None = None,
|
||||
subject_kind: str | None = None,
|
||||
) -> tuple[SemanticDocumentationEntry, ...]:
|
||||
query = session.query(SemanticDocumentationEntry).filter(
|
||||
SemanticDocumentationEntry.tenant_id == _principal_tenant_id(principal)
|
||||
)
|
||||
if module_id:
|
||||
query = query.filter(SemanticDocumentationEntry.subject_module_id == module_id)
|
||||
if subject_kind:
|
||||
query = query.filter(SemanticDocumentationEntry.subject_kind == subject_kind)
|
||||
return tuple(
|
||||
query.order_by(
|
||||
SemanticDocumentationEntry.subject_module_id.asc(),
|
||||
SemanticDocumentationEntry.subject_kind.asc(),
|
||||
SemanticDocumentationEntry.subject_id.asc(),
|
||||
SemanticDocumentationEntry.locale.asc(),
|
||||
).all()
|
||||
)
|
||||
|
||||
|
||||
def semantic_entry_history(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
entry_id: str,
|
||||
) -> tuple[SemanticDocumentationRevision, ...]:
|
||||
entry = get_semantic_entry(session, principal, entry_id=entry_id)
|
||||
return tuple(
|
||||
session.query(SemanticDocumentationRevision)
|
||||
.filter(
|
||||
SemanticDocumentationRevision.entry_id == entry.id,
|
||||
SemanticDocumentationRevision.tenant_id == entry.tenant_id,
|
||||
)
|
||||
.order_by(SemanticDocumentationRevision.revision.desc())
|
||||
.all()
|
||||
)
|
||||
|
||||
|
||||
def current_revision(
|
||||
session: Session,
|
||||
entry: SemanticDocumentationEntry,
|
||||
) -> SemanticDocumentationRevision:
|
||||
revision = (
|
||||
session.get(SemanticDocumentationRevision, entry.current_revision_id)
|
||||
if entry.current_revision_id
|
||||
else None
|
||||
)
|
||||
if revision is None or revision.entry_id != entry.id:
|
||||
raise SemanticDocumentationError(
|
||||
"Semantic documentation current revision is unavailable."
|
||||
)
|
||||
return revision
|
||||
|
||||
|
||||
def published_revision(
|
||||
session: Session,
|
||||
entry: SemanticDocumentationEntry,
|
||||
) -> SemanticDocumentationRevision | None:
|
||||
if not entry.published_revision_id:
|
||||
return None
|
||||
revision = session.get(
|
||||
SemanticDocumentationRevision,
|
||||
entry.published_revision_id,
|
||||
)
|
||||
return revision if revision is not None and revision.entry_id == entry.id else None
|
||||
|
||||
|
||||
def entry_subject_reference(
|
||||
entry: SemanticDocumentationEntry,
|
||||
*,
|
||||
revision: SemanticDocumentationRevision | None = None,
|
||||
) -> SemanticDocumentationSubjectReference:
|
||||
from govoplan_core.core.semantic_documentation import (
|
||||
SemanticDocumentationSubjectAnchor,
|
||||
)
|
||||
|
||||
return SemanticDocumentationSubjectReference(
|
||||
module_id=entry.subject_module_id,
|
||||
tenant_id=entry.tenant_id,
|
||||
subject_kind=entry.subject_kind,
|
||||
subject_id=entry.subject_id,
|
||||
anchor=(
|
||||
SemanticDocumentationSubjectAnchor(
|
||||
kind=entry.anchor_kind,
|
||||
id=entry.anchor_id,
|
||||
)
|
||||
if entry.anchor_kind and entry.anchor_id
|
||||
else None
|
||||
),
|
||||
observed_revision=revision.subject_revision if revision else None,
|
||||
observed_fingerprint=revision.subject_fingerprint if revision else None,
|
||||
)
|
||||
|
||||
|
||||
def resolve_entry_subject(
|
||||
session: Session,
|
||||
registry: object,
|
||||
principal: object,
|
||||
*,
|
||||
entry: SemanticDocumentationEntry,
|
||||
revision: SemanticDocumentationRevision,
|
||||
) -> SemanticDocumentationSubjectResolution | None:
|
||||
return resolve_semantic_documentation_subject(
|
||||
registry,
|
||||
session,
|
||||
principal,
|
||||
reference=entry_subject_reference(entry, revision=revision),
|
||||
)
|
||||
|
||||
|
||||
def content_visible_to_principal(
|
||||
content: Mapping[str, object],
|
||||
principal: object,
|
||||
) -> bool:
|
||||
audience = tuple(str(item) for item in content.get("audience", ()) or ())
|
||||
classification = str(content.get("classification") or "internal")
|
||||
if classification == "restricted" and not audience:
|
||||
return False
|
||||
if not audience or "authenticated" in audience:
|
||||
return True
|
||||
tokens = _principal_audience_tokens(principal)
|
||||
return any(selector in tokens for selector in audience)
|
||||
|
||||
|
||||
def semantic_entry_payload(
|
||||
session: Session,
|
||||
registry: object,
|
||||
principal: object,
|
||||
*,
|
||||
entry: SemanticDocumentationEntry,
|
||||
editor: bool,
|
||||
requested_locale: str | None = None,
|
||||
) -> dict[str, object] | None:
|
||||
current = current_revision(session, entry)
|
||||
published = published_revision(session, entry)
|
||||
selected = current if editor else published
|
||||
if selected is None:
|
||||
return None
|
||||
resolution = resolve_entry_subject(
|
||||
session,
|
||||
registry,
|
||||
principal,
|
||||
entry=entry,
|
||||
revision=selected,
|
||||
)
|
||||
if resolution is None:
|
||||
return None
|
||||
unavailable = resolution.availability == "temporarily_unavailable"
|
||||
if unavailable and not editor:
|
||||
return None
|
||||
if not content_visible_to_principal(selected.content, principal):
|
||||
return None
|
||||
subject = resolution.subject
|
||||
required_scopes = subject.required_scopes if subject is not None else ()
|
||||
if any(not _principal_has(principal, scope) for scope in required_scopes):
|
||||
return None
|
||||
return {
|
||||
"id": entry.id,
|
||||
"tenant_id": entry.tenant_id,
|
||||
"subject": entry_subject_reference(entry, revision=selected).to_dict(),
|
||||
"subject_stable_key": entry.subject_stable_key,
|
||||
"subject_resolution": resolution.to_dict(),
|
||||
"locale": entry.locale,
|
||||
"requested_locale": requested_locale or entry.locale,
|
||||
"locale_fallback": bool(requested_locale and requested_locale != entry.locale),
|
||||
"lifecycle_state": entry.lifecycle_state,
|
||||
"effective_state": selected.lifecycle_state,
|
||||
"pending_draft": bool(
|
||||
published is not None and current.id != published.id
|
||||
),
|
||||
"current_revision": entry.current_revision,
|
||||
"selected_revision": selected.revision,
|
||||
"published_revision": published.revision if published else None,
|
||||
"content": {} if unavailable else dict(selected.content),
|
||||
"content_redacted": unavailable,
|
||||
"content_hash": selected.content_hash,
|
||||
"subject_revision": selected.subject_revision,
|
||||
"subject_fingerprint": selected.subject_fingerprint,
|
||||
"authorship": {
|
||||
"created_by": entry.created_by,
|
||||
"updated_by": entry.updated_by,
|
||||
"authored_by": selected.authored_by,
|
||||
"reviewed_by": selected.reviewed_by,
|
||||
"published_by": entry.published_by,
|
||||
},
|
||||
"published_at": entry.published_at.isoformat() if entry.published_at else None,
|
||||
"retired_at": entry.retired_at.isoformat() if entry.retired_at else None,
|
||||
"superseded_by_entry_id": entry.superseded_by_entry_id,
|
||||
"created_at": entry.created_at.isoformat(),
|
||||
"updated_at": entry.updated_at.isoformat(),
|
||||
}
|
||||
|
||||
|
||||
def select_locale_entries(
|
||||
entries: Sequence[SemanticDocumentationEntry],
|
||||
*,
|
||||
locale: str,
|
||||
) -> tuple[SemanticDocumentationEntry, ...]:
|
||||
requested = _locale(locale)
|
||||
language = requested.split("-", 1)[0]
|
||||
grouped: dict[str, list[SemanticDocumentationEntry]] = {}
|
||||
for entry in entries:
|
||||
grouped.setdefault(entry.subject_stable_key, []).append(entry)
|
||||
selected: list[SemanticDocumentationEntry] = []
|
||||
for candidates in grouped.values():
|
||||
order = (
|
||||
requested,
|
||||
language,
|
||||
"de",
|
||||
"en",
|
||||
)
|
||||
candidate = next(
|
||||
(
|
||||
item
|
||||
for target in order
|
||||
for item in candidates
|
||||
if item.locale.casefold() == target.casefold()
|
||||
),
|
||||
sorted(candidates, key=lambda item: item.locale)[0],
|
||||
)
|
||||
selected.append(candidate)
|
||||
return tuple(
|
||||
sorted(
|
||||
selected,
|
||||
key=lambda item: (
|
||||
item.subject_module_id,
|
||||
item.subject_kind,
|
||||
item.subject_id,
|
||||
item.anchor_kind or "",
|
||||
item.anchor_id or "",
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def revision_payload(revision: SemanticDocumentationRevision) -> dict[str, object]:
|
||||
return {
|
||||
"id": revision.id,
|
||||
"entry_id": revision.entry_id,
|
||||
"revision": revision.revision,
|
||||
"lifecycle_state": revision.lifecycle_state,
|
||||
"action": revision.action,
|
||||
"change_reason": revision.change_reason,
|
||||
"content": dict(revision.content),
|
||||
"content_hash": revision.content_hash,
|
||||
"subject_revision": revision.subject_revision,
|
||||
"subject_fingerprint": revision.subject_fingerprint,
|
||||
"authored_by": revision.authored_by,
|
||||
"reviewed_by": revision.reviewed_by,
|
||||
"published_at": (
|
||||
revision.published_at.isoformat() if revision.published_at else None
|
||||
),
|
||||
"provenance": dict(revision.provenance or {}),
|
||||
"recoverable": revision.recoverable,
|
||||
"created_at": revision.created_at.isoformat(),
|
||||
}
|
||||
|
||||
|
||||
def _new_revision(
|
||||
entry: SemanticDocumentationEntry,
|
||||
*,
|
||||
revision: int,
|
||||
lifecycle_state: str,
|
||||
action: str,
|
||||
content: Mapping[str, object],
|
||||
subject: SemanticDocumentationSubjectReference,
|
||||
actor_id: str,
|
||||
change_reason: str,
|
||||
now: datetime | None,
|
||||
reviewer_id: str | None = None,
|
||||
) -> SemanticDocumentationRevision:
|
||||
normalized = _normalize_content(content)
|
||||
timestamp = _utc(now)
|
||||
return SemanticDocumentationRevision(
|
||||
tenant_id=entry.tenant_id,
|
||||
entry_id=entry.id,
|
||||
revision=revision,
|
||||
lifecycle_state=lifecycle_state,
|
||||
action=action,
|
||||
change_reason=_bounded_required(change_reason, "Change reason", 1000),
|
||||
content=normalized,
|
||||
content_hash=semantic_documentation_fingerprint(normalized),
|
||||
subject_revision=subject.observed_revision,
|
||||
subject_fingerprint=subject.observed_fingerprint,
|
||||
authored_by=actor_id,
|
||||
reviewed_by=reviewer_id,
|
||||
provenance={
|
||||
"subject_stable_key": subject.stable_key,
|
||||
"recorded_at": timestamp.isoformat(),
|
||||
"action_by": reviewer_id or actor_id,
|
||||
},
|
||||
recoverable=True,
|
||||
search_text=_search_text(normalized),
|
||||
)
|
||||
|
||||
|
||||
def _lifecycle_revision(
|
||||
entry: SemanticDocumentationEntry,
|
||||
*,
|
||||
current: SemanticDocumentationRevision,
|
||||
lifecycle_state: Literal["superseded", "retired"],
|
||||
action: str,
|
||||
actor_id: str,
|
||||
change_reason: str,
|
||||
now: datetime | None,
|
||||
provenance: Mapping[str, object] | None = None,
|
||||
) -> SemanticDocumentationRevision:
|
||||
timestamp = _utc(now)
|
||||
return SemanticDocumentationRevision(
|
||||
tenant_id=entry.tenant_id,
|
||||
entry_id=entry.id,
|
||||
revision=entry.current_revision + 1,
|
||||
lifecycle_state=lifecycle_state,
|
||||
action=action,
|
||||
change_reason=_bounded_required(change_reason, "Change reason", 1000),
|
||||
content=dict(current.content),
|
||||
content_hash=current.content_hash,
|
||||
subject_revision=current.subject_revision,
|
||||
subject_fingerprint=current.subject_fingerprint,
|
||||
authored_by=current.authored_by,
|
||||
reviewed_by=actor_id,
|
||||
provenance={
|
||||
**dict(current.provenance or {}),
|
||||
**dict(provenance or {}),
|
||||
"recorded_at": timestamp.isoformat(),
|
||||
"action_by": actor_id,
|
||||
},
|
||||
recoverable=True,
|
||||
search_text=current.search_text,
|
||||
)
|
||||
|
||||
|
||||
def _resolved_current_subject(
|
||||
session: Session,
|
||||
registry: object,
|
||||
principal: object,
|
||||
reference: SemanticDocumentationSubjectReference,
|
||||
) -> SemanticDocumentationSubjectReference:
|
||||
resolution = resolve_semantic_documentation_subject(
|
||||
registry,
|
||||
session,
|
||||
principal,
|
||||
reference=reference,
|
||||
)
|
||||
if resolution is None:
|
||||
raise SemanticDocumentationNotFoundError(
|
||||
"Semantic documentation subject not found."
|
||||
)
|
||||
if resolution.availability not in {"available", "changed"} or resolution.subject is None:
|
||||
raise SemanticDocumentationSubjectError(
|
||||
"Semantic documentation subject is not currently available."
|
||||
)
|
||||
return resolution.subject.reference
|
||||
|
||||
|
||||
def _normalize_content(content: Mapping[str, object]) -> dict[str, object]:
|
||||
allowed = {
|
||||
"title",
|
||||
"summary",
|
||||
"body",
|
||||
"meaning",
|
||||
"intended_use",
|
||||
"non_intended_use",
|
||||
"examples",
|
||||
"owner_account_id",
|
||||
"steward_account_id",
|
||||
"audience",
|
||||
"classification",
|
||||
"links",
|
||||
}
|
||||
unknown = sorted(str(key) for key in content if key not in allowed)
|
||||
if unknown:
|
||||
raise SemanticDocumentationError(
|
||||
"Unsupported semantic content fields: " + ", ".join(unknown)
|
||||
)
|
||||
normalized = {
|
||||
"title": _plain_text(content.get("title"), "Title", 500, required=True),
|
||||
"summary": _plain_text(content.get("summary"), "Summary", 4000),
|
||||
"body": _plain_text(content.get("body"), "Body", 200_000),
|
||||
"meaning": _plain_text(content.get("meaning"), "Meaning", 20_000),
|
||||
"intended_use": _plain_text(
|
||||
content.get("intended_use"), "Intended use", 20_000
|
||||
),
|
||||
"non_intended_use": _plain_text(
|
||||
content.get("non_intended_use"), "Non-intended use", 20_000
|
||||
),
|
||||
"examples": _string_list(content.get("examples"), "Examples", 50, 4000),
|
||||
"owner_account_id": _optional_id(content.get("owner_account_id")),
|
||||
"steward_account_id": _optional_id(content.get("steward_account_id")),
|
||||
"audience": _audience(content.get("audience")),
|
||||
"classification": str(content.get("classification") or "internal"),
|
||||
"links": _links(content.get("links")),
|
||||
}
|
||||
if normalized["classification"] not in {"internal", "restricted"}:
|
||||
raise SemanticDocumentationError(
|
||||
"Semantic content classification must be internal or restricted."
|
||||
)
|
||||
if normalized["classification"] == "restricted" and not normalized["audience"]:
|
||||
raise SemanticDocumentationError(
|
||||
"Restricted semantic documentation requires an audience."
|
||||
)
|
||||
return normalized
|
||||
|
||||
|
||||
def _plain_text(
|
||||
value: object,
|
||||
label: str,
|
||||
maximum: int,
|
||||
*,
|
||||
required: bool = False,
|
||||
) -> str:
|
||||
text = str(value or "").strip()
|
||||
if required and not text:
|
||||
raise SemanticDocumentationError(f"{label} is required.")
|
||||
if len(text) > maximum:
|
||||
raise SemanticDocumentationError(
|
||||
f"{label} is limited to {maximum} characters."
|
||||
)
|
||||
if any(ord(character) < 32 and character not in "\n\t" for character in text):
|
||||
raise SemanticDocumentationError(f"{label} contains control characters.")
|
||||
if "<script" in text.casefold() or "javascript:" in text.casefold():
|
||||
raise SemanticDocumentationError(f"{label} contains unsafe active content.")
|
||||
return text
|
||||
|
||||
|
||||
def _string_list(
|
||||
value: object,
|
||||
label: str,
|
||||
maximum_items: int,
|
||||
maximum_length: int,
|
||||
) -> list[str]:
|
||||
if value is None:
|
||||
return []
|
||||
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
|
||||
raise SemanticDocumentationError(f"{label} must be a list.")
|
||||
if len(value) > maximum_items:
|
||||
raise SemanticDocumentationError(
|
||||
f"{label} are limited to {maximum_items} items."
|
||||
)
|
||||
return list(
|
||||
dict.fromkeys(
|
||||
_plain_text(item, label, maximum_length, required=True) for item in value
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _audience(value: object) -> list[str]:
|
||||
selectors = _string_list(value, "Audience", 100, 500)
|
||||
prefixes = ("account:", "group:", "role:", "function:", "scope:")
|
||||
if any(
|
||||
selector != "authenticated" and not selector.startswith(prefixes)
|
||||
for selector in selectors
|
||||
):
|
||||
raise SemanticDocumentationError(
|
||||
"Semantic audience selectors must be typed."
|
||||
)
|
||||
return selectors
|
||||
|
||||
|
||||
def _links(value: object) -> list[dict[str, str]]:
|
||||
if value is None:
|
||||
return []
|
||||
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
|
||||
raise SemanticDocumentationError("Links must be a list.")
|
||||
if len(value) > 50:
|
||||
raise SemanticDocumentationError("Links are limited to 50 items.")
|
||||
result: list[dict[str, str]] = []
|
||||
for item in value:
|
||||
if not isinstance(item, Mapping) or set(item) != {"label", "href"}:
|
||||
raise SemanticDocumentationError(
|
||||
"Semantic links require only label and href."
|
||||
)
|
||||
label = _plain_text(item.get("label"), "Link label", 300, required=True)
|
||||
href = _plain_text(item.get("href"), "Link href", 2000, required=True)
|
||||
if not (
|
||||
(href.startswith("/") and not href.startswith("//"))
|
||||
or href.startswith("https://")
|
||||
):
|
||||
raise SemanticDocumentationError(
|
||||
"Semantic links must use HTTPS or a local path."
|
||||
)
|
||||
if href.startswith("https://") and "@" in href.split("/", 3)[2]:
|
||||
raise SemanticDocumentationError("Semantic links cannot contain credentials.")
|
||||
result.append({"label": label, "href": href})
|
||||
return result
|
||||
|
||||
|
||||
def _search_text(content: Mapping[str, object]) -> str:
|
||||
values = [
|
||||
content.get("title"),
|
||||
content.get("summary"),
|
||||
content.get("body"),
|
||||
content.get("meaning"),
|
||||
content.get("intended_use"),
|
||||
content.get("non_intended_use"),
|
||||
*(content.get("examples") or ()),
|
||||
]
|
||||
return "\n".join(str(value) for value in values if value).strip()
|
||||
|
||||
|
||||
def _principal_audience_tokens(principal: object) -> frozenset[str]:
|
||||
return frozenset(
|
||||
{
|
||||
"authenticated",
|
||||
f"account:{_principal_account_id(principal)}",
|
||||
*(f"group:{value}" for value in getattr(principal, "group_ids", ())),
|
||||
*(f"role:{value}" for value in getattr(principal, "role_ids", ())),
|
||||
*(
|
||||
f"function:{value}"
|
||||
for value in getattr(principal, "function_assignment_ids", ())
|
||||
),
|
||||
*(f"scope:{value}" for value in getattr(principal, "scopes", ())),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _principal_has(principal: object, scope: str) -> bool:
|
||||
checker = getattr(principal, "has", None)
|
||||
return bool(checker(scope)) if callable(checker) else scope in getattr(
|
||||
principal, "scopes", ()
|
||||
)
|
||||
|
||||
|
||||
def _principal_tenant_id(principal: object) -> str:
|
||||
tenant_id = str(getattr(principal, "tenant_id", "") or "").strip()
|
||||
if not tenant_id:
|
||||
raise SemanticDocumentationAuthorizationError(
|
||||
"Semantic documentation requires an active tenant."
|
||||
)
|
||||
return tenant_id
|
||||
|
||||
|
||||
def _principal_account_id(principal: object) -> str:
|
||||
account_id = str(getattr(principal, "account_id", "") or "").strip()
|
||||
if not account_id:
|
||||
raise SemanticDocumentationAuthorizationError(
|
||||
"Semantic documentation requires an authenticated account."
|
||||
)
|
||||
return account_id
|
||||
|
||||
|
||||
def _require_tenant(
|
||||
reference: SemanticDocumentationSubjectReference,
|
||||
tenant_id: str,
|
||||
) -> None:
|
||||
if reference.tenant_id != tenant_id:
|
||||
raise SemanticDocumentationNotFoundError(
|
||||
"Semantic documentation subject not found."
|
||||
)
|
||||
|
||||
|
||||
def _require_expected_revision(
|
||||
entry: SemanticDocumentationEntry,
|
||||
expected_revision: int,
|
||||
) -> None:
|
||||
if entry.current_revision != expected_revision:
|
||||
raise SemanticDocumentationConflictError(
|
||||
"Semantic documentation changed; reload before saving."
|
||||
)
|
||||
|
||||
|
||||
def _require_editable(entry: SemanticDocumentationEntry) -> None:
|
||||
if entry.lifecycle_state in {"superseded", "retired"}:
|
||||
raise SemanticDocumentationConflictError(
|
||||
f"{entry.lifecycle_state.title()} semantic documentation cannot be edited."
|
||||
)
|
||||
|
||||
|
||||
def _locale(value: str) -> str:
|
||||
clean = str(value or "").strip()
|
||||
if not _LOCALE_RE.fullmatch(clean):
|
||||
raise SemanticDocumentationError("Semantic documentation locale is invalid.")
|
||||
return clean
|
||||
|
||||
|
||||
def _optional_id(value: object) -> str | None:
|
||||
clean = str(value or "").strip()
|
||||
if not clean:
|
||||
return None
|
||||
if len(clean) > 255 or any(ord(character) < 32 for character in clean):
|
||||
raise SemanticDocumentationError("Account references must be bounded text.")
|
||||
return clean
|
||||
|
||||
|
||||
def _bounded_required(value: object, label: str, maximum: int) -> str:
|
||||
return _plain_text(value, label, maximum, required=True)
|
||||
|
||||
|
||||
def _utc(value: datetime | None) -> datetime:
|
||||
timestamp = value or datetime.now(UTC)
|
||||
if timestamp.tzinfo is None:
|
||||
return timestamp.replace(tzinfo=UTC)
|
||||
return timestamp.astimezone(UTC)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"SEMANTIC_LIFECYCLE_STATES",
|
||||
"SEMANTIC_PUBLICATION_MODES",
|
||||
"SEMANTIC_PUBLICATION_POLICY_KEY",
|
||||
"SemanticDocumentationAuthorizationError",
|
||||
"SemanticDocumentationConflictError",
|
||||
"SemanticDocumentationError",
|
||||
"SemanticDocumentationNotFoundError",
|
||||
"SemanticDocumentationSubjectError",
|
||||
"content_visible_to_principal",
|
||||
"create_semantic_entry",
|
||||
"current_revision",
|
||||
"entry_subject_reference",
|
||||
"get_semantic_entry",
|
||||
"list_semantic_entries",
|
||||
"publication_policy",
|
||||
"publish_semantic_entry",
|
||||
"published_revision",
|
||||
"resolve_entry_subject",
|
||||
"retire_semantic_entry",
|
||||
"revision_payload",
|
||||
"select_locale_entries",
|
||||
"semantic_entry_history",
|
||||
"semantic_entry_payload",
|
||||
"set_publication_policy",
|
||||
"supersede_semantic_entry",
|
||||
"update_semantic_entry",
|
||||
]
|
||||
Reference in New Issue
Block a user