ros/spec-kit
1
0
Fork 0
mirror of https://github.com/github/spec-kit.git synced 2026-03-17 19:03:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(extensions): support multiple active catalogs simultaneously (#1720)

Browse Source 
* Initial plan

* feat(extensions): implement multi-catalog stack support

- Add CatalogEntry dataclass to represent catalog entries
- Add get_active_catalogs() reading SPECKIT_CATALOG_URL, project config,
  user config, or built-in default stack (org-approved + community)
- Add _load_catalog_config() to parse .specify/extension-catalogs.yml
- Add _validate_catalog_url() HTTPS validation helper
- Add _fetch_single_catalog() with per-URL caching, backward-compat for DEFAULT_CATALOG_URL
- Add _get_merged_extensions() that merges all catalogs (priority wins on conflict)
- Update search() and get_extension_info() to use merged results
  annotated with _catalog_name and _install_allowed
- Update clear_cache() to also remove per-URL hash cache files
- Add extension_catalogs CLI command to list active catalogs
- Add catalog add/remove sub-commands for .specify/extension-catalogs.yml
- Update extension_add to enforce install_allowed=false policy
- Update extension_search to show source catalog per result
- Update extension_info to show source catalog with install_allowed status
- Add 13 new tests covering catalog stack, merge conflict resolution,
  install_allowed enforcement, and catalog metadata

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>

* docs: update RFC, user guide, and API reference for multi-catalog support

- RFC: replace FUTURE FEATURE section with full implementation docs,
  add catalog stack resolution order, config file examples, merge
  conflict resolution, and install_allowed behavior
- EXTENSION-USER-GUIDE.md: add multi-catalog section with CLI examples
  for catalogs/catalog-add/catalog-remove, update catalog config docs
- EXTENSION-API-REFERENCE.md: add CatalogEntry class docs, update
  ExtensionCatalog docs with new methods and result annotations,
  add catalog CLI commands (catalogs, catalog add, catalog remove)

Also fix extension_catalogs command to correctly show "Using built-in
default catalog stack" when config file exists but has empty catalogs

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>

* Potential fix for pull request finding 'Empty except'

Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>

* fix: remove extraneous f-string prefixes (ruff F541)

Remove f-prefix from strings with no placeholders in catalog_remove
and extension_search commands.

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>

* fix: address PR review feedback for multi-catalog support

- Rename 'org-approved' catalog to 'default'
- Move 'catalogs' command to 'catalog list' for consistency
- Add 'description' field to CatalogEntry dataclass
- Add --description option to 'catalog add' CLI command
- Align install_allowed default to False in _load_catalog_config
- Add user-level config detection in catalog list footer
- Fix _load_catalog_config docstring (document ValidationError)
- Fix test isolation for test_search_by_tag, test_search_by_query,
  test_search_verified_only, test_get_extension_info
- Update version to 0.1.14 and CHANGELOG
- Update all docs (RFC, User Guide, API Reference)

* fix: wrap _load_catalog_config() calls in catalog_list with try/except

- Check SPECKIT_CATALOG_URL first (matching get_active_catalogs() resolution order)
- Wrap both _load_catalog_config() calls in try/except ValidationError so a
  malformed config file cannot crash `specify extension catalog list` after
  the active catalogs have already been printed successfully

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
This commit is contained in:
Copilot
2026-03-09 14:30:27 -05:00
committed by GitHub
parent 1df24f1953
commit ee922cbde9
7 changed files with 1287 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

232
src/specify_cli/__init__.py
View File

