fix(cli): add Invariants and Validations panels to project show rich output
CI / lint (pull_request) Failing after 1m0s
CI / quality (pull_request) Successful in 48s
CI / typecheck (pull_request) Successful in 1m2s
CI / build (pull_request) Successful in 40s
CI / helm (pull_request) Successful in 28s
CI / push-validation (pull_request) Successful in 25s
CI / security (pull_request) Successful in 1m37s
CI / coverage (pull_request) Has been skipped
CI / e2e_tests (pull_request) Successful in 3m35s
CI / integration_tests (pull_request) Successful in 4m55s
CI / unit_tests (pull_request) Failing after 5m35s
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 3s

This commit is contained in:
2026-05-05 19:05:32 +00:00
parent 2f2a1e6dc8
commit 0bd9e61ba4
4 changed files with 185 additions and 20 deletions
@@ -0,0 +1,141 @@
"""Application service for namespaced project management.
Wraps ``NamespacedProjectRepository`` with higher-level operations used by
the CLI layer.
"""
from __future__ import annotations
import re
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from cleveragents.infrastructure.database.repositories import (
NamespacedProjectRepository,
)
# Regex for a valid bare name segment (letter, then alphanumeric/hyphen/underscore)
_BARE_NAME_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*$")
# Default namespace when none is specified
_DEFAULT_NAMESPACE = "local"
class NamespacedProjectService:
"""Coordinate namespaced project lifecycle operations.
Delegates all persistence to the repository layer. Designed to be
injected with a :class:`NamespacedProjectRepository`.
"""
def __init__(self, project_repo: NamespacedProjectRepository) -> None:
self._repo = project_repo
# -- Validation helpers --------------------------------------------------
def validate_project_name(self, name: str) -> None:
"""Validate a project name (bare or namespace/name).
Args:
name: The project name to validate.
Raises:
ValueError: If the name is invalid.
"""
if "/" in name:
parts = name.split("/", 1)
namespace, bare_name = parts[0], parts[1]
if not _BARE_NAME_RE.match(namespace):
raise ValueError(
f"Namespace '{namespace}' is invalid. Must start with a letter "
f"and contain only letters, digits, hyphens, and underscores."
)
if not _BARE_NAME_RE.match(bare_name):
raise ValueError(
f"Project name '{bare_name}' is invalid. Must start with a letter "
f"and contain only letters, digits, hyphens, and underscores."
)
else:
if not _BARE_NAME_RE.match(name):
raise ValueError(
f"Project name '{name}' is invalid. Must start with a letter "
f"and contain only letters, digits, hyphens, and underscores."
)
# -- CRUD operations -----------------------------------------------------
def create_project(
self,
name: str,
description: str | None = None,
) -> Any:
"""Create a new namespaced project.
Args:
name: The project name (bare or namespace/name).
description: Optional project description.
Returns:
The created ``NamespacedProject`` domain object.
Raises:
DatabaseError: If the project already exists or a DB error occurs.
"""
from cleveragents.domain.models.core.project import NamespacedProject
if "/" in name:
namespace, bare_name = name.split("/", 1)
else:
namespace = _DEFAULT_NAMESPACE
bare_name = name
project = NamespacedProject(
name=bare_name,
namespace=namespace,
description=description,
)
return self._repo.create(project)
def get_project(self, namespaced_name: str) -> Any:
"""Retrieve a project by its namespaced name.
Args:
namespaced_name: The ``namespace/name`` string.
Returns:
The ``NamespacedProject`` domain object.
Raises:
ProjectNotFoundError: If the project is not found.
"""
return self._repo.get(namespaced_name)
def list_projects(
self,
namespace: str | None = None,
limit: int = 100,
) -> list[Any]:
"""List projects, optionally filtered by namespace.
Args:
namespace: Optional namespace filter.
limit: Maximum number of results (default 100).
Returns:
List of ``NamespacedProject`` domain objects.
"""
return self._repo.list_projects(namespace=namespace, limit=limit)
def delete_project(self, namespaced_name: str) -> bool:
"""Delete a project by namespaced name.
Args:
namespaced_name: The ``namespace/name`` string.
Returns:
``True`` if the project was deleted, ``False`` if not found.
Raises:
DatabaseError: On transient or unexpected DB errors.
"""
return self._repo.delete(namespaced_name)
@@ -7,7 +7,10 @@ These commands are preserved for backward compatibility.
from __future__ import annotations
from pathlib import Path
from typing import Annotated
from typing import TYPE_CHECKING, Annotated, Any
if TYPE_CHECKING:
from cleveragents.application.services.project_service import ProjectService
import typer
from rich.panel import Panel
@@ -43,7 +46,6 @@ def init_command(
) -> None:
"""Programmatic interface for initializing a new CleverAgents project."""
from cleveragents.application.container import get_container
from cleveragents.application.services.project_service import ProjectService
if path is None:
path = Path.cwd()
@@ -64,7 +66,7 @@ def init_command(
prompt_for_migration=lambda _: True,
)
project_service: ProjectService = container.project_service()
project_service = container.project_service()
project = project_service.initialize_project(
name=project_name,
@@ -121,12 +123,16 @@ def init_command(
# ---------------------------------------------------------------------------
def _get_project_service_and_current_project_legacy() -> tuple[object, object]:
def _get_project_service_and_current_project_legacy() -> (
tuple[ProjectService, Any]
):
from cleveragents.application.container import get_container
from cleveragents.application.services.project_service import ProjectService
from cleveragents.application.services.project_service import (
ProjectService as _ProjectService,
)
container = get_container()
project_service: ProjectService = container.project_service()
project_service: _ProjectService = container.project_service()
project = project_service.get_current_project()
if not project:
err_console.print("[red]Error: No project found in current directory.[/red]")
@@ -349,11 +355,10 @@ def register_legacy_commands(app: typer.Typer) -> None:
def status() -> None:
"""Show current project status and information."""
from cleveragents.application.container import get_container
from cleveragents.application.services.project_service import ProjectService
try:
container = get_container()
project_service: ProjectService = container.project_service()
project_service = container.project_service()
# Get current project
project = project_service.get_current_project()
+28 -11
View File
@@ -46,22 +46,39 @@ def _get_project_invariants(project: Any) -> list[dict[str, Any]]:
def _get_project_validations(project: Any) -> list[dict[str, Any]]:
"""Fetch validation attachments for a project."""
"""Fetch validation attachments for a project.
Collects all validation attachments scoped to this project by querying
each linked resource's validation attachments and filtering by project name.
"""
from cleveragents.application.container import get_container
from cleveragents.application.services.tool_registry_service import (
ToolRegistryService,
)
container = get_container()
try:
svc_obj = container.tool_registry_service()
svc: ToolRegistryService = container.tool_registry_service()
attachments: list[dict[str, Any]] = []
if hasattr(svc_obj, "list_attachments_for_project"):
attachments = svc_obj.list_attachments_for_project(project.namespaced_name)
elif hasattr(svc_obj, "list_attachments"):
all_attachments = svc_obj.list_attachments()
attachments = [
a
for a in all_attachments
if getattr(a, "project_name", None) == project.namespaced_name
]
for lr in getattr(project, "linked_resources", []):
resource_id = getattr(lr, "resource_id", None)
if not resource_id:
continue
resource_attachments = svc.list_validations_for_resource(
resource_id=resource_id,
project_name=project.namespaced_name,
)
for att in resource_attachments:
if isinstance(att, dict):
attachments.append(att)
else:
attachments.append(
{
"validation_name": getattr(att, "validation_name", ""),
"resource_id": getattr(att, "resource_id", ""),
"mode": getattr(att, "mode", "required"),
}
)
return attachments
except Exception:
return []
+3 -1
View File
@@ -455,7 +455,9 @@ def init(
] = False,
) -> None:
"""Initialize a new CleverAgents project in the current directory."""
from cleveragents.cli.commands.project import init_command as project_init_command
from cleveragents.cli.commands.project_legacy import (
init_command as project_init_command,
)
try:
project_init_command(