Class WorkflowRegistry

src/specify_cli/workflows/catalog.py:58–121

Manages the registry of installed workflows. Tracks installed workflows and their metadata in ``.specify/workflows/workflow-registry.json``.

class WorkflowRegistry:
    """Manages the registry of installed workflows.

    Tracks installed workflows and their metadata in
    ``.specify/workflows/workflow-registry.json``.
    """

    REGISTRY_FILE = "workflow-registry.json"
    SCHEMA_VERSION = "1.0"

    def __init__(self, project_root: Path) -> None:
        self.project_root = project_root
        self.workflows_dir = project_root / ".specify" / "workflows"
        self.registry_path = self.workflows_dir / self.REGISTRY_FILE
        self.data = self._load()

    def _load(self) -> dict[str, Any]:
        """Load registry from disk or create default."""
        if self.registry_path.exists():
            try:
                with open(self.registry_path, encoding="utf-8") as f:
                    return json.load(f)
            except (json.JSONDecodeError, ValueError):
                # Corrupted registry file — reset to default
                return {"schema_version": self.SCHEMA_VERSION, "workflows": {}}
        return {"schema_version": self.SCHEMA_VERSION, "workflows": {}}

    def save(self) -> None:
        """Persist registry to disk."""
        self.workflows_dir.mkdir(parents=True, exist_ok=True)
        with open(self.registry_path, "w", encoding="utf-8") as f:
            json.dump(self.data, f, indent=2)

    def add(self, workflow_id: str, metadata: dict[str, Any]) -> None:
        """Add or update an installed workflow entry."""
        from datetime import datetime, timezone

        existing = self.data["workflows"].get(workflow_id, {})
        metadata["installed_at"] = existing.get(
            "installed_at", datetime.now(timezone.utc).isoformat()
        )
        metadata["updated_at"] = datetime.now(timezone.utc).isoformat()
        self.data["workflows"][workflow_id] = metadata
        self.save()

    def remove(self, workflow_id: str) -> bool:
        """Remove an installed workflow entry. Returns True if found."""
        if workflow_id in self.data["workflows"]:
            del self.data["workflows"][workflow_id]
            self.save()
            return True
        return False

    def get(self, workflow_id: str) -> dict[str, Any] | None:
        """Get metadata for an installed workflow."""
        return self.data["workflows"].get(workflow_id)

    def list(self) -> dict[str, dict[str, Any]]: