Skip to content

refactor(catalogs): extract integration catalog config loading #2497

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
PascalThuet wants to merge 1 commit into
base: main
Choose a base branch
from
+188 −135
Open

refactor(catalogs): extract integration catalog config loading #2497

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
180 changes: 180 additions & 0 deletions src/specify_cli/catalogs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,180 @@
"""Shared catalog stack config primitives.

Catalog-backed features use the same local config shape and URL validation
rules. This module keeps those narrow primitives in one place while individual
catalog types keep their active source resolution, fetch, cache, and
domain-specific validation behavior.
"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import ClassVar

import yaml


@dataclass
class CatalogEntry:
"""Represents a single catalog source in a catalog stack."""

url: str
name: str
priority: int
install_allowed: bool
description: str = ""


class CatalogStackBase:
"""Base class for ordered catalog-source resolution.

Subclasses provide catalog-specific metadata and exception classes. Fetching
and schema validation stay in each concrete catalog because those formats
differ across integrations, extensions, presets, and workflows.
"""

ENTRY_CLASS: ClassVar[type[CatalogEntry]] = CatalogEntry
ERROR_TYPE: ClassVar[type[Exception]] = ValueError
VALIDATION_ERROR_TYPE: ClassVar[type[Exception]] = ValueError

CONFIG_FILENAME: ClassVar[str]

@classmethod
def _error(cls, message: str) -> Exception:
return cls.ERROR_TYPE(message)

@classmethod
def _validation_error(cls, message: str) -> Exception:
return cls.VALIDATION_ERROR_TYPE(message)

@classmethod
def _entry(
cls,
*,
url: str,
name: str,
priority: int,
install_allowed: bool,
description: str = "",
) -> CatalogEntry:
return cls.ENTRY_CLASS(
url=url,
name=name,
priority=priority,
install_allowed=install_allowed,
description=description,
)

@classmethod
def _validate_catalog_url(cls, url: str) -> None:
"""Validate that a catalog URL uses HTTPS, except localhost HTTP."""
from urllib.parse import urlparse

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 cls._error(
f"Catalog URL must use HTTPS (got {parsed.scheme}://). "
"HTTP is only allowed for localhost."
)
if not parsed.netloc:
raise cls._error("Catalog URL must be a valid URL with a host.")

def _load_catalog_config(self, config_path: Path) -> list[CatalogEntry] | None:
"""Load catalog stack configuration from a YAML file.

Returns ``None`` when the file does not exist. Existing files fail
closed when they are malformed, empty, or contain no usable URLs.
"""
if not config_path.exists():
return None
try:
data = yaml.safe_load(config_path.read_text(encoding="utf-8"))
except (yaml.YAMLError, OSError, UnicodeError) as exc:
raise self._validation_error(
f"Failed to read catalog config {config_path}: {exc}"
) from exc
if data is None:
data = {}
if not isinstance(data, dict):
raise self._validation_error(
f"Invalid catalog config {config_path}: expected a YAML mapping at the root"
)

catalogs_data = data.get("catalogs", [])
if not isinstance(catalogs_data, list):
raise self._validation_error(
f"Invalid catalog config {config_path}: 'catalogs' must be a list, "
f"got {type(catalogs_data).__name__}"
)
if not catalogs_data:
raise self._validation_error(
f"Catalog config {config_path} exists but contains no 'catalogs' entries. "
f"Remove the file to use built-in defaults, or add valid catalog entries."
)

entries: list[CatalogEntry] = []
skipped: list[int] = []
for idx, item in enumerate(catalogs_data):
if not isinstance(item, dict):
raise self._validation_error(
f"Invalid catalog config {config_path}: catalog entry at index {idx}: "
f"expected a mapping, got {type(item).__name__}"
)
url = str(item.get("url", "")).strip()
if not url:
skipped.append(idx)
continue
try:
self._validate_catalog_url(url)
except self.ERROR_TYPE as exc:
raise self._validation_error(
f"Invalid catalog URL in {config_path} at index {idx}: {exc}"
) from exc

raw_priority = item.get("priority", idx + 1)
if isinstance(raw_priority, bool):
raise self._validation_error(
f"Invalid catalog config {config_path}: "
f"Invalid priority for catalog '{item.get('name', idx + 1)}': "
f"expected integer, got {raw_priority!r}"
)
try:
priority = int(raw_priority)
except (TypeError, ValueError):
raise self._validation_error(
f"Invalid catalog config {config_path}: "
f"Invalid priority for catalog '{item.get('name', idx + 1)}': "
f"expected integer, got {raw_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)

raw_name = item.get("name")
name = str(raw_name).strip() if raw_name is not None else ""
if not name:
name = f"catalog-{len(entries) + 1}"

entries.append(
self._entry(
url=url,
name=name,
priority=priority,
install_allowed=install_allowed,
description=str(item.get("description", "")),
)
)

entries.sort(key=lambda e: e.priority)
if not entries:
raise self._validation_error(
f"Catalog config {config_path} contains {len(catalogs_data)} "
f"entries but none have valid URLs (entries at indices {skipped} "
f"were skipped). Each catalog entry must have a 'url' field."
)
return entries
143 changes: 8 additions & 135 deletions src/specify_cli/integrations/catalog.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
import yaml
from packaging import version as pkg_version

from ..catalogs import CatalogEntry, CatalogStackBase


# ---------------------------------------------------------------------------
# Errors
Expand All @@ -43,21 +45,15 @@ class IntegrationDescriptorError(Exception):
# ---------------------------------------------------------------------------

@dataclass
class IntegrationCatalogEntry:
class IntegrationCatalogEntry(CatalogEntry):
"""Represents a single catalog source in the catalog stack."""

url: str
name: str
priority: int
install_allowed: bool
description: str = ""


# ---------------------------------------------------------------------------
# IntegrationCatalog
# ---------------------------------------------------------------------------

class IntegrationCatalog:
class IntegrationCatalog(CatalogStackBase):
"""Manages integration catalog fetching, caching, and searching."""

DEFAULT_CATALOG_URL = (
Expand All @@ -67,136 +63,15 @@ class IntegrationCatalog:
"https://raw.githubusercontent.com/github/spec-kit/main/integrations/catalog.community.json"
)
CACHE_DURATION = 3600 # 1 hour
CONFIG_FILENAME = "integration-catalogs.yml"
ENTRY_CLASS = IntegrationCatalogEntry
ERROR_TYPE = IntegrationCatalogError
VALIDATION_ERROR_TYPE = IntegrationValidationError

def __init__(self, project_root: Path) -> None:
self.project_root = project_root
self.cache_dir = project_root / ".specify" / "integrations" / ".cache"

# -- URL validation ---------------------------------------------------

@staticmethod
def _validate_catalog_url(url: str) -> None:
from urllib.parse import urlparse

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 IntegrationCatalogError(
f"Catalog URL must use HTTPS (got {parsed.scheme}://). "
"HTTP is only allowed for localhost."
)
if not parsed.netloc:
raise IntegrationCatalogError(
"Catalog URL must be a valid URL with a host."
)

# -- Catalog stack ----------------------------------------------------

def _load_catalog_config(
self, config_path: Path
) -> Optional[List[IntegrationCatalogEntry]]:
"""Load catalog stack from a YAML file.

