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

feat: complete governed documentation sources

Browse Source
This commit is contained in:
zemion
2026-07-30 14:26:43 +02:00
parent 511930ea56
commit da15645dba
6 changed files with 1104 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_docs/backend/api/v1/routes.py
+431 -28
View File
@@ -7,6 +7,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from govoplan_core.auth import ApiPrincipal, has_scope, require_any_scope
from govoplan_core.core.modules import (
DocumentationConfigurationDecision,
DocumentationCondition,
DocumentationContext,
DocumentationLink,
@@ -22,10 +23,18 @@ from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.db.session import get_database
from govoplan_docs.backend.manifest import DOCS_ADMIN_READ_SCOPES, DOCS_READ_SCOPES
from govoplan_docs.backend.sources import (
RegisteredDocumentationSource,
build_documentation_source_registry,
)
router = APIRouter(prefix="/docs", tags=["docs"])
TOPIC_KINDS = ("workflow", "reference", "pattern", "system")
_CONFIGURATION_ACTIVE_STATES = frozenset({"enabled", "inherited"})
_CONFIGURATION_STATES = frozenset(
{"enabled", "disabled", "inherited", "unavailable"}
)
@router.get("/context")
@@ -46,6 +55,17 @@ def docs_context(
route_items = _route_items(registry.manifests(), principal)
visible_route_items = [item for item in route_items if item["visible"]]
documentation_layers = _documentation_layers(request, registry, principal, documentation_type=documentation_type, locale=resolved_locale)
evidence_sources = (
_documentation_source_summaries(
request,
registry,
principal,
documentation_type=documentation_type,
locale=resolved_locale,
)
if documentation_type == "admin"
else []
)
if documentation_type == "admin":
catalog = _admin_documentation_catalog(
registry,
@@ -76,11 +96,81 @@ def docs_context(
"layers": _documentation_layer_payload(
catalog,
documentation_layers=documentation_layers,
include_evidence_sources=documentation_type == "admin",
evidence_sources=evidence_sources,
),
}
@router.get("/sources")
def list_documentation_sources(
request: Request,
documentation_type: DocumentationType = Query(
default="admin",
alias="type",
pattern="^(admin|user)$",
),
locale: str | None = Query(default=None, min_length=2, max_length=20),
principal: ApiPrincipal = Depends(require_any_scope(*DOCS_READ_SCOPES)),
) -> dict[str, Any]:
_require_documentation_type_access(principal, documentation_type)
registry = _registry(request)
items = _documentation_source_items(
request,
registry,
principal,
documentation_type=documentation_type,
locale=_preferred_locale(request, locale),
)
return {
"items": [source.item.summary() for source in items],
"total": len(items),
}
@router.get("/sources/{source_id}")
def inspect_documentation_source(
source_id: str,
request: Request,
documentation_type: DocumentationType = Query(
default="admin",
alias="type",
pattern="^(admin|user)$",
),
locale: str | None = Query(default=None, min_length=2, max_length=20),
principal: ApiPrincipal = Depends(require_any_scope(*DOCS_READ_SCOPES)),
) -> dict[str, Any]:
_require_documentation_type_access(principal, documentation_type)
registry = _registry(request)
items = _documentation_source_items(
request,
registry,
principal,
documentation_type=documentation_type,
locale=_preferred_locale(request, locale),
)
source = next((item for item in items if item.item.id == source_id), None)
if source is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Documentation source not found",
)
return source.item.model_dump(mode="json")
def _require_documentation_type_access(
principal: ApiPrincipal,
documentation_type: DocumentationType,
) -> None:
if documentation_type == "admin" and not _has_any_scope(
principal,
DOCS_ADMIN_READ_SCOPES,
):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Administrative documentation requires documentation-administrator authority",
)
def _admin_documentation_catalog(
registry: PlatformRegistry,
principal: ApiPrincipal,
@@ -186,7 +276,7 @@ def _documentation_layer_payload(
catalog: Mapping[str, list[dict[str, Any]]],
*,
documentation_layers: Mapping[str, list[dict[str, Any]]],
include_evidence_sources: bool,
evidence_sources: list[dict[str, Any]],
) -> dict[str, dict[str, object]]:
permissions = catalog["permissions"]
return {
@@ -206,7 +296,7 @@ def _documentation_layer_payload(
},
"evidence": {
"optional_modules": catalog["optional_modules"],
"sources": _evidence_sources() if include_evidence_sources else [],
"sources": evidence_sources,
"documentation": documentation_layers["evidence"],
},
}
@@ -398,6 +488,138 @@ def _settings(request: Request) -> object | None:
return getattr(lifecycle, "settings", None)
def _documentation_source_summaries(
request: Request,
registry: PlatformRegistry,
principal: ApiPrincipal,
*,
documentation_type: DocumentationType,
locale: str,
) -> list[dict[str, Any]]:
return [
source.item.summary()
for source in _documentation_source_items(
request,
registry,
principal,
documentation_type=documentation_type,
locale=locale,
)
]
def _documentation_source_items(
request: Request,
registry: PlatformRegistry,
principal: ApiPrincipal,
*,
documentation_type: DocumentationType,
locale: str,
) -> list[RegisteredDocumentationSource]:
source_registry = build_documentation_source_registry(registry.manifests())
settings = _settings(request)
try:
with get_database().SessionLocal() as session:
return _visible_documentation_sources(
source_registry.sources(),
registry,
principal,
settings=settings,
session=session,
documentation_type=documentation_type,
locale=locale,
)
except RuntimeError:
return _visible_documentation_sources(
source_registry.sources(),
registry,
principal,
settings=settings,
session=None,
documentation_type=documentation_type,
locale=locale,
)
def _visible_documentation_sources(
sources: tuple[RegisteredDocumentationSource, ...],
registry: PlatformRegistry,
principal: ApiPrincipal,
*,
settings: object | None,
session: object | None,
documentation_type: DocumentationType,
locale: str,
) -> list[RegisteredDocumentationSource]:
installed = {manifest.id for manifest in registry.manifests()}
context = DocumentationContext(
registry=registry,
principal=principal,
settings=settings,
session=session,
documentation_type=documentation_type,
locale=locale,
)
visible: list[RegisteredDocumentationSource] = []
for source in sources:
if documentation_type not in source.documentation_types:
continue
keys = tuple(dict.fromkeys((
*source.condition.configuration_keys,
*([source.configuration_key] if source.configuration_key else []),
)))
configuration = _resolve_documentation_configuration(
registry,
source.item.owner_module_id,
keys,
context=context,
)
active, reason, blockers = _condition_visibility(
source.condition,
installed,
registry,
principal,
configuration=configuration,
)
if blockers["scopes"]:
continue
state = source.item.state
source_configuration = (
configuration.get(source.configuration_key)
if source.configuration_key
else None
)
if state == "configured" and source_configuration is not None:
if source_configuration.state == "unavailable":
state = "unavailable"
reason = source_configuration.reason or reason
elif source_configuration.state == "disabled":
state = "disabled"
reason = source_configuration.reason or reason
if state == "configured" and not active:
state = (
"unavailable"
if any(
decision.state == "unavailable"
for decision in configuration.values()
)
or blockers["modules"]
or blockers["capabilities"]
else "disabled"
)
item = source.item.model_copy(update={
"state": state,
"state_reason": None if state == "configured" else reason,
})
visible.append(RegisteredDocumentationSource(
item=item,
condition=source.condition,
documentation_types=source.documentation_types,
configuration_key=source.configuration_key,
))
return visible
def _classify_documentation(
registry: PlatformRegistry,
principal: ApiPrincipal,
@@ -417,14 +639,40 @@ def _classify_documentation(
if item["visible"]
),
]) if documentation_type == "user" else frozenset()
context = DocumentationContext(
registry=registry,
principal=principal,
settings=settings,
session=session,
documentation_type=documentation_type,
locale=locale,
)
topics = _collect_documentation_topics(registry, principal, settings=settings, session=session, documentation_type=documentation_type, locale=locale)
for source_module_id, topic in sorted(topics, key=lambda item: (item[1].order, item[0], item[1].id)):
if not _topic_matches_documentation_type(topic, documentation_type):
continue
active, reason, blockers = _documentation_visibility(topic, installed, registry, principal)
configuration_keys = _documentation_configuration_keys(topic)
configuration = _resolve_documentation_configuration(
registry,
topic.source_module_id or source_module_id,
configuration_keys,
context=context,
)
active, reason, blockers = _documentation_visibility(
topic,
installed,
registry,
principal,
configuration=configuration,
)
if documentation_type == "user" and not active:
continue
target_layer = _documentation_target_layer(topic, active, blockers)
target_layer = _documentation_target_layer(
topic,
active,
blockers,
configuration=configuration,
)
if target_layer not in layers:
target_layer = "evidence"
layers[target_layer].append(_documentation_topic_payload(
@@ -437,6 +685,7 @@ def _classify_documentation(
locale=locale,
documentation_type=documentation_type,
visible_runtime_paths=visible_runtime_paths,
configuration=configuration,
))
return layers
@@ -494,21 +743,130 @@ def _topic_matches_documentation_type(topic: DocumentationTopic, documentation_t
return documentation_type in (topic.documentation_types or ("admin",))
def _documentation_configuration_keys(
topic: DocumentationTopic,
) -> tuple[str, ...]:
return tuple(dict.fromkeys((
*topic.configuration_keys,
*(
key
for condition in topic.conditions
for key in condition.configuration_keys
),
)))
def _resolve_documentation_configuration(
registry: PlatformRegistry,
module_id: str,
keys: tuple[str, ...],
*,
context: DocumentationContext,
) -> dict[str, DocumentationConfigurationDecision]:
decisions = {
key: DocumentationConfigurationDecision(
key=key,
state="unavailable",
reason="No configuration-state provider is registered.",
)
for key in keys
}
manifest = registry.get(module_id)
if manifest is None or not keys:
return decisions
for registration in manifest.documentation_configuration_providers:
requested = tuple(key for key in keys if key in registration.keys)
if not requested:
continue
try:
provided = registration.resolve(context, requested)
except Exception as exc:
for key in requested:
decisions[key] = DocumentationConfigurationDecision(
key=key,
state="unavailable",
reason=f"Configuration-state provider failed ({type(exc).__name__}).",
)
continue
for key in requested:
decision = provided.get(key)
if (
not isinstance(decision, DocumentationConfigurationDecision)
or decision.key != key
or decision.state not in _CONFIGURATION_STATES
):
decisions[key] = DocumentationConfigurationDecision(
key=key,
state="unavailable",
reason="Configuration-state provider returned an invalid decision.",
)
continue
decisions[key] = DocumentationConfigurationDecision(
key=key,
state=decision.state,
source=_bounded_configuration_text(decision.source),
reason=_bounded_configuration_text(decision.reason),
)
return decisions
def _bounded_configuration_text(value: object | None) -> str | None:
if value is None:
return None
clean = str(value).strip()
return clean[:500] if clean else None
def _documentation_configuration_payload(
decision: DocumentationConfigurationDecision,
) -> dict[str, str | None]:
return {
"key": decision.key,
"state": decision.state,
"source": decision.source,
"reason": decision.reason,
}
def _documentation_visibility(
topic: DocumentationTopic,
installed: set[str],
registry: PlatformRegistry,
principal: ApiPrincipal,
*,
configuration: Mapping[str, DocumentationConfigurationDecision] | None = None,
) -> tuple[bool, str, dict[str, list[str]]]:
if not topic.conditions:
return True, "documented", {"modules": [], "capabilities": [], "scopes": []}
return True, "documented", {
"modules": [],
"capabilities": [],
"scopes": [],
"configuration": [],
}
reasons: list[str] = []
blockers = {"modules": [], "capabilities": [], "scopes": []}
blockers = {
"modules": [],
"capabilities": [],
"scopes": [],
"configuration": [],
}
for condition in topic.conditions:
active, reason, condition_blockers = _condition_visibility(condition, installed, registry, principal)
active, reason, condition_blockers = _condition_visibility(
condition,
installed,
registry,
principal,
configuration=configuration,
)
if active:
return True, reason, {"modules": [], "capabilities": [], "scopes": []}
return True, reason, {
"modules": [],
"capabilities": [],
"scopes": [],
"configuration": [],
}
reasons.append(reason)
for key, values in condition_blockers.items():
blockers[key].extend(value for value in values if value not in blockers[key])
@@ -520,6 +878,8 @@ def _condition_visibility(
installed: set[str],
registry: PlatformRegistry,
principal: ApiPrincipal,
*,
configuration: Mapping[str, DocumentationConfigurationDecision] | None = None,
) -> tuple[bool, str, dict[str, list[str]]]:
missing_required_modules = _missing_required_modules(condition, installed)
unsatisfied_any_modules = _unsatisfied_any_modules(condition, installed)
@@ -527,6 +887,10 @@ def _condition_visibility(
missing_capabilities = _missing_required_capabilities(condition, registry)
missing_scopes = _missing_required_scopes(condition, principal)
unsatisfied_any_scopes = _unsatisfied_any_scopes(condition, principal)
unavailable_configuration = _unavailable_configuration_keys(
condition,
configuration or {},
)
blockers = _condition_blockers(
missing_required_modules=missing_required_modules,
unsatisfied_any_modules=unsatisfied_any_modules,
@@ -534,6 +898,7 @@ def _condition_visibility(
missing_capabilities=missing_capabilities,
missing_scopes=missing_scopes,
unsatisfied_any_scopes=unsatisfied_any_scopes,
unavailable_configuration=unavailable_configuration,
)
if _condition_has_no_blockers(blockers):
@@ -546,6 +911,7 @@ def _condition_visibility(
missing_capabilities=missing_capabilities,
missing_scopes=missing_scopes,
unsatisfied_any_scopes=unsatisfied_any_scopes,
unavailable_configuration=unavailable_configuration,
)
return False, reason, blockers
@@ -582,6 +948,20 @@ def _unsatisfied_any_scopes(condition: DocumentationCondition, principal: ApiPri
return list(condition.any_scopes)
def _unavailable_configuration_keys(
condition: DocumentationCondition,
configuration: Mapping[str, DocumentationConfigurationDecision],
) -> list[str]:
return [
key
for key in condition.configuration_keys
if configuration.get(
key,
DocumentationConfigurationDecision(key=key, state="unavailable"),
).state not in _CONFIGURATION_ACTIVE_STATES
]
def _condition_blockers(
*,
missing_required_modules: list[str],
@@ -590,14 +970,21 @@ def _condition_blockers(
missing_capabilities: list[str],
missing_scopes: list[str],
unsatisfied_any_scopes: list[str],
unavailable_configuration: list[str],
) -> dict[str, list[str]]:
blockers = {"modules": [], "capabilities": [], "scopes": []}
blockers = {
"modules": [],
"capabilities": [],
"scopes": [],
"configuration": [],
}
_extend_unique(blockers["modules"], missing_required_modules)
_extend_unique(blockers["modules"], unsatisfied_any_modules)
_extend_unique(blockers["modules"], conflicting_modules)
_extend_unique(blockers["capabilities"], missing_capabilities)
_extend_unique(blockers["scopes"], missing_scopes)
_extend_unique(blockers["scopes"], unsatisfied_any_scopes)
_extend_unique(blockers["configuration"], unavailable_configuration)
return blockers
@@ -606,7 +993,7 @@ def _extend_unique(target: list[str], values: list[str]) -> None:
def _condition_has_no_blockers(blockers: dict[str, list[str]]) -> bool:
return not blockers["modules"] and not blockers["capabilities"] and not blockers["scopes"]
return not any(blockers.values())
def _condition_blocker_reason(
@@ -617,6 +1004,7 @@ def _condition_blocker_reason(
missing_capabilities: list[str],
missing_scopes: list[str],
unsatisfied_any_scopes: list[str],
unavailable_configuration: list[str],
) -> str:
parts: list[str] = []
if missing_required_modules:
@@ -631,16 +1019,32 @@ def _condition_blocker_reason(
parts.append("missing scopes: " + ", ".join(missing_scopes))
if unsatisfied_any_scopes:
parts.append("requires one scope from: " + ", ".join(unsatisfied_any_scopes))
if unavailable_configuration:
parts.append(
"configuration is disabled or unavailable: "
+ ", ".join(unavailable_configuration)
)
return "; ".join(parts)
def _documentation_target_layer(topic: DocumentationTopic, active: bool, blockers: dict[str, list[str]]) -> str:
def _documentation_target_layer(
topic: DocumentationTopic,
active: bool,
blockers: dict[str, list[str]],
*,
configuration: Mapping[str, DocumentationConfigurationDecision] | None = None,
) -> str:
if topic.layer == "always":
return "always"
if active:
return topic.layer
if blockers["modules"] or blockers["capabilities"]:
return "evidence"
if any(
decision.state == "unavailable"
for decision in (configuration or {}).values()
):
return "evidence"
return "available"
@@ -655,6 +1059,7 @@ def _documentation_topic_payload(
locale: str,
documentation_type: DocumentationType,
visible_runtime_paths: frozenset[str],
configuration: Mapping[str, DocumentationConfigurationDecision],
) -> dict[str, Any]:
module_id = topic.source_module_id or source_module_id
translation_locale, translation = _translation_for_locale(topic, locale)
@@ -676,6 +1081,9 @@ def _documentation_topic_payload(
"modules": sorted(dict.fromkeys(blockers.get("modules", ()))),
"capabilities": sorted(dict.fromkeys(blockers.get("capabilities", ()))),
"scopes": sorted(dict.fromkeys(blockers.get("scopes", ()))),
"configuration": sorted(
dict.fromkeys(blockers.get("configuration", ()))
),
},
"audience": list(topic.audience),
"order": topic.order,
@@ -687,6 +1095,10 @@ def _documentation_topic_payload(
"related_modules": list(topic.related_modules),
"unlocks": list(topic.unlocks),
"configuration_keys": sorted({*topic.configuration_keys, *(key for condition in topic.conditions for key in condition.configuration_keys)}),
"configuration_states": [
_documentation_configuration_payload(configuration[key])
for key in sorted(configuration)
],
"metadata": dict(topic.metadata),
}
if documentation_type == "admin":
@@ -704,7 +1116,12 @@ def _documentation_topic_payload(
"documentation_types": ["user"],
"active": True,
"reason": "documented",
"blockers": {"modules": [], "capabilities": [], "scopes": []},
"blockers": {
"modules": [],
"capabilities": [],
"scopes": [],
"configuration": [],
},
"audience": [],
"order": payload["order"],
"i18n_key": "",
@@ -719,6 +1136,7 @@ def _documentation_topic_payload(
"related_modules": [],
"unlocks": list(topic.unlocks),
"configuration_keys": [],
"configuration_states": [],
"metadata": _user_topic_metadata(kind, topic.metadata),
}
@@ -909,18 +1327,3 @@ def _documentation_link_payload(link: DocumentationLink) -> dict[str, str]:
"href": link.href,
"kind": link.kind,
}
def _evidence_sources() -> list[dict[str, str]]:
return [
{
"label": "Documentation layer concept",
"source": "govoplan-docs/docs/DOCUMENTATION_LAYER_CONCEPT.md",
"layer": "evidence",
},
{
"label": "Master roadmap",
"source": "govoplan-core/docs/GOVOPLAN_MASTER_ROADMAP.md",
"layer": "evidence",
},
]