Add project registration policy and template

This commit is contained in:
2026-04-06 08:46:37 -04:00
parent d8028f406e
commit 827dcf2cd1
8 changed files with 238 additions and 0 deletions

View File

@@ -34,6 +34,7 @@ from atocore.memory.service import (
)
from atocore.observability.logger import get_logger
from atocore.projects.registry import (
get_project_registry_template,
list_registered_projects,
refresh_registered_project,
)
@@ -169,6 +170,16 @@ def api_projects() -> dict:
}
@router.get("/projects/template")
def api_projects_template() -> dict:
"""Return a starter template for project registry entries."""
return {
"template": get_project_registry_template(),
"registry_path": str(_config.settings.resolved_project_registry_path),
"allowed_sources": ["vault", "drive"],
}
@router.post("/projects/{project_name}/refresh", response_model=ProjectRefreshResponse)
def api_refresh_project(project_name: str, purge_deleted: bool = False) -> ProjectRefreshResponse:
"""Refresh one registered project from its configured ingest roots."""

View File

@@ -25,6 +25,26 @@ class RegisteredProject:
ingest_roots: tuple[ProjectSourceRef, ...]
def get_project_registry_template() -> dict:
"""Return a minimal template for registering a new project."""
return {
"projects": [
{
"id": "p07-example",
"aliases": ["p07", "example-project"],
"description": "Short description of the project and staged corpus.",
"ingest_roots": [
{
"source": "vault",
"subpath": "incoming/projects/p07-example",
"label": "Primary staged project docs",
}
],
}
]
}
def load_project_registry() -> list[RegisteredProject]:
"""Load project registry entries from JSON config."""
registry_path = _config.settings.resolved_project_registry_path
@@ -37,6 +57,8 @@ def load_project_registry() -> list[RegisteredProject]:
for entry in entries:
project_id = str(entry["id"]).strip()
if not project_id:
raise ValueError("Project registry entry is missing a non-empty id")
aliases = tuple(
alias.strip()
for alias in entry.get("aliases", [])
@@ -53,6 +75,8 @@ def load_project_registry() -> list[RegisteredProject]:
if str(root.get("source", "")).strip()
and str(root.get("subpath", "")).strip()
)
if not ingest_roots:
raise ValueError(f"Project registry entry '{project_id}' has no ingest_roots")
projects.append(
RegisteredProject(
project_id=project_id,
@@ -62,6 +86,7 @@ def load_project_registry() -> list[RegisteredProject]:
)
)
_validate_unique_project_names(projects)
return projects
@@ -150,3 +175,17 @@ def _resolve_ingest_root(source_ref: ProjectSourceRef) -> Path:
raise ValueError(f"Unsupported source root: {source_ref.source}") from exc
return (base_dir / source_ref.subpath).resolve(strict=False)
def _validate_unique_project_names(projects: list[RegisteredProject]) -> None:
seen: dict[str, str] = {}
for project in projects:
names = [project.project_id, *project.aliases]
for name in names:
key = name.lower()
if key in seen and seen[key] != project.project_id:
raise ValueError(
f"Project registry name collision: '{name}' is used by both "
f"'{seen[key]}' and '{project.project_id}'"
)
seen[key] = project.project_id