feat: add web UI project registry and onboarding (#427)
Load projects from versioned webui/data/projects.registry.json with profile
mappings, workflow/schema paths, and read-only onboarding checklist UI.
Seeds Gitea-Tools; exposes /projects, /projects/{id}, and /api/projects.
Closes #427
This commit is contained in:
@@ -0,0 +1,196 @@
|
||||
"""Load and validate the web UI project registry (#427)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_FORBIDDEN_EXACT_KEYS = frozenset({
|
||||
"token",
|
||||
"password",
|
||||
"secret",
|
||||
"credential",
|
||||
"auth",
|
||||
"api_key",
|
||||
"api-key",
|
||||
})
|
||||
_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
|
||||
_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
|
||||
|
||||
|
||||
def _is_forbidden_key(key: str) -> bool:
|
||||
lowered = key.lower()
|
||||
if lowered in _FORBIDDEN_EXACT_KEYS:
|
||||
return True
|
||||
return (
|
||||
lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
|
||||
or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
|
||||
)
|
||||
|
||||
_REQUIRED_PROJECT_FIELDS = (
|
||||
"id",
|
||||
"repo_name",
|
||||
"gitea_owner",
|
||||
"remote_host",
|
||||
"default_branch",
|
||||
"local_checkout_path",
|
||||
"profiles",
|
||||
"workflow_paths",
|
||||
)
|
||||
|
||||
_REQUIRED_PROFILE_ROLES = ("author", "reviewer", "reconciler")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OnboardingStep:
|
||||
id: str
|
||||
title: str
|
||||
description: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProjectRecord:
|
||||
id: str
|
||||
repo_name: str
|
||||
gitea_owner: str
|
||||
remote_host: str
|
||||
default_branch: str
|
||||
local_checkout_path: str
|
||||
profiles: dict[str, str]
|
||||
workflow_paths: dict[str, str]
|
||||
schema_paths: dict[str, str]
|
||||
onboarding_checklist: tuple[OnboardingStep, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProjectRegistry:
|
||||
version: int
|
||||
projects: tuple[ProjectRecord, ...]
|
||||
source_path: Path
|
||||
|
||||
|
||||
def default_registry_path() -> Path:
|
||||
override = os.environ.get("WEBUI_PROJECT_REGISTRY", "").strip()
|
||||
if override:
|
||||
return Path(override).expanduser().resolve()
|
||||
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
|
||||
|
||||
|
||||
def _reject_credential_keys(obj: Any, *, path: str = "") -> None:
|
||||
if isinstance(obj, dict):
|
||||
for key, value in obj.items():
|
||||
key_path = f"{path}.{key}" if path else key
|
||||
if _is_forbidden_key(key):
|
||||
raise ValueError(f"registry must not store credentials ({key_path})")
|
||||
_reject_credential_keys(value, path=key_path)
|
||||
elif isinstance(obj, list):
|
||||
for index, item in enumerate(obj):
|
||||
_reject_credential_keys(item, path=f"{path}[{index}]")
|
||||
|
||||
|
||||
def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
|
||||
if not raw:
|
||||
return ()
|
||||
steps: list[OnboardingStep] = []
|
||||
for item in raw:
|
||||
steps.append(
|
||||
OnboardingStep(
|
||||
id=str(item["id"]),
|
||||
title=str(item["title"]),
|
||||
description=str(item["description"]),
|
||||
)
|
||||
)
|
||||
return tuple(steps)
|
||||
|
||||
|
||||
def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
|
||||
missing = [field for field in _REQUIRED_PROJECT_FIELDS if field not in raw]
|
||||
if missing:
|
||||
raise ValueError(f"project missing required fields: {', '.join(missing)}")
|
||||
|
||||
profiles = raw["profiles"]
|
||||
if not isinstance(profiles, dict):
|
||||
raise ValueError("profiles must be an object")
|
||||
for role in _REQUIRED_PROFILE_ROLES:
|
||||
if role not in profiles or not profiles[role]:
|
||||
raise ValueError(f"profiles.{role} is required")
|
||||
|
||||
workflow_paths = raw["workflow_paths"]
|
||||
if not isinstance(workflow_paths, dict) or not workflow_paths:
|
||||
raise ValueError("workflow_paths must be a non-empty object")
|
||||
|
||||
schema_paths = raw.get("schema_paths") or {}
|
||||
if not isinstance(schema_paths, dict):
|
||||
raise ValueError("schema_paths must be an object when present")
|
||||
|
||||
return ProjectRecord(
|
||||
id=str(raw["id"]),
|
||||
repo_name=str(raw["repo_name"]),
|
||||
gitea_owner=str(raw["gitea_owner"]),
|
||||
remote_host=str(raw["remote_host"]),
|
||||
default_branch=str(raw["default_branch"]),
|
||||
local_checkout_path=str(raw["local_checkout_path"]),
|
||||
profiles={role: str(profiles[role]) for role in _REQUIRED_PROFILE_ROLES},
|
||||
workflow_paths={key: str(value) for key, value in workflow_paths.items()},
|
||||
schema_paths={key: str(value) for key, value in schema_paths.items()},
|
||||
onboarding_checklist=_parse_onboarding(raw.get("onboarding_checklist")),
|
||||
)
|
||||
|
||||
|
||||
def load_registry(path: Path | None = None) -> ProjectRegistry:
|
||||
"""Load the versioned project registry from disk."""
|
||||
source = (path or default_registry_path()).resolve()
|
||||
raw_text = source.read_text(encoding="utf-8")
|
||||
payload = json.loads(raw_text)
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("registry root must be an object")
|
||||
|
||||
version = payload.get("version")
|
||||
if version != 1:
|
||||
raise ValueError(f"unsupported registry version: {version!r}")
|
||||
|
||||
_reject_credential_keys(payload)
|
||||
|
||||
projects_raw = payload.get("projects")
|
||||
if not isinstance(projects_raw, list) or not projects_raw:
|
||||
raise ValueError("projects must be a non-empty array")
|
||||
|
||||
projects = tuple(_parse_project(item) for item in projects_raw)
|
||||
return ProjectRegistry(version=version, projects=projects, source_path=source)
|
||||
|
||||
|
||||
def project_to_dict(project: ProjectRecord) -> dict[str, Any]:
|
||||
"""Serialize a project for JSON API responses."""
|
||||
return {
|
||||
"id": project.id,
|
||||
"repo_name": project.repo_name,
|
||||
"gitea_owner": project.gitea_owner,
|
||||
"remote_host": project.remote_host,
|
||||
"default_branch": project.default_branch,
|
||||
"local_checkout_path": project.local_checkout_path,
|
||||
"profiles": dict(project.profiles),
|
||||
"workflow_paths": dict(project.workflow_paths),
|
||||
"schema_paths": dict(project.schema_paths),
|
||||
"onboarding_checklist": [
|
||||
{"id": step.id, "title": step.title, "description": step.description}
|
||||
for step in project.onboarding_checklist
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def registry_to_dict(registry: ProjectRegistry) -> dict[str, Any]:
|
||||
return {
|
||||
"version": registry.version,
|
||||
"source_path": str(registry.source_path),
|
||||
"projects": [project_to_dict(project) for project in registry.projects],
|
||||
}
|
||||
|
||||
|
||||
def find_project(registry: ProjectRegistry, project_id: str) -> ProjectRecord | None:
|
||||
for project in registry.projects:
|
||||
if project.id == project_id:
|
||||
return project
|
||||
return None
|
||||
Reference in New Issue
Block a user