Returns None when the file does not exist.

Raises:
IntegrationValidationError: on any local-config / YAML problem
(parse failures, wrong shape, missing/invalid fields,
invalid catalog URLs, etc.). This is a subclass of
:class:`IntegrationCatalogError`, so any caller that already
catches ``IntegrationCatalogError`` keeps working — but
callers that want to distinguish *local config* problems
from *remote/network* problems can match the subclass.
"""
if not config_path.exists():
return None
try:
data = yaml.safe_load(config_path.read_text(encoding="utf-8"))
except (yaml.YAMLError, OSError, UnicodeError) as exc:
raise IntegrationValidationError(
f"Failed to read catalog config {config_path}: {exc}"
) from exc
if data is None:
data = {}
if not isinstance(data, dict):
raise IntegrationValidationError(
f"Invalid catalog config {config_path}: expected a YAML mapping at the root"
)
catalogs_data = data.get("catalogs", [])
if not isinstance(catalogs_data, list):
raise IntegrationValidationError(
f"Invalid catalog config {config_path}: 'catalogs' must be a list, "
f"got {type(catalogs_data).__name__}"
)
if not catalogs_data:
raise IntegrationValidationError(
f"Catalog config {config_path} exists but contains no 'catalogs' entries. "
f"Remove the file to use built-in defaults, or add valid catalog entries."
)
entries: List[IntegrationCatalogEntry] = []
skipped: List[int] = []
for idx, item in enumerate(catalogs_data):
if not isinstance(item, dict):
raise IntegrationValidationError(
f"Invalid catalog config {config_path}: catalog entry at index {idx}: "
f"expected a mapping, got {type(item).__name__}"
)
url = str(item.get("url", "")).strip()
if not url:
skipped.append(idx)
continue
try:
self._validate_catalog_url(url)
except IntegrationCatalogError as exc:
# ``_validate_catalog_url`` raises the base class for direct
# callers (e.g. ``add_catalog`` validating user input); when
# the bad URL came from a local config file, surface it as a
# validation error so CLI handlers can route it accordingly.
raise IntegrationValidationError(
f"Invalid catalog URL in {config_path} at index {idx}: {exc}"
) from exc
raw_priority = item.get("priority", idx + 1)
if isinstance(raw_priority, bool):
raise IntegrationValidationError(
f"Invalid catalog config {config_path}: "
f"Invalid priority for catalog '{item.get('name', idx + 1)}': "
f"expected integer, got {raw_priority!r}"
)
try:
priority = int(raw_priority)
except (TypeError, ValueError):
raise IntegrationValidationError(
f"Invalid catalog config {config_path}: "
f"Invalid priority for catalog '{item.get('name', idx + 1)}': "
f"expected integer, got {raw_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)
raw_name = item.get("name")
name = str(raw_name).strip() if raw_name is not None else ""
if not name:
name = f"catalog-{len(entries) + 1}"
entries.append(
IntegrationCatalogEntry(
url=url,
name=name,
priority=priority,
install_allowed=install_allowed,
description=str(item.get("description", "")),
)
)
entries.sort(key=lambda e: e.priority)
if not entries:
raise IntegrationValidationError(
f"Catalog config {config_path} contains {len(catalogs_data)} "
f"entries but none have valid URLs (entries at indices {skipped} "
f"were skipped). Each catalog entry must have a 'url' field."
)
return entries

def get_active_catalogs(self) -> List[IntegrationCatalogEntry]:
"""Return the ordered list of active integration catalogs.

Expand Down Expand Up @@ -444,8 +319,6 @@ def clear_cache(self) -> None:

# -- Catalog-source management ----------------------------------------

CONFIG_FILENAME = "integration-catalogs.yml"

def get_catalog_configs(self) -> List[Dict[str, Any]]:
"""Return the active catalog stack as a list of dicts.

Expand Down