feat: expose capability and catalog documentation

This commit is contained in:
2026-07-31 04:21:34 +02:00
parent af18f072d8
commit e2bf104c53
7 changed files with 555 additions and 44 deletions
-17
View File
@@ -195,23 +195,6 @@ manifest = ModuleManifest(
),
),
documentation_sources=(
DocumentationSourceDefinition(
id="docs.configuration.packages",
kind="configuration_package",
label="Configuration package catalog",
condition=DocumentationCondition(
any_scopes=("system:settings:read", "admin:settings:read"),
),
provenance={
"source": "core_configuration_package_contract",
"version": "1",
},
inspection={
"package_id": "govoplan.configuration-packages",
"schema_version": "1",
"description": "Signed, preflighted platform configuration packages.",
},
),
DocumentationSourceDefinition(
id="docs.project.wiki",
kind="wiki",
+231 -5
View File
@@ -58,18 +58,58 @@ class RouteInspection(_SourceModel):
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):
@@ -90,6 +130,7 @@ DocumentationSourceInspection = Annotated[
| RouteInspection
| CapabilityInspection
| PolicyInspection
| ReleaseCatalogInspection
| ConfigurationPackageInspection
| WikiInspection
| RepositoryInspection,
@@ -142,6 +183,8 @@ class DocumentationSourceRegistry:
def build_documentation_source_registry(
manifests: tuple[ModuleManifest, ...],
*,
include_runtime_catalogs: bool = True,
) -> DocumentationSourceRegistry:
registry = DocumentationSourceRegistry()
for manifest in manifests:
@@ -155,6 +198,9 @@ def build_documentation_source_registry(
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
@@ -203,21 +249,37 @@ def _route_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSou
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)
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)
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=f"{manifest.name} {kind} {capability}",
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)
@@ -240,6 +302,7 @@ def _defined_source(
condition=definition.condition,
documentation_types=definition.documentation_types,
state=definition.state,
state_reason=definition.state_reason,
configuration_key=definition.configuration_key,
)
@@ -249,8 +312,15 @@ def _defined_inspection(definition: DocumentationSourceDefinition) -> Documentat
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":
@@ -271,9 +341,23 @@ def _defined_inspection(definition: DocumentationSourceDefinition) -> Documentat
order=int(safe.get("order") or 0),
)
if definition.kind == "policy":
return PolicyInspection(capability=str(safe.get("capability") or definition.id))
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))
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),
@@ -319,6 +403,7 @@ def _registered_source(
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)
@@ -329,6 +414,7 @@ def _registered_source(
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"),
@@ -346,6 +432,130 @@ def _registered_source(
)
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, ...],
@@ -385,6 +595,22 @@ def _optional_text(value: object) -> str | None:
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 []