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

Expose searchable module catalog compatibility

Browse Source
This commit is contained in:
zemion
2026-08-06 22:42:13 +02:00
parent 1e55a80d3c
commit 9d522ef6e0
8 changed files with 501 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_admin/backend/api/v1/routes.py
+169 -7
View File
@@ -84,6 +84,7 @@ from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, Maintenance
from govoplan_core.core.lifecycle import ModuleLifecycleManager
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.core.runtime import get_registry
from govoplan_core.core.versioning import compare_versions, format_version_range, version_satisfies_range
from govoplan_core.db.session import get_session
from govoplan_core.i18n import (
REFERENCE_LANGUAGE_CODE,
@@ -635,6 +636,12 @@ def _catalog_install_or_update_item(result: dict[str, object], module_id: str) -
continue
if raw_item.get("module_id") != module_id or raw_item.get("action") not in {"install", "update"}:
continue
if raw_item.get("availability") == "withdrawn":
reason = str(raw_item.get("availability_reason") or "Catalog release has been withdrawn.")
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Catalog entry {module_id} cannot be installed: {reason}",
)
return raw_item
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Catalog install/update entry not found: {module_id}")
@@ -709,18 +716,21 @@ def _catalog_license_features(raw_item: dict[str, object]) -> list[str]:
def _module_package_catalog_item(
raw_item: dict[str, object],
available: Mapping[str, Any],
catalog_items: list[dict[str, object]] | None = None,
) -> ModulePackageCatalogItem:
payload = dict(raw_item)
manifest = available.get(str(raw_item.get("module_id") or ""))
module_id = str(raw_item.get("module_id") or "")
manifest = available.get(module_id)
installed_version = str(getattr(manifest, "version", "") or "") or None
target_version = raw_item.get("version")
version_comparison = (
compare_versions(target_version, installed_version)
if installed_version and isinstance(target_version, str)
else None
)
payload["installed"] = manifest is not None
payload["installed_version"] = installed_version
payload["update_available"] = bool(
installed_version
and isinstance(target_version, str)
and target_version != installed_version
)
payload["update_available"] = version_comparison is not None and version_comparison > 0
decision = module_license_decision(_catalog_license_features(raw_item))
payload["license_allowed"] = bool(decision.get("allowed"))
payload["license_enforced"] = bool(decision.get("enforced"))
@@ -728,9 +738,157 @@ def _module_package_catalog_item(
payload["license_missing_features"] = [str(item) for item in missing] if isinstance(missing, list) else []
reason = decision.get("reason")
payload["license_reason"] = reason if isinstance(reason, str) else None
compatibility_reasons = _catalog_compatibility_reasons(
raw_item,
available=available,
catalog_items=catalog_items or [raw_item],
)
if not bool(decision.get("allowed")):
compatibility_reasons.append(
str(reason or f"The active license policy does not allow {module_id}.")
)
if raw_item.get("action") not in {"install", "update"}:
compatibility_reasons.append("This catalog action cannot be added to an install plan.")
compatibility_reasons = list(dict.fromkeys(compatibility_reasons))
payload["compatible"] = not compatibility_reasons
payload["plan_allowed"] = bool(
not compatibility_reasons
and raw_item.get("action") in {"install", "update"}
and (
manifest is None
or (version_comparison is not None and version_comparison > 0)
or (version_comparison is not None and version_comparison < 0 and raw_item.get("allow_downgrade") is True)
or (version_comparison == 0 and raw_item.get("allow_same_version") is True)
)
)
payload["compatibility_reasons"] = compatibility_reasons
if raw_item.get("availability") == "withdrawn":
payload["catalog_state"] = "withdrawn"
elif compatibility_reasons:
payload["catalog_state"] = "blocked"
elif payload["update_available"]:
payload["catalog_state"] = "update_available"
elif payload["installed"]:
payload["catalog_state"] = "installed"
else:
payload["catalog_state"] = "available"
return ModulePackageCatalogItem.model_validate(payload)
def _catalog_compatibility_reasons(
raw_item: dict[str, object],
*,
available: Mapping[str, Any],
catalog_items: list[dict[str, object]],
) -> list[str]:
module_id = str(raw_item.get("module_id") or "")
reasons: list[str] = []
if raw_item.get("availability") == "withdrawn":
reasons.append(
str(raw_item.get("availability_reason") or "This catalog release has been withdrawn.")
)
manifest = available.get(module_id)
installed_version = str(getattr(manifest, "version", "") or "") or None
version_min = _catalog_string(raw_item, "current_version_min")
version_max = _catalog_string(raw_item, "current_version_max_exclusive")
if installed_version and not version_satisfies_range(
installed_version,
version_min=version_min,
version_max_exclusive=version_max,
):
reasons.append(
f"Installed version {installed_version} is outside the supported update window "
f"({format_version_range(version_min=version_min, version_max_exclusive=version_max)})."
)
target_version = _catalog_string(raw_item, "version")
if (
installed_version
and target_version
and compare_versions(target_version, installed_version) < 0
and raw_item.get("allow_downgrade") is not True
):
reasons.append(
f"Catalog version {target_version} is older than installed version {installed_version}; "
"the release does not allow a downgrade."
)
catalog_module_ids = {
str(item.get("module_id") or "")
for item in catalog_items
if item.get("availability") != "withdrawn"
}
for dependency in _catalog_string_values(raw_item.get("dependencies")):
if dependency not in available and dependency not in catalog_module_ids:
reasons.append(f"Required module {dependency} is neither installed nor available in this catalog.")
providers = _catalog_interface_providers(catalog_items, available=available)
requirements = raw_item.get("requires_interfaces")
if isinstance(requirements, list):
for requirement in requirements:
if not isinstance(requirement, Mapping) or requirement.get("optional") is True:
continue
interface_name = str(requirement.get("name") or "").strip()
if not interface_name:
continue
required_min = _catalog_mapping_string(requirement, "version_min")
required_max = _catalog_mapping_string(requirement, "version_max_exclusive")
if any(
version_satisfies_range(
provider_version,
version_min=required_min,
version_max_exclusive=required_max,
)
for provider_version in providers.get(interface_name, ())
):
continue
reasons.append(
f"Required interface {interface_name} "
f"({format_version_range(version_min=required_min, version_max_exclusive=required_max)}) "
"has no compatible installed or catalog provider."
)
return reasons
def _catalog_interface_providers(
catalog_items: list[dict[str, object]],
*,
available: Mapping[str, Any],
) -> dict[str, set[str]]:
providers: dict[str, set[str]] = {}
for item in catalog_items:
if item.get("availability") == "withdrawn":
continue
declared = item.get("provides_interfaces")
if not isinstance(declared, list):
continue
for provider in declared:
if not isinstance(provider, Mapping):
continue
name = str(provider.get("name") or "").strip()
version = str(provider.get("version") or "").strip()
if name and version:
providers.setdefault(name, set()).add(version)
for manifest in available.values():
for provider in getattr(manifest, "provides_interfaces", ()):
name = str(getattr(provider, "name", "") or "").strip()
version = str(getattr(provider, "version", "") or "").strip()
if name and version:
providers.setdefault(name, set()).add(version)
return providers
def _catalog_string_values(value: object) -> list[str]:
if not isinstance(value, list):
return []
return [str(item).strip() for item in value if str(item).strip()]
def _catalog_mapping_string(value: Mapping[str, object], field: str) -> str | None:
raw = value.get(field)
return str(raw).strip() if isinstance(raw, str) and raw.strip() else None
def _catalog_required_license_features(result: dict[str, object]) -> list[str]:
required: list[str] = []
for raw_item in result.get("modules", []):
@@ -1260,9 +1418,13 @@ def read_module_package_catalog(
if isinstance(lifecycle, ModuleLifecycleManager)
else available_module_manifests(ignore_load_errors=True)
)
catalog_items = [item for item in result["modules"] if isinstance(item, dict)]
license_diagnostics = module_license_diagnostics(required_features=_catalog_required_license_features(result))
return ModulePackageCatalogResponse(
modules=[_module_package_catalog_item(item, available) for item in result["modules"] if isinstance(item, dict)],
modules=[
_module_package_catalog_item(item, available, catalog_items)
for item in catalog_items
],
configured=bool(result["configured"]),
valid=bool(result["valid"]),
path=result["path"] if isinstance(result["path"], str) else None,
src/govoplan_admin/backend/api/v1/schemas.py
+34
View File
@@ -387,6 +387,30 @@ class ModuleInterfaceRequirementItem(BaseModel):
optional: bool = False
class ModulePackageCatalogSource(BaseModel):
repository: str
tag: str
commit: str
repository_url: str | None = None
revision_url: str | None = None
class ModulePackageArtifactIdentity(BaseModel):
ref: str | None = None
path: str | None = None
artifact_path: str | None = None
url: str | None = None
filename: str | None = None
sha256: str | None = None
size: int | None = None
integrity: str | None = None
sbom_url: str | None = None
provenance_url: str | None = None
registry_identity: str | None = None
git_ref: str | None = None
source_commit: str | None = None
class ModulePackageCatalogItem(BaseModel):
module_id: str
name: str
@@ -396,6 +420,16 @@ class ModulePackageCatalogItem(BaseModel):
installed: bool = False
installed_version: str | None = None
update_available: bool = False
availability: Literal["available", "withdrawn"] = "available"
availability_reason: str | None = None
configuration_requirements: list[str] = Field(default_factory=list)
release_notes_url: str | None = None
source: ModulePackageCatalogSource | None = None
artifact_integrity: dict[str, ModulePackageArtifactIdentity] = Field(default_factory=dict)
compatible: bool = True
plan_allowed: bool = True
compatibility_reasons: list[str] = Field(default_factory=list)
catalog_state: Literal["available", "installed", "update_available", "blocked", "withdrawn"] = "available"
dependencies: list[str] = Field(default_factory=list)
optional_dependencies: list[str] = Field(default_factory=list)
migration_safety: Literal["automatic", "requires_review", "forward_only", "destructive"] = "automatic"