@@ -1772,6 +1772,13 @@ extension_app = typer.Typer(
)
app.add_typer(extension_app, name="extension")
catalog_app = typer.Typer(
name="catalog",
help="Manage extension catalogs",
add_completion=False,
)
extension_app.add_typer(catalog_app, name="catalog")
def get_speckit_version() -> str:
"""Get current spec-kit version."""
@@ -1837,6 +1844,181 @@ def extension_list(
console.print(" [cyan]specify extension add <name>[/cyan]")
@catalog_app.command("list")
def catalog_list():
"""List all active extension catalogs."""
from .extensions import ExtensionCatalog, ValidationError
project_root = Path.cwd()
specify_dir = project_root / ".specify"
if not specify_dir.exists():
console.print("[red]Error:[/red] Not a spec-kit project (no .specify/ directory)")
console.print("Run this command from a spec-kit project root")
raise typer.Exit(1)
catalog = ExtensionCatalog(project_root)
try:
active_catalogs = catalog.get_active_catalogs()
except ValidationError as e:
console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(1)
console.print("\n[bold cyan]Active Extension Catalogs:[/bold cyan]\n")
for entry in active_catalogs:
install_str = (
"[green]install allowed[/green]"
if entry.install_allowed
else "[yellow]discovery only[/yellow]"
)
console.print(f" [bold]{entry.name}[/bold] (priority {entry.priority})")
if entry.description:
console.print(f" {entry.description}")
console.print(f" URL: {entry.url}")
console.print(f" Install: {install_str}")
console.print()
config_path = project_root / ".specify" / "extension-catalogs.yml"
user_config_path = Path.home() / ".specify" / "extension-catalogs.yml"
if os.environ.get("SPECKIT_CATALOG_URL"):
console.print("[dim]Catalog configured via SPECKIT_CATALOG_URL environment variable.[/dim]")
else:
try:
proj_loaded = config_path.exists() and catalog._load_catalog_config(config_path) is not None
except ValidationError:
proj_loaded = False
if proj_loaded:
console.print(f"[dim]Config: {config_path.relative_to(project_root)}[/dim]")
else:
try:
user_loaded = user_config_path.exists() and catalog._load_catalog_config(user_config_path) is not None
except ValidationError:
user_loaded = False
if user_loaded:
console.print("[dim]Config: ~/.specify/extension-catalogs.yml[/dim]")
else:
console.print("[dim]Using built-in default catalog stack.[/dim]")
console.print(
"[dim]Add .specify/extension-catalogs.yml to customize.[/dim]"
)
@catalog_app.command("add")
def catalog_add(
url: str = typer.Argument(help="Catalog URL (must use HTTPS)"),
name: str = typer.Option(..., "--name", help="Catalog name"),
priority: int = typer.Option(10, "--priority", help="Priority (lower = higher priority)"),
install_allowed: bool = typer.Option(
False, "--install-allowed/--no-install-allowed",
help="Allow extensions from this catalog to be installed",
),
description: str = typer.Option("", "--description", help="Description of the catalog"),
):
"""Add a catalog to .specify/extension-catalogs.yml."""
from .extensions import ExtensionCatalog, ValidationError
project_root = Path.cwd()
specify_dir = project_root / ".specify"
if not specify_dir.exists():
console.print("[red]Error:[/red] Not a spec-kit project (no .specify/ directory)")
console.print("Run this command from a spec-kit project root")
raise typer.Exit(1)
# Validate URL
tmp_catalog = ExtensionCatalog(project_root)
try:
tmp_catalog._validate_catalog_url(url)
except ValidationError as e:
console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(1)
config_path = specify_dir / "extension-catalogs.yml"
# Load existing config
if config_path.exists():
try:
config = yaml.safe_load(config_path.read_text()) or {}
except Exception as e:
console.print(f"[red]Error:[/red] Failed to read {config_path}: {e}")
raise typer.Exit(1)
else:
config = {}
catalogs = config.get("catalogs", [])
if not isinstance(catalogs, list):
console.print("[red]Error:[/red] Invalid catalog config: 'catalogs' must be a list.")
raise typer.Exit(1)
# Check for duplicate name
for existing in catalogs:
if isinstance(existing, dict) and existing.get("name") == name:
console.print(f"[yellow]Warning:[/yellow] A catalog named '{name}' already exists.")
console.print("Use 'specify extension catalog remove' first, or choose a different name.")
raise typer.Exit(1)
catalogs.append({
"name": name,
"url": url,
"priority": priority,
"install_allowed": install_allowed,
"description": description,
})
config["catalogs"] = catalogs
config_path.write_text(yaml.dump(config, default_flow_style=False, sort_keys=False))
install_label = "install allowed" if install_allowed else "discovery only"
console.print(f"\n[green]✓[/green] Added catalog '[bold]{name}[/bold]' ({install_label})")
console.print(f" URL: {url}")
console.print(f" Priority: {priority}")
console.print(f"\nConfig saved to {config_path.relative_to(project_root)}")
@catalog_app.command("remove")
def catalog_remove(
name: str = typer.Argument(help="Catalog name to remove"),
):
"""Remove a catalog from .specify/extension-catalogs.yml."""
project_root = Path.cwd()
specify_dir = project_root / ".specify"
if not specify_dir.exists():
console.print("[red]Error:[/red] Not a spec-kit project (no .specify/ directory)")
console.print("Run this command from a spec-kit project root")
raise typer.Exit(1)
config_path = specify_dir / "extension-catalogs.yml"
if not config_path.exists():
console.print("[red]Error:[/red] No catalog config found. Nothing to remove.")
raise typer.Exit(1)
try:
config = yaml.safe_load(config_path.read_text()) or {}
except Exception:
console.print("[red]Error:[/red] Failed to read catalog config.")
raise typer.Exit(1)
catalogs = config.get("catalogs", [])
if not isinstance(catalogs, list):
console.print("[red]Error:[/red] Invalid catalog config: 'catalogs' must be a list.")
raise typer.Exit(1)
original_count = len(catalogs)
catalogs = [c for c in catalogs if isinstance(c, dict) and c.get("name") != name]
if len(catalogs) == original_count:
console.print(f"[red]Error:[/red] Catalog '{name}' not found.")
raise typer.Exit(1)
config["catalogs"] = catalogs
config_path.write_text(yaml.dump(config, default_flow_style=False, sort_keys=False))
console.print(f"[green]✓[/green] Removed catalog '{name}'")
if not catalogs:
console.print("\n[dim]No catalogs remain in config. Built-in defaults will be used.[/dim]")
@extension_app.command("add")
def extension_add(
extension: str = typer.Argument(help="Extension name or path"),
@@ -1925,6 +2107,19 @@ def extension_add(
console.print(" specify extension search")
raise typer.Exit(1)
# Enforce install_allowed policy
if not ext_info.get("_install_allowed", True):
catalog_name = ext_info.get("_catalog_name", "community")
console.print(
f"[red]Error:[/red] '{extension}' is available in the "
f"'{catalog_name}' catalog but installation is not allowed from that catalog."
)
console.print(
f"\nTo enable installation, add '{extension}' to an approved catalog "
f"(install_allowed: true) in .specify/extension-catalogs.yml."
)
raise typer.Exit(1)
# Download extension ZIP
console.print(f"Downloading {ext_info['name']} v{ext_info.get('version', 'unknown')}...")
zip_path = catalog.download_extension(extension)
@@ -2069,6 +2264,15 @@ def extension_search(
tags_str = ", ".join(ext['tags'])
console.print(f" [dim]Tags:[/dim] {tags_str}")
# Source catalog
catalog_name = ext.get("_catalog_name", "")
install_allowed = ext.get("_install_allowed", True)
if catalog_name:
if install_allowed:
console.print(f" [dim]Catalog:[/dim] {catalog_name}")
else:
console.print(f" [dim]Catalog:[/dim] {catalog_name} [yellow](discovery only — not installable)[/yellow]")
# Stats
stats = []
if ext.get('downloads') is not None:
@@ -2082,8 +2286,15 @@ def extension_search(
if ext.get('repository'):
console.print(f" [dim]Repository:[/dim] {ext['repository']}")
# Install command
console.print(f"\n [cyan]Install:[/cyan] specify extension add {ext['id']}")
# Install command (show warning if not installable)
if install_allowed:
console.print(f"\n [cyan]Install:[/cyan] specify extension add {ext['id']}")
else:
console.print(f"\n [yellow]⚠[/yellow] Not directly installable from '{catalog_name}'.")
console.print(
f" Add to an approved catalog with install_allowed: true, "
f"or install from a ZIP URL: specify extension add {ext['id']} --from <zip-url>"
)
console.print()
except ExtensionError as e:
@@ -2132,6 +2343,12 @@ def extension_info(
# Author and License
console.print(f"[dim]Author:[/dim] {ext_info.get('author', 'Unknown')}")
console.print(f"[dim]License:[/dim] {ext_info.get('license', 'Unknown')}")
# Source catalog
if ext_info.get("_catalog_name"):
install_allowed = ext_info.get("_install_allowed", True)
install_note = "" if install_allowed else " [yellow](discovery only)[/yellow]"
console.print(f"[dim]Source catalog:[/dim] {ext_info['_catalog_name']}{install_note}")
console.print()
# Requirements
@@ -2188,12 +2405,21 @@ def extension_info(
# Installation status and command
is_installed = manager.registry.is_installed(ext_info['id'])
install_allowed = ext_info.get("_install_allowed", True)
if is_installed:
console.print("[green]✓ Installed[/green]")
console.print(f"\nTo remove: specify extension remove {ext_info['id']}")
else:
elif install_allowed:
console.print("[yellow]Not installed[/yellow]")
console.print(f"\n[cyan]Install:[/cyan] specify extension add {ext_info['id']}")
else:
catalog_name = ext_info.get("_catalog_name", "community")
console.print("[yellow]Not installed[/yellow]")
console.print(
f"\n[yellow]⚠[/yellow] '{ext_info['id']}' is available in the '{catalog_name}' catalog "
f"but not in your approved catalog. Add it to .specify/extension-catalogs.yml "
f"with install_allowed: true to enable installation."
)
except ExtensionError as e:
console.print(f"\n[red]Error:[/red] {e}")

337
src/specify_cli/extensions.py
View File

@@ -8,9 +8,11 @@ without bloating the core framework.
import json
import hashlib
import os
import tempfile
import zipfile
import shutil
from dataclasses import dataclass
from pathlib import Path
from typing import Optional, Dict, List, Any
from datetime import datetime, timezone
@@ -36,6 +38,16 @@ class CompatibilityError(ExtensionError):
pass
@dataclass
class CatalogEntry:
"""Represents a single catalog entry in the catalog stack."""
url: str
name: str
priority: int
install_allowed: bool
description: str = ""
class ExtensionManifest:
"""Represents and validates an extension manifest (extension.yml)."""
@@ -976,6 +988,7 @@ class ExtensionCatalog:
"""Manages extension catalog fetching, caching, and searching."""
DEFAULT_CATALOG_URL = "https://raw.githubusercontent.com/github/spec-kit/main/extensions/catalog.json"
COMMUNITY_CATALOG_URL = "https://raw.githubusercontent.com/github/spec-kit/main/extensions/catalog.community.json"
CACHE_DURATION = 3600 # 1 hour in seconds
def __init__(self, project_root: Path):
@@ -990,43 +1003,109 @@ class ExtensionCatalog:
self.cache_file = self.cache_dir / "catalog.json"
self.cache_metadata_file = self.cache_dir / "catalog-metadata.json"
def get_catalog_url(self) -> str:
"""Get catalog URL from config or use default.
def _validate_catalog_url(self, url: str) -> None:
"""Validate that a catalog URL uses HTTPS (localhost HTTP allowed).
Checks in order:
1. SPECKIT_CATALOG_URL environment variable
2. Default catalog URL
Returns:
URL to fetch catalog from
Args:
url: URL to validate
Raises:
ValidationError: If custom URL is invalid (non-HTTPS)
ValidationError: If URL is invalid or uses non-HTTPS scheme
"""
import os
import sys
from urllib.parse import urlparse
# Environment variable override (useful for testing)
parsed = urlparse(url)
is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1")
if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost):
raise ValidationError(
f"Catalog URL must use HTTPS (got {parsed.scheme}://). "
"HTTP is only allowed for localhost."
)
if not parsed.netloc:
raise ValidationError("Catalog URL must be a valid URL with a host.")
def _load_catalog_config(self, config_path: Path) -> Optional[List[CatalogEntry]]:
"""Load catalog stack configuration from a YAML file.
Args:
config_path: Path to extension-catalogs.yml
Returns:
Ordered list of CatalogEntry objects, or None if file doesn't exist
or contains no valid catalog entries.
Raises:
ValidationError: If any catalog entry has an invalid URL,
the file cannot be parsed, or a priority value is invalid.
"""
if not config_path.exists():
return None
try:
data = yaml.safe_load(config_path.read_text()) or {}
except (yaml.YAMLError, OSError) as e:
raise ValidationError(
f"Failed to read catalog config {config_path}: {e}"
)
catalogs_data = data.get("catalogs", [])
if not catalogs_data:
return None
if not isinstance(catalogs_data, list):
raise ValidationError(
f"Invalid catalog config: 'catalogs' must be a list, got {type(catalogs_data).__name__}"
)
entries: List[CatalogEntry] = []
for idx, item in enumerate(catalogs_data):
if not isinstance(item, dict):
raise ValidationError(
f"Invalid catalog entry at index {idx}: expected a mapping, got {type(item).__name__}"
)
url = str(item.get("url", "")).strip()
if not url:
continue
self._validate_catalog_url(url)
try:
priority = int(item.get("priority", idx + 1))
except (TypeError, ValueError):
raise ValidationError(
f"Invalid priority for catalog '{item.get('name', idx + 1)}': "
f"expected integer, got {item.get('priority')!r}"
)
raw_install = item.get("install_allowed", False)
if isinstance(raw_install, str):
install_allowed = raw_install.strip().lower() in ("true", "yes", "1")
else:
install_allowed = bool(raw_install)
entries.append(CatalogEntry(
url=url,
name=str(item.get("name", f"catalog-{idx + 1}")),
priority=priority,
install_allowed=install_allowed,
description=str(item.get("description", "")),
))
entries.sort(key=lambda e: e.priority)
return entries if entries else None
def get_active_catalogs(self) -> List[CatalogEntry]:
"""Get the ordered list of active catalogs.
Resolution order:
1. SPECKIT_CATALOG_URL env var — single catalog replacing all defaults
2. Project-level .specify/extension-catalogs.yml
3. User-level ~/.specify/extension-catalogs.yml
4. Built-in default stack (default + community)
Returns:
List of CatalogEntry objects sorted by priority (ascending)
Raises:
ValidationError: If a catalog URL is invalid
"""
import sys
# 1. SPECKIT_CATALOG_URL env var replaces all defaults for backward compat
if env_value := os.environ.get("SPECKIT_CATALOG_URL"):
catalog_url = env_value.strip()
parsed = urlparse(catalog_url)
# Require HTTPS for security (prevent man-in-the-middle attacks)
# Allow http://localhost for local development/testing
is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1")
if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost):
raise ValidationError(
f"Invalid SPECKIT_CATALOG_URL: must use HTTPS (got {parsed.scheme}://). "
"HTTP is only allowed for localhost."
)
if not parsed.netloc:
raise ValidationError(
"Invalid SPECKIT_CATALOG_URL: must be a valid URL with a host."
)
# Warn users when using a non-default catalog (once per instance)
self._validate_catalog_url(catalog_url)
if catalog_url != self.DEFAULT_CATALOG_URL:
if not getattr(self, "_non_default_catalog_warning_shown", False):
print(
@@ -1035,11 +1114,163 @@ class ExtensionCatalog:
file=sys.stderr,
)
self._non_default_catalog_warning_shown = True
return [CatalogEntry(url=catalog_url, name="custom", priority=1, install_allowed=True, description="Custom catalog via SPECKIT_CATALOG_URL")]
return catalog_url
# 2. Project-level config overrides all defaults
project_config_path = self.project_root / ".specify" / "extension-catalogs.yml"
catalogs = self._load_catalog_config(project_config_path)
if catalogs is not None:
return catalogs
# TODO: Support custom catalogs from .specify/extension-catalogs.yml
return self.DEFAULT_CATALOG_URL
# 3. User-level config
user_config_path = Path.home() / ".specify" / "extension-catalogs.yml"
catalogs = self._load_catalog_config(user_config_path)
if catalogs is not None:
return catalogs
# 4. Built-in default stack
return [
CatalogEntry(url=self.DEFAULT_CATALOG_URL, name="default", priority=1, install_allowed=True, description="Built-in catalog of installable extensions"),
CatalogEntry(url=self.COMMUNITY_CATALOG_URL, name="community", priority=2, install_allowed=False, description="Community-contributed extensions (discovery only)"),
]
def get_catalog_url(self) -> str:
"""Get the primary catalog URL.
Returns the URL of the highest-priority catalog. Kept for backward
compatibility. Use get_active_catalogs() for full multi-catalog support.
Returns:
URL of the primary catalog
Raises:
ValidationError: If a catalog URL is invalid
"""
active = self.get_active_catalogs()
return active[0].url if active else self.DEFAULT_CATALOG_URL
def _fetch_single_catalog(self, entry: CatalogEntry, force_refresh: bool = False) -> Dict[str, Any]:
"""Fetch a single catalog with per-URL caching.
For the DEFAULT_CATALOG_URL, uses legacy cache files (self.cache_file /
self.cache_metadata_file) for backward compatibility. For all other URLs,
uses URL-hash-based cache files in self.cache_dir.
Args:
entry: CatalogEntry describing the catalog to fetch
force_refresh: If True, bypass cache
Returns:
Catalog data dictionary
Raises:
ExtensionError: If catalog cannot be fetched or has invalid format
"""
import urllib.request
import urllib.error
# Determine cache file paths (backward compat for default catalog)
if entry.url == self.DEFAULT_CATALOG_URL:
cache_file = self.cache_file
cache_meta_file = self.cache_metadata_file
is_valid = not force_refresh and self.is_cache_valid()
else:
url_hash = hashlib.sha256(entry.url.encode()).hexdigest()[:16]
cache_file = self.cache_dir / f"catalog-{url_hash}.json"
cache_meta_file = self.cache_dir / f"catalog-{url_hash}-metadata.json"
is_valid = False
if not force_refresh and cache_file.exists() and cache_meta_file.exists():
try:
metadata = json.loads(cache_meta_file.read_text())
cached_at = datetime.fromisoformat(metadata.get("cached_at", ""))
if cached_at.tzinfo is None:
cached_at = cached_at.replace(tzinfo=timezone.utc)
age = (datetime.now(timezone.utc) - cached_at).total_seconds()
is_valid = age < self.CACHE_DURATION
except (json.JSONDecodeError, ValueError, KeyError, TypeError):
# If metadata is invalid or missing expected fields, treat cache as invalid
pass
# Use cache if valid
if is_valid:
try:
return json.loads(cache_file.read_text())
except json.JSONDecodeError:
pass
# Fetch from network
try:
with urllib.request.urlopen(entry.url, timeout=10) as response:
catalog_data = json.loads(response.read())
if "schema_version" not in catalog_data or "extensions" not in catalog_data:
raise ExtensionError(f"Invalid catalog format from {entry.url}")
# Save to cache
self.cache_dir.mkdir(parents=True, exist_ok=True)
cache_file.write_text(json.dumps(catalog_data, indent=2))
cache_meta_file.write_text(json.dumps({
"cached_at": datetime.now(timezone.utc).isoformat(),
"catalog_url": entry.url,
}, indent=2))
return catalog_data
except urllib.error.URLError as e:
raise ExtensionError(f"Failed to fetch catalog from {entry.url}: {e}")
except json.JSONDecodeError as e:
raise ExtensionError(f"Invalid JSON in catalog from {entry.url}: {e}")
def _get_merged_extensions(self, force_refresh: bool = False) -> List[Dict[str, Any]]:
"""Fetch and merge extensions from all active catalogs.
Higher-priority (lower priority number) catalogs win on conflicts
(same extension id in two catalogs). Each extension dict is annotated with:
- _catalog_name: name of the source catalog
- _install_allowed: whether installation is allowed from this catalog
Catalogs that fail to fetch are skipped. Raises ExtensionError only if
ALL catalogs fail.
Args:
force_refresh: If True, bypass all caches
Returns:
List of merged extension dicts
Raises:
ExtensionError: If all catalogs fail to fetch
"""
import sys
active_catalogs = self.get_active_catalogs()
merged: Dict[str, Dict[str, Any]] = {}
any_success = False
for catalog_entry in active_catalogs:
try:
catalog_data = self._fetch_single_catalog(catalog_entry, force_refresh)
any_success = True
except ExtensionError as e:
print(
f"Warning: Could not fetch catalog '{catalog_entry.name}': {e}",
file=sys.stderr,
)
continue
for ext_id, ext_data in catalog_data.get("extensions", {}).items():
if ext_id not in merged: # Higher-priority catalog wins
merged[ext_id] = {
**ext_data,
"id": ext_id,
"_catalog_name": catalog_entry.name,
"_install_allowed": catalog_entry.install_allowed,
}
if not any_success and active_catalogs:
raise ExtensionError("Failed to fetch any extension catalog")
return list(merged.values())
def is_cache_valid(self) -> bool:
"""Check if cached catalog is still valid.
@@ -1053,9 +1284,11 @@ class ExtensionCatalog:
try:
metadata = json.loads(self.cache_metadata_file.read_text())
cached_at = datetime.fromisoformat(metadata.get("cached_at", ""))
if cached_at.tzinfo is None:
cached_at = cached_at.replace(tzinfo=timezone.utc)
age_seconds = (datetime.now(timezone.utc) - cached_at).total_seconds()
return age_seconds < self.CACHE_DURATION
except (json.JSONDecodeError, ValueError, KeyError):
except (json.JSONDecodeError, ValueError, KeyError, TypeError):
return False
def fetch_catalog(self, force_refresh: bool = False) -> Dict[str, Any]:
@@ -1116,7 +1349,7 @@ class ExtensionCatalog:
author: Optional[str] = None,
verified_only: bool = False,
) -> List[Dict[str, Any]]:
"""Search catalog for extensions.
"""Search catalog for extensions across all active catalogs.
Args:
query: Search query (searches name, description, tags)
@@ -1125,14 +1358,16 @@ class ExtensionCatalog:
verified_only: If True, show only verified extensions
Returns:
List of matching extension metadata
List of matching extension metadata, each annotated with
``_catalog_name`` and ``_install_allowed`` from its source catalog.
"""
catalog = self.fetch_catalog()
extensions = catalog.get("extensions", {})
all_extensions = self._get_merged_extensions()
results = []
for ext_id, ext_data in extensions.items():
for ext_data in all_extensions:
ext_id = ext_data["id"]
# Apply filters
if verified_only and not ext_data.get("verified", False):
continue
@@ -1158,25 +1393,26 @@ class ExtensionCatalog:
if query_lower not in searchable_text:
continue
results.append({"id": ext_id, **ext_data})
results.append(ext_data)
return results
def get_extension_info(self, extension_id: str) -> Optional[Dict[str, Any]]:
"""Get detailed information about a specific extension.
Searches all active catalogs in priority order.
Args:
extension_id: ID of the extension
Returns:
Extension metadata or None if not found
Extension metadata (annotated with ``_catalog_name`` and
``_install_allowed``) or None if not found.
"""
catalog = self.fetch_catalog()
extensions = catalog.get("extensions", {})
if extension_id in extensions:
return {"id": extension_id, **extensions[extension_id]}
all_extensions = self._get_merged_extensions()
for ext_data in all_extensions:
if ext_data["id"] == extension_id:
return ext_data
return None
def download_extension(self, extension_id: str, target_dir: Optional[Path] = None) -> Path:
@@ -1236,11 +1472,18 @@ class ExtensionCatalog:
raise ExtensionError(f"Failed to save extension ZIP: {e}")
def clear_cache(self):
"""Clear the catalog cache."""
"""Clear the catalog cache (both legacy and URL-hash-based files)."""
if self.cache_file.exists():
self.cache_file.unlink()
if self.cache_metadata_file.exists():
self.cache_metadata_file.unlink()
# Also clear any per-URL hash-based cache files
if self.cache_dir.exists():
for extra_cache in self.cache_dir.glob("catalog-*.json"):
if extra_cache != self.cache_file:
extra_cache.unlink(missing_ok=True)
for extra_meta in self.cache_dir.glob("catalog-*-metadata.json"):
extra_meta.unlink(missing_ok=True)
class ConfigManager: