617 lines
23 KiB
Python
617 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from dataclasses import dataclass
|
|
from typing import Annotated, Any, Literal, Mapping
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from govoplan_core.core.modules import (
|
|
DocumentationCondition,
|
|
DocumentationSourceDefinition,
|
|
DocumentationType,
|
|
ModuleManifest,
|
|
)
|
|
from govoplan_core.security.redaction import redact_secret_values
|
|
|
|
|
|
class _SourceModel(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
|
|
class DocumentationSourceProvenance(_SourceModel):
|
|
source: str
|
|
version: str | None = None
|
|
revision: str | None = None
|
|
published_at: str | None = None
|
|
checksum: str | None = None
|
|
|
|
|
|
class DocumentationSourceVisibility(_SourceModel):
|
|
documentation_types: list[DocumentationType]
|
|
required_modules: list[str] = Field(default_factory=list)
|
|
any_modules: list[str] = Field(default_factory=list)
|
|
missing_modules: list[str] = Field(default_factory=list)
|
|
required_capabilities: list[str] = Field(default_factory=list)
|
|
required_scopes: list[str] = Field(default_factory=list)
|
|
any_scopes: list[str] = Field(default_factory=list)
|
|
configuration_keys: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class ManifestInspection(_SourceModel):
|
|
kind: Literal["manifest"] = "manifest"
|
|
module_id: str
|
|
name: str
|
|
version: str
|
|
dependencies: list[str]
|
|
optional_dependencies: list[str]
|
|
|
|
|
|
class RouteInspection(_SourceModel):
|
|
kind: Literal["route"] = "route"
|
|
path: str
|
|
component: str | None = None
|
|
order: int
|
|
|
|
|
|
class CapabilityInspection(_SourceModel):
|
|
kind: Literal["capability"] = "capability"
|
|
capability: str
|
|
label: str | None = None
|
|
summary: str | None = None
|
|
contract_version: str | None = None
|
|
stability: Literal["experimental", "stable", "deprecated"] | None = None
|
|
audience: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class PolicyInspection(_SourceModel):
|
|
kind: Literal["policy"] = "policy"
|
|
capability: str
|
|
label: str | None = None
|
|
summary: str | None = None
|
|
contract_version: str | None = None
|
|
stability: Literal["experimental", "stable", "deprecated"] | None = None
|
|
audience: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class ReleaseCatalogEntryInspection(_SourceModel):
|
|
id: str
|
|
name: str
|
|
version: str | None = None
|
|
description: str | None = None
|
|
action: str | None = None
|
|
tags: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class ReleaseCatalogInspection(_SourceModel):
|
|
kind: Literal["release_catalog"] = "release_catalog"
|
|
catalog_type: Literal["modules", "configuration_packages"]
|
|
channel: str | None = None
|
|
sequence: int | None = None
|
|
generated_at: str | None = None
|
|
entry_count: int
|
|
signed: bool
|
|
trusted: bool
|
|
cache_used: bool
|
|
warnings: list[str] = Field(default_factory=list)
|
|
entries: list[ReleaseCatalogEntryInspection] = Field(default_factory=list)
|
|
|
|
|
|
class ConfigurationPackageInspection(_SourceModel):
|
|
kind: Literal["configuration_package"] = "configuration_package"
|
|
package_id: str
|
|
name: str | None = None
|
|
version: str | None = None
|
|
schema_version: str | None = None
|
|
description: str | None = None
|
|
publisher: str | None = None
|
|
category: str | None = None
|
|
tags: list[str] = Field(default_factory=list)
|
|
required_modules: list[str] = Field(default_factory=list)
|
|
required_capabilities: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class WikiInspection(_SourceModel):
|
|
kind: Literal["wiki"] = "wiki"
|
|
href: str
|
|
published_at: str | None = None
|
|
revision: str | None = None
|
|
|
|
|
|
class RepositoryInspection(_SourceModel):
|
|
kind: Literal["repository"] = "repository"
|
|
href: str
|
|
revision: str | None = None
|
|
|
|
|
|
DocumentationSourceInspection = Annotated[
|
|
ManifestInspection
|
|
| RouteInspection
|
|
| CapabilityInspection
|
|
| PolicyInspection
|
|
| ReleaseCatalogInspection
|
|
| ConfigurationPackageInspection
|
|
| WikiInspection
|
|
| RepositoryInspection,
|
|
Field(discriminator="kind"),
|
|
]
|
|
|
|
|
|
class DocumentationSourceItem(_SourceModel):
|
|
id: str
|
|
kind: str
|
|
owner_module_id: str
|
|
label: str
|
|
state: Literal["configured", "disabled", "unavailable"]
|
|
state_reason: str | None = None
|
|
inspection_url: str
|
|
provenance: DocumentationSourceProvenance
|
|
visibility: DocumentationSourceVisibility
|
|
inspection: DocumentationSourceInspection
|
|
|
|
def summary(self) -> dict[str, Any]:
|
|
return self.model_dump(mode="json", exclude={"inspection"})
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RegisteredDocumentationSource:
|
|
item: DocumentationSourceItem
|
|
condition: DocumentationCondition
|
|
documentation_types: tuple[DocumentationType, ...]
|
|
configuration_key: str | None = None
|
|
|
|
|
|
class DocumentationSourceRegistry:
|
|
def __init__(self) -> None:
|
|
self._sources: dict[str, RegisteredDocumentationSource] = {}
|
|
|
|
def register(self, source: RegisteredDocumentationSource) -> None:
|
|
if source.item.id in self._sources:
|
|
raise ValueError(f"Duplicate documentation source id: {source.item.id}")
|
|
self._sources[source.item.id] = source
|
|
|
|
def sources(self) -> tuple[RegisteredDocumentationSource, ...]:
|
|
return tuple(
|
|
self._sources[source_id]
|
|
for source_id in sorted(self._sources)
|
|
)
|
|
|
|
def get(self, source_id: str) -> RegisteredDocumentationSource | None:
|
|
return self._sources.get(source_id)
|
|
|
|
|
|
def build_documentation_source_registry(
|
|
manifests: tuple[ModuleManifest, ...],
|
|
*,
|
|
include_runtime_catalogs: bool = True,
|
|
) -> DocumentationSourceRegistry:
|
|
registry = DocumentationSourceRegistry()
|
|
for manifest in manifests:
|
|
registry.register(_manifest_source(manifest))
|
|
for source in _route_sources(manifest):
|
|
registry.register(source)
|
|
for source in _capability_sources(manifest):
|
|
registry.register(source)
|
|
for definition in manifest.documentation_sources:
|
|
registry.register(_defined_source(manifest, definition))
|
|
for source in _linked_sources(manifest):
|
|
if registry.get(source.item.id) is None:
|
|
registry.register(source)
|
|
if include_runtime_catalogs:
|
|
for source in _catalog_sources():
|
|
registry.register(source)
|
|
return registry
|
|
|
|
|
|
def _manifest_source(manifest: ModuleManifest) -> RegisteredDocumentationSource:
|
|
return _registered_source(
|
|
source_id=f"{manifest.id}.manifest",
|
|
kind="manifest",
|
|
owner_module_id=manifest.id,
|
|
label=f"{manifest.name} module manifest",
|
|
provenance={"source": "module_manifest", "version": manifest.version},
|
|
inspection=ManifestInspection(
|
|
module_id=manifest.id,
|
|
name=manifest.name,
|
|
version=manifest.version,
|
|
dependencies=list(manifest.dependencies),
|
|
optional_dependencies=list(manifest.optional_dependencies),
|
|
),
|
|
)
|
|
|
|
|
|
def _route_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]:
|
|
if manifest.frontend is None:
|
|
return ()
|
|
return tuple(
|
|
_registered_source(
|
|
source_id=_derived_source_id(manifest.id, "route", route.path),
|
|
kind="route",
|
|
owner_module_id=manifest.id,
|
|
label=f"{manifest.name} route {route.path}",
|
|
provenance={"source": "frontend_manifest", "version": manifest.version},
|
|
inspection=RouteInspection(
|
|
path=route.path,
|
|
component=route.component,
|
|
order=route.order,
|
|
),
|
|
condition=DocumentationCondition(
|
|
required_modules=(manifest.id,),
|
|
required_scopes=route.required_all,
|
|
any_scopes=route.required_any,
|
|
),
|
|
)
|
|
for route in manifest.frontend.routes
|
|
)
|
|
|
|
|
|
def _capability_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]:
|
|
sources: list[RegisteredDocumentationSource] = []
|
|
for capability in sorted(manifest.capability_factories):
|
|
metadata = manifest.capability_documentation.get(capability)
|
|
is_policy = capability.startswith("policy.")
|
|
kind = "policy" if is_policy else "capability"
|
|
inspection: DocumentationSourceInspection = (
|
|
PolicyInspection(
|
|
capability=capability,
|
|
label=metadata.label if metadata else None,
|
|
summary=metadata.summary if metadata else None,
|
|
contract_version=metadata.contract_version if metadata else None,
|
|
stability=metadata.stability if metadata else None,
|
|
audience=list(metadata.audience) if metadata else [],
|
|
)
|
|
if is_policy
|
|
else CapabilityInspection(
|
|
capability=capability,
|
|
label=metadata.label if metadata else None,
|
|
summary=metadata.summary if metadata else None,
|
|
contract_version=metadata.contract_version if metadata else None,
|
|
stability=metadata.stability if metadata else None,
|
|
audience=list(metadata.audience) if metadata else [],
|
|
)
|
|
)
|
|
sources.append(_registered_source(
|
|
source_id=_derived_source_id(manifest.id, kind, capability),
|
|
kind=kind,
|
|
owner_module_id=manifest.id,
|
|
label=metadata.label if metadata else f"{manifest.name} {kind} {capability}",
|
|
provenance={"source": "module_manifest", "version": manifest.version},
|
|
inspection=inspection,
|
|
condition=DocumentationCondition(required_modules=(manifest.id,)),
|
|
documentation_types=metadata.documentation_types if metadata else ("admin",),
|
|
))
|
|
return tuple(sources)
|
|
|
|
|
|
def _defined_source(
|
|
manifest: ModuleManifest,
|
|
definition: DocumentationSourceDefinition,
|
|
) -> RegisteredDocumentationSource:
|
|
inspection = _defined_inspection(definition)
|
|
return _registered_source(
|
|
source_id=definition.id,
|
|
kind=definition.kind,
|
|
owner_module_id=manifest.id,
|
|
label=definition.label,
|
|
provenance={
|
|
"source": str(definition.provenance.get("source") or "module_manifest"),
|
|
**_safe_source_fields(definition.provenance),
|
|
},
|
|
inspection=inspection,
|
|
condition=definition.condition,
|
|
documentation_types=definition.documentation_types,
|
|
state=definition.state,
|
|
state_reason=definition.state_reason,
|
|
configuration_key=definition.configuration_key,
|
|
)
|
|
|
|
|
|
def _defined_inspection(definition: DocumentationSourceDefinition) -> DocumentationSourceInspection:
|
|
safe = _safe_source_fields(definition.inspection)
|
|
if definition.kind == "configuration_package":
|
|
return ConfigurationPackageInspection(
|
|
package_id=str(safe.get("package_id") or definition.id),
|
|
name=_optional_text(safe.get("name")),
|
|
version=_optional_text(safe.get("version")),
|
|
schema_version=_optional_text(safe.get("schema_version")),
|
|
description=_optional_text(safe.get("description")),
|
|
publisher=_optional_text(safe.get("publisher")),
|
|
category=_optional_text(safe.get("category")),
|
|
tags=_string_list(safe.get("tags")),
|
|
required_modules=_string_list(safe.get("required_modules")),
|
|
required_capabilities=_string_list(safe.get("required_capabilities")),
|
|
)
|
|
href = definition.link.href if definition.link else str(safe.get("href") or "")
|
|
if definition.kind == "wiki":
|
|
return WikiInspection(
|
|
href=href,
|
|
published_at=_optional_text(safe.get("published_at")),
|
|
revision=_optional_text(safe.get("revision")),
|
|
)
|
|
if definition.kind == "repository":
|
|
return RepositoryInspection(
|
|
href=href,
|
|
revision=_optional_text(safe.get("revision")),
|
|
)
|
|
if definition.kind == "route":
|
|
return RouteInspection(
|
|
path=str(safe.get("path") or ""),
|
|
component=_optional_text(safe.get("component")),
|
|
order=int(safe.get("order") or 0),
|
|
)
|
|
if definition.kind == "policy":
|
|
return PolicyInspection(
|
|
capability=str(safe.get("capability") or definition.id),
|
|
label=_optional_text(safe.get("label")),
|
|
summary=_optional_text(safe.get("summary")),
|
|
contract_version=_optional_text(safe.get("contract_version")),
|
|
stability=_capability_stability(safe.get("stability")),
|
|
audience=_string_list(safe.get("audience")),
|
|
)
|
|
if definition.kind == "capability":
|
|
return CapabilityInspection(
|
|
capability=str(safe.get("capability") or definition.id),
|
|
label=_optional_text(safe.get("label")),
|
|
summary=_optional_text(safe.get("summary")),
|
|
contract_version=_optional_text(safe.get("contract_version")),
|
|
stability=_capability_stability(safe.get("stability")),
|
|
audience=_string_list(safe.get("audience")),
|
|
)
|
|
return ManifestInspection(
|
|
module_id=str(safe.get("module_id") or definition.id.split(".", 1)[0]),
|
|
name=str(safe.get("name") or definition.label),
|
|
version=str(safe.get("version") or ""),
|
|
dependencies=_string_list(safe.get("dependencies")),
|
|
optional_dependencies=_string_list(safe.get("optional_dependencies")),
|
|
)
|
|
|
|
|
|
def _linked_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]:
|
|
sources: list[RegisteredDocumentationSource] = []
|
|
for topic in manifest.documentation:
|
|
for link in topic.links:
|
|
if link.kind not in {"wiki", "repository"}:
|
|
continue
|
|
source_id = _derived_source_id(manifest.id, link.kind, link.href)
|
|
inspection: DocumentationSourceInspection = (
|
|
WikiInspection(href=link.href)
|
|
if link.kind == "wiki"
|
|
else RepositoryInspection(href=link.href)
|
|
)
|
|
sources.append(_registered_source(
|
|
source_id=source_id,
|
|
kind=link.kind,
|
|
owner_module_id=manifest.id,
|
|
label=link.label,
|
|
provenance={"source": "documentation_link", "version": manifest.version},
|
|
inspection=inspection,
|
|
condition=topic.conditions[0] if len(topic.conditions) == 1 else DocumentationCondition(),
|
|
documentation_types=topic.documentation_types,
|
|
))
|
|
return tuple(sources)
|
|
|
|
|
|
def _registered_source(
|
|
*,
|
|
source_id: str,
|
|
kind: str,
|
|
owner_module_id: str,
|
|
label: str,
|
|
provenance: Mapping[str, Any],
|
|
inspection: DocumentationSourceInspection,
|
|
condition: DocumentationCondition | None = None,
|
|
documentation_types: tuple[DocumentationType, ...] = ("admin",),
|
|
state: Literal["configured", "disabled", "unavailable"] = "configured",
|
|
state_reason: str | None = None,
|
|
configuration_key: str | None = None,
|
|
) -> RegisteredDocumentationSource:
|
|
clean_provenance = _safe_source_fields(provenance)
|
|
return RegisteredDocumentationSource(
|
|
item=DocumentationSourceItem(
|
|
id=source_id,
|
|
kind=kind,
|
|
owner_module_id=owner_module_id,
|
|
label=label,
|
|
state=state,
|
|
state_reason=state_reason,
|
|
inspection_url=f"/api/v1/docs/sources/{source_id}",
|
|
provenance=DocumentationSourceProvenance(
|
|
source=str(clean_provenance.get("source") or "unknown"),
|
|
version=_optional_text(clean_provenance.get("version")),
|
|
revision=_optional_text(clean_provenance.get("revision")),
|
|
published_at=_optional_text(clean_provenance.get("published_at")),
|
|
checksum=_optional_text(clean_provenance.get("checksum")),
|
|
),
|
|
visibility=_visibility_payload(condition or DocumentationCondition(), documentation_types),
|
|
inspection=inspection,
|
|
),
|
|
condition=condition or DocumentationCondition(),
|
|
documentation_types=documentation_types,
|
|
configuration_key=configuration_key,
|
|
)
|
|
|
|
|
|
def _catalog_sources() -> tuple[RegisteredDocumentationSource, ...]:
|
|
from govoplan_core.core.configuration_packages import validate_configuration_package_catalog
|
|
from govoplan_core.core.module_package_catalog import validate_module_package_catalog
|
|
|
|
return (
|
|
*_release_catalog_sources(
|
|
"modules",
|
|
"Module release catalog",
|
|
validate_module_package_catalog(),
|
|
),
|
|
*_release_catalog_sources(
|
|
"configuration_packages",
|
|
"Configuration package catalog",
|
|
validate_configuration_package_catalog(),
|
|
),
|
|
)
|
|
|
|
|
|
def _release_catalog_sources(
|
|
catalog_type: Literal["modules", "configuration_packages"],
|
|
label: str,
|
|
validation: Mapping[str, object],
|
|
) -> tuple[RegisteredDocumentationSource, ...]:
|
|
entry_key = "modules" if catalog_type == "modules" else "packages"
|
|
raw_entries = validation.get(entry_key)
|
|
entries = [item for item in raw_entries if isinstance(item, Mapping)] if isinstance(raw_entries, list) else []
|
|
configured = bool(validation.get("configured"))
|
|
valid = bool(validation.get("valid"))
|
|
state: Literal["configured", "disabled", "unavailable"] = (
|
|
"configured" if configured and valid else "unavailable" if configured else "disabled"
|
|
)
|
|
reason = _optional_text(validation.get("error"))
|
|
if state == "disabled":
|
|
reason = "No catalog source is configured."
|
|
catalog_id = f"docs.release-catalog.{catalog_type.replace('_', '-')}"
|
|
catalog_source = _registered_source(
|
|
source_id=catalog_id,
|
|
kind="release_catalog",
|
|
owner_module_id="docs",
|
|
label=label,
|
|
provenance={
|
|
"source": "core_catalog_contract",
|
|
"published_at": validation.get("generated_at"),
|
|
},
|
|
inspection=ReleaseCatalogInspection(
|
|
catalog_type=catalog_type,
|
|
channel=_optional_text(validation.get("channel")),
|
|
sequence=_optional_int(validation.get("sequence")),
|
|
generated_at=_optional_text(validation.get("generated_at")),
|
|
entry_count=len(entries),
|
|
signed=bool(validation.get("signed")),
|
|
trusted=bool(validation.get("trusted")),
|
|
cache_used=bool(validation.get("cache_used")),
|
|
warnings=_string_list(validation.get("warnings")),
|
|
entries=[_release_catalog_entry(item, catalog_type) for item in entries],
|
|
),
|
|
state=state,
|
|
state_reason=reason,
|
|
)
|
|
if catalog_type == "modules" or not valid:
|
|
return (catalog_source,)
|
|
return (
|
|
catalog_source,
|
|
*(
|
|
_configuration_package_source(item, validation)
|
|
for item in entries
|
|
),
|
|
)
|
|
|
|
|
|
def _release_catalog_entry(
|
|
item: Mapping[str, object],
|
|
catalog_type: Literal["modules", "configuration_packages"],
|
|
) -> ReleaseCatalogEntryInspection:
|
|
item_id = (
|
|
_optional_text(item.get("module_id"))
|
|
if catalog_type == "modules"
|
|
else _optional_text(item.get("package_id"))
|
|
)
|
|
return ReleaseCatalogEntryInspection(
|
|
id=item_id or "unknown",
|
|
name=_optional_text(item.get("name")) or item_id or "Unknown",
|
|
version=_optional_text(item.get("version")),
|
|
description=_optional_text(item.get("description")),
|
|
action=_optional_text(item.get("action")),
|
|
tags=_string_list(item.get("tags")),
|
|
)
|
|
|
|
|
|
def _configuration_package_source(
|
|
item: Mapping[str, object],
|
|
validation: Mapping[str, object],
|
|
) -> RegisteredDocumentationSource:
|
|
package_id = _optional_text(item.get("package_id")) or "unknown"
|
|
required_modules = [
|
|
str(requirement.get("module_id"))
|
|
for requirement in item.get("required_modules", ())
|
|
if isinstance(requirement, Mapping) and requirement.get("module_id")
|
|
] if isinstance(item.get("required_modules"), (list, tuple)) else []
|
|
return _registered_source(
|
|
source_id=_derived_source_id("docs", "configuration-package", package_id),
|
|
kind="configuration_package",
|
|
owner_module_id="docs",
|
|
label=_optional_text(item.get("name")) or package_id,
|
|
provenance={
|
|
"source": "configuration_package_catalog",
|
|
"version": item.get("version"),
|
|
"published_at": validation.get("generated_at"),
|
|
},
|
|
inspection=ConfigurationPackageInspection(
|
|
package_id=package_id,
|
|
name=_optional_text(item.get("name")),
|
|
version=_optional_text(item.get("version")),
|
|
schema_version=_optional_text(item.get("schema_version")),
|
|
description=_optional_text(item.get("description")),
|
|
publisher=_optional_text(item.get("publisher")),
|
|
category=_optional_text(item.get("category")),
|
|
tags=_string_list(item.get("tags")),
|
|
required_modules=required_modules,
|
|
required_capabilities=_string_list(item.get("required_capabilities")),
|
|
),
|
|
)
|
|
|
|
|
|
def _visibility_payload(
|
|
condition: DocumentationCondition,
|
|
documentation_types: tuple[DocumentationType, ...],
|
|
) -> DocumentationSourceVisibility:
|
|
return DocumentationSourceVisibility(
|
|
documentation_types=list(documentation_types),
|
|
required_modules=list(condition.required_modules),
|
|
any_modules=list(condition.any_modules),
|
|
missing_modules=list(condition.missing_modules),
|
|
required_capabilities=list(condition.required_capabilities),
|
|
required_scopes=list(condition.required_scopes),
|
|
any_scopes=list(condition.any_scopes),
|
|
configuration_keys=list(condition.configuration_keys),
|
|
)
|
|
|
|
|
|
def _derived_source_id(module_id: str, kind: str, value: str) -> str:
|
|
digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
|
|
return f"{module_id}.{kind}.{digest}"
|
|
|
|
|
|
def _safe_source_fields(value: Mapping[str, Any]) -> dict[str, Any]:
|
|
redacted = redact_secret_values(dict(value))
|
|
if not isinstance(redacted, Mapping):
|
|
return {}
|
|
return {
|
|
str(key): item
|
|
for key, item in redacted.items()
|
|
if isinstance(item, (str, int, float, bool, list, tuple)) or item is None
|
|
}
|
|
|
|
|
|
def _optional_text(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text[:2_000] if text else None
|
|
|
|
|
|
def _optional_int(value: object) -> int | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _capability_stability(
|
|
value: object,
|
|
) -> Literal["experimental", "stable", "deprecated"] | None:
|
|
text = _optional_text(value)
|
|
return text if text in {"experimental", "stable", "deprecated"} else None
|
|
|
|
|
|
def _string_list(value: object) -> list[str]:
|
|
if not isinstance(value, (list, tuple)):
|
|
return []
|
|
return [str(item)[:500] for item in value[:100]]
|