429 lines
14 KiB
Python
429 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from urllib.parse import quote
|
|
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.semantic_documentation import (
|
|
SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION,
|
|
SemanticDocumentationBreadcrumb,
|
|
SemanticDocumentationSubjectAnchor,
|
|
SemanticDocumentationSubjectDescriptor,
|
|
SemanticDocumentationSubjectPage,
|
|
SemanticDocumentationSubjectQuery,
|
|
SemanticDocumentationSubjectReference,
|
|
SemanticDocumentationSubjectResolution,
|
|
semantic_documentation_fingerprint,
|
|
)
|
|
from govoplan_core.security.redaction import redact_secret_values
|
|
from govoplan_workflow.backend.db.models import (
|
|
WorkflowDefinition,
|
|
WorkflowDefinitionRevision,
|
|
)
|
|
from govoplan_workflow.backend.governance import definition_decision
|
|
from govoplan_workflow.backend.service import (
|
|
get_definition_revision,
|
|
list_definition_revisions,
|
|
list_definitions,
|
|
)
|
|
|
|
|
|
SUBJECT_KIND = "workflow_definition"
|
|
ADMIN_SCOPE = "workflow:instance:admin"
|
|
DEFINITION_READ_SCOPE = "workflow:definition:read"
|
|
_MAX_SUBJECTS = 20_000
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _LineageState:
|
|
node_ids: Mapping[str, str]
|
|
historical_node_ids: frozenset[str]
|
|
|
|
|
|
class WorkflowSemanticDocumentationSubjectProvider:
|
|
provider_id = "workflow.semantic_subjects"
|
|
module_id = "workflow"
|
|
contract_version = SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION
|
|
|
|
def __init__(self, registry: object | None) -> None:
|
|
self._registry = registry
|
|
|
|
def list_subjects(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
request: SemanticDocumentationSubjectQuery,
|
|
) -> SemanticDocumentationSubjectPage:
|
|
actor = _principal(principal, tenant_id=request.tenant_id)
|
|
if actor is None:
|
|
return SemanticDocumentationSubjectPage()
|
|
if request.subject_kinds and SUBJECT_KIND not in request.subject_kinds:
|
|
return SemanticDocumentationSubjectPage()
|
|
db = _session(session)
|
|
subjects: list[SemanticDocumentationSubjectDescriptor] = []
|
|
for definition in list_definitions(db, tenant_id=request.tenant_id):
|
|
if not self._can_view(definition, actor):
|
|
continue
|
|
revision = get_definition_revision(db, definition=definition)
|
|
lineage = _lineage_state(db, definition)
|
|
subjects.extend(
|
|
_descriptors(
|
|
definition,
|
|
revision,
|
|
lineage,
|
|
tenant_id=request.tenant_id,
|
|
)
|
|
)
|
|
if len(subjects) > _MAX_SUBJECTS:
|
|
raise ValueError(
|
|
"Workflow semantic subject limit exceeded; narrow the query."
|
|
)
|
|
query = request.query.casefold().strip()
|
|
if query:
|
|
subjects = [
|
|
item
|
|
for item in subjects
|
|
if query in _search_text(item).casefold()
|
|
]
|
|
offset = _cursor_offset(request.cursor)
|
|
selected = tuple(subjects[offset : offset + request.limit])
|
|
next_offset = offset + len(selected)
|
|
has_more = next_offset < len(subjects)
|
|
return SemanticDocumentationSubjectPage(
|
|
subjects=selected,
|
|
next_cursor=str(next_offset) if has_more else None,
|
|
has_more=has_more,
|
|
)
|
|
|
|
def resolve_subject(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
reference: SemanticDocumentationSubjectReference,
|
|
) -> SemanticDocumentationSubjectResolution | None:
|
|
actor = _principal(principal, tenant_id=reference.tenant_id)
|
|
if (
|
|
reference.module_id != self.module_id
|
|
or reference.subject_kind != SUBJECT_KIND
|
|
or actor is None
|
|
):
|
|
return None
|
|
db = _session(session)
|
|
definition = db.scalar(
|
|
select(WorkflowDefinition).where(
|
|
WorkflowDefinition.id == reference.subject_id,
|
|
or_(
|
|
WorkflowDefinition.tenant_id == reference.tenant_id,
|
|
WorkflowDefinition.tenant_id.is_(None),
|
|
),
|
|
)
|
|
)
|
|
if definition is None or not self._can_view(definition, actor):
|
|
return None
|
|
if definition.deleted_at is not None:
|
|
return SemanticDocumentationSubjectResolution(
|
|
requested_reference=reference,
|
|
availability="missing",
|
|
reason_code="definition_deleted",
|
|
)
|
|
revision = get_definition_revision(db, definition=definition)
|
|
lineage = _lineage_state(db, definition)
|
|
descriptor = next(
|
|
(
|
|
item
|
|
for item in _descriptors(
|
|
definition,
|
|
revision,
|
|
lineage,
|
|
tenant_id=reference.tenant_id,
|
|
)
|
|
if item.reference.stable_key == reference.stable_key
|
|
),
|
|
None,
|
|
)
|
|
if descriptor is None:
|
|
anchor = reference.anchor
|
|
reason = "subject_missing"
|
|
if anchor is not None and anchor.kind == "step":
|
|
reason = (
|
|
"step_deleted"
|
|
if anchor.id in lineage.historical_node_ids
|
|
else "step_missing"
|
|
)
|
|
return SemanticDocumentationSubjectResolution(
|
|
requested_reference=reference,
|
|
availability="missing",
|
|
reason_code=reason,
|
|
)
|
|
changed = any(
|
|
expected is not None and expected != actual
|
|
for expected, actual in (
|
|
(reference.observed_revision, descriptor.reference.observed_revision),
|
|
(
|
|
reference.observed_fingerprint,
|
|
descriptor.reference.observed_fingerprint,
|
|
),
|
|
)
|
|
)
|
|
return SemanticDocumentationSubjectResolution(
|
|
requested_reference=reference,
|
|
availability="changed" if changed else "available",
|
|
subject=descriptor,
|
|
)
|
|
|
|
def _can_view(
|
|
self,
|
|
definition: WorkflowDefinition,
|
|
principal: ApiPrincipal,
|
|
) -> bool:
|
|
return definition_decision(
|
|
definition,
|
|
principal=principal,
|
|
registry=self._registry,
|
|
action="view",
|
|
).allowed
|
|
|
|
|
|
def _descriptors(
|
|
definition: WorkflowDefinition,
|
|
revision: WorkflowDefinitionRevision,
|
|
lineage: _LineageState,
|
|
*,
|
|
tenant_id: str,
|
|
) -> tuple[SemanticDocumentationSubjectDescriptor, ...]:
|
|
route = f"/workflow?definition={quote(definition.id, safe='')}"
|
|
definition_fingerprint = semantic_documentation_fingerprint(
|
|
{
|
|
"revision": definition.current_revision,
|
|
"content_hash": revision.content_hash,
|
|
"name": definition.name,
|
|
"description": definition.description,
|
|
"status": definition.status,
|
|
"active_revision": definition.active_revision,
|
|
"scope_type": definition.scope_type,
|
|
"definition_kind": definition.definition_kind,
|
|
"allow_start": definition.allow_start,
|
|
"allow_reuse": definition.allow_reuse,
|
|
"allow_automation": definition.allow_automation,
|
|
}
|
|
)
|
|
result = [
|
|
SemanticDocumentationSubjectDescriptor(
|
|
reference=_reference(
|
|
definition,
|
|
tenant_id=tenant_id,
|
|
revision=str(definition.current_revision),
|
|
fingerprint=definition_fingerprint,
|
|
),
|
|
labels={"en": definition.name},
|
|
descriptions=(
|
|
{"en": definition.description} if definition.description else {}
|
|
),
|
|
route=route,
|
|
# The provider already applies the engine's READ-or-ADMIN and
|
|
# per-definition governance decision. The descriptor contract
|
|
# represents an AND-only scope list, so it cannot restate that
|
|
# disjunction without incorrectly excluding administrators.
|
|
required_scopes=(),
|
|
)
|
|
]
|
|
graph = revision.graph if isinstance(revision.graph, Mapping) else {}
|
|
nodes = tuple(
|
|
item for item in graph.get("nodes", ()) if isinstance(item, Mapping)
|
|
)
|
|
edges = tuple(
|
|
item for item in graph.get("edges", ()) if isinstance(item, Mapping)
|
|
)
|
|
node_by_id = {str(item.get("id")): item for item in nodes if item.get("id")}
|
|
for node_id, node in node_by_id.items():
|
|
identity = lineage.node_ids[node_id]
|
|
fingerprint = _node_fingerprint(node, edges)
|
|
label = str(node.get("label") or node.get("type") or node_id)[:300]
|
|
breadcrumbs = [
|
|
SemanticDocumentationBreadcrumb(
|
|
label=definition.name,
|
|
subject_kind=SUBJECT_KIND,
|
|
subject_id=definition.id,
|
|
)
|
|
]
|
|
parent_id = str(node.get("parent_id") or "")
|
|
parent = node_by_id.get(parent_id)
|
|
if parent is not None and parent_id in lineage.node_ids:
|
|
breadcrumbs.append(
|
|
SemanticDocumentationBreadcrumb(
|
|
label=str(parent.get("label") or parent_id)[:300],
|
|
subject_kind=SUBJECT_KIND,
|
|
subject_id=definition.id,
|
|
anchor=SemanticDocumentationSubjectAnchor(
|
|
kind="step",
|
|
id=lineage.node_ids[parent_id],
|
|
),
|
|
)
|
|
)
|
|
result.append(
|
|
SemanticDocumentationSubjectDescriptor(
|
|
reference=_reference(
|
|
definition,
|
|
tenant_id=tenant_id,
|
|
anchor=SemanticDocumentationSubjectAnchor(
|
|
kind="step", id=identity
|
|
),
|
|
revision=fingerprint,
|
|
fingerprint=fingerprint,
|
|
),
|
|
labels={"en": label},
|
|
descriptions={
|
|
"en": f"Configured {str(node.get('type') or 'workflow step')[:200]} step."
|
|
},
|
|
breadcrumbs=tuple(breadcrumbs),
|
|
route=route,
|
|
route_anchor=_route_anchor(node_id),
|
|
required_scopes=(),
|
|
)
|
|
)
|
|
return tuple(result)
|
|
|
|
|
|
def _reference(
|
|
definition: WorkflowDefinition,
|
|
*,
|
|
tenant_id: str,
|
|
revision: str,
|
|
fingerprint: str,
|
|
anchor: SemanticDocumentationSubjectAnchor | None = None,
|
|
) -> SemanticDocumentationSubjectReference:
|
|
return SemanticDocumentationSubjectReference(
|
|
module_id="workflow",
|
|
tenant_id=tenant_id,
|
|
subject_kind=SUBJECT_KIND,
|
|
subject_id=definition.id,
|
|
anchor=anchor,
|
|
observed_revision=revision,
|
|
observed_fingerprint=fingerprint,
|
|
)
|
|
|
|
|
|
def _lineage_state(
|
|
session: Session,
|
|
definition: WorkflowDefinition,
|
|
) -> _LineageState:
|
|
revisions = sorted(
|
|
list_definition_revisions(session, definition=definition),
|
|
key=lambda item: item.revision,
|
|
)
|
|
active: dict[str, str] = {}
|
|
historical: set[str] = set()
|
|
for revision in revisions:
|
|
graph = revision.graph if isinstance(revision.graph, Mapping) else {}
|
|
node_ids = {
|
|
str(item.get("id"))
|
|
for item in graph.get("nodes", ())
|
|
if isinstance(item, Mapping) and item.get("id")
|
|
}
|
|
active = {key: value for key, value in active.items() if key in node_ids}
|
|
for node_id in sorted(node_ids):
|
|
active.setdefault(node_id, _lineage_id(revision.id, node_id))
|
|
historical.add(active[node_id])
|
|
return _LineageState(
|
|
node_ids=active,
|
|
historical_node_ids=frozenset(historical),
|
|
)
|
|
|
|
|
|
def _node_fingerprint(
|
|
node: Mapping[str, object],
|
|
edges: tuple[Mapping[str, object], ...],
|
|
) -> str:
|
|
node_id = str(node.get("id") or "")
|
|
connected = [
|
|
{
|
|
"id": edge.get("id"),
|
|
"type": edge.get("type"),
|
|
"label": edge.get("label"),
|
|
"source": edge.get("source"),
|
|
"target": edge.get("target"),
|
|
"source_port": edge.get("source_port"),
|
|
"target_port": edge.get("target_port"),
|
|
"config": redact_secret_values(edge.get("config") or {}),
|
|
}
|
|
for edge in edges
|
|
if node_id in {str(edge.get("source") or ""), str(edge.get("target") or "")}
|
|
]
|
|
connected.sort(key=lambda item: str(item["id"] or ""))
|
|
return semantic_documentation_fingerprint(
|
|
{
|
|
"type": node.get("type"),
|
|
"label": node.get("label"),
|
|
"parent_id": node.get("parent_id"),
|
|
"process_id": node.get("process_id"),
|
|
"config": redact_secret_values(node.get("config") or {}),
|
|
"connections": connected,
|
|
}
|
|
)
|
|
|
|
|
|
def _lineage_id(revision_id: str, node_id: str) -> str:
|
|
digest = hashlib.sha256(f"{revision_id}\x1f{node_id}".encode()).hexdigest()
|
|
return f"step-{digest[:40]}"
|
|
|
|
|
|
def _route_anchor(node_id: str) -> str:
|
|
route_anchor = f"workflow-node-{node_id}"
|
|
if len(route_anchor) <= 255:
|
|
return route_anchor
|
|
digest = hashlib.sha256(node_id.encode()).hexdigest()
|
|
return f"workflow-node-{digest[:40]}"
|
|
|
|
|
|
def _search_text(item: SemanticDocumentationSubjectDescriptor) -> str:
|
|
return " ".join(
|
|
(
|
|
item.reference.subject_id,
|
|
*item.labels.values(),
|
|
*item.descriptions.values(),
|
|
*(breadcrumb.label for breadcrumb in item.breadcrumbs),
|
|
)
|
|
)
|
|
|
|
|
|
def _principal(
|
|
principal: object,
|
|
*,
|
|
tenant_id: str,
|
|
) -> ApiPrincipal | None:
|
|
if not isinstance(principal, ApiPrincipal) or principal.tenant_id != tenant_id:
|
|
return None
|
|
if not (
|
|
principal.has(DEFINITION_READ_SCOPE)
|
|
or principal.has(ADMIN_SCOPE)
|
|
):
|
|
return None
|
|
return principal
|
|
|
|
|
|
def _cursor_offset(value: str | None) -> int:
|
|
if value is None:
|
|
return 0
|
|
if not value.isdigit() or int(value) < 0:
|
|
raise ValueError("Workflow semantic subject cursor is invalid.")
|
|
return int(value)
|
|
|
|
|
|
def _session(value: object) -> Session:
|
|
if not isinstance(value, Session):
|
|
raise TypeError("Workflow semantic subjects require a SQLAlchemy session.")
|
|
return value
|
|
|
|
|
|
__all__ = [
|
|
"SUBJECT_KIND",
|
|
"WorkflowSemanticDocumentationSubjectProvider",
|
|
]
|