Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
57 changes: 54 additions & 3 deletions fixtures/catalog/templates.json
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,59 @@
{
"$schema": "https://raw.githubusercontent.com/Create-Python-App/cpa-templates/main/templates.schema.json",
"categories": [
{
"slug": "backend-applications",
"name": "Backend Applications",
"description": "API and service starters for FastAPI and similar Python backends.",
"details": "Use when the deliverable is an HTTP API or background worker.",
"labels": ["Backend", "API", "Python", "FastAPI"]
},
{
"slug": "tooling",
"name": "Tooling",
"description": "Extensions that add CI, containers, databases, and developer ergonomics.",
"details": "Layer these on top of a compatible template.",
"labels": ["DevOps", "CI", "Docker", "Tooling"]
}
],
"templates": [
{"slug": "example-cli", "category": "cli", "url": "file://."}
{
"name": "FastAPI Starter",
"slug": "fastapi-starter",
"description": "Production-ready FastAPI API with uv, Ruff, pytest, and pydantic-settings",
"url": "https://github.com/Create-Python-App/cpa-templates?subdir=templates/fastapi-starter",
"type": "fastapi-backend",
"category": "backend-applications",
"labels": ["FastAPI", "API", "Python", "uv", "Backend"]
},
{
"name": "Example CLI",
"slug": "example-cli",
"description": "Minimal local fixture template for tests",
"url": "file://.",
"type": "cli",
"category": "tooling",
"labels": ["Example"]
}
],
"addons": [
{"slug": "ruff-setup", "category": "tooling"}
"extensions": [
{
"name": "Ruff Setup",
"slug": "ruff-setup",
"description": "Local fixture extension for tests",
"url": "file://.",
"type": ["cli", "fastapi-backend"],
"category": "tooling",
"labels": ["Ruff"]
},
{
"name": "GitHub Setup",
"slug": "github-setup",
"description": "GitHub Actions CI, Dependabot, issue templates",
"url": "https://github.com/Create-Python-App/cpa-templates?subdir=extensions/github-setup",
"type": ["fastapi-backend"],
"category": "tooling",
"labels": ["GitHub", "CI"]
}
]
}
Original file line numberDiff line numberDiff line change
@@ -1,45 +1,182 @@
"""Template catalog listing (stub URL / fixtures)."""
"""Template catalog fetch and listing."""

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

from create_python_app_core.paths import default_cache_dir, resolve_source
from rich.console import Console
from rich.table import Table

console = Console()
from create_awesome_python_app import __version__

console = Console(stderr=True)

DEFAULT_CATALOG_URL = (
"https://raw.githubusercontent.com/Create-Python-App/cpa-templates/main/templates.json"
)
CACHE_TTL_SECONDS = 3600
FETCH_TIMEOUT_SECONDS = 10
USER_AGENT = f"create-awesome-python-app/{__version__} (https://github.com/Create-Python-App/create-python-app)"

_FIXTURE = (
Path(__file__).resolve().parents[4] / "fixtures" / "catalog" / "templates.json"
)

_memory_cache: dict[str, Any] | None = None
_memory_ts: float = 0.0


def catalog_url() -> str:
return os.environ.get("CPA_CATALOG_URL", DEFAULT_CATALOG_URL)


def catalog_cache_path() -> Path:
return default_cache_dir() / "catalog" / "templates.json"

def _load() -> dict:

def _read_json_file(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))


def _read_fixture() -> dict[str, Any]:
if _FIXTURE.is_file():
return json.loads(_FIXTURE.read_text(encoding="utf-8"))
return {
"templates": [{"slug": "example-cli", "category": "cli", "url": "file://."}],
"addons": [{"slug": "ruff-setup", "category": "tooling"}],
}
return _read_json_file(_FIXTURE)
return {"templates": [], "extensions": [], "categories": []}


def _read_disk_cache() -> dict[str, Any] | None:
path = catalog_cache_path()
if not path.is_file():
return None
try:
return _read_json_file(path)
except json.JSONDecodeError:
return None


def _write_disk_cache(data: dict[str, Any]) -> None:
path = catalog_cache_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data), encoding="utf-8")


def _fetch_file_url(url: str) -> dict[str, Any]:
source = resolve_source(url)
if source.local_path is None:
raise OSError(f"Invalid file catalog URL: {url}")
base = source.local_path
if source.subdir:
base = base / source.subdir
catalog_file = base / "templates.json"
if not catalog_file.is_file():
raise FileNotFoundError(f"Catalog not found: {catalog_file}")
return _read_json_file(catalog_file)


def _fetch_remote(url: str) -> dict[str, Any]:
if url.startswith("file://"):
return _fetch_file_url(url)
req = urllib.request.Request(
url,
headers={"Accept": "application/json", "User-Agent": USER_AGENT},
)
with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT_SECONDS) as resp:
payload = resp.read().decode("utf-8")
return json.loads(payload)


def get_catalog_data(*, force_refresh: bool = False) -> dict[str, Any]:
"""Load templates.json from remote URL, disk cache, or local fixture."""
global _memory_cache, _memory_ts

if (
not force_refresh
and _memory_cache is not None
and os.environ.get("CPA_NO_CATALOG_CACHE") != "1"
and time.time() - _memory_ts <= CACHE_TTL_SECONDS
):
return _memory_cache

if os.environ.get("CPA_CATALOG_FIXTURE") == "1":
data = _read_fixture()
else:
url = catalog_url()
try:
data = _fetch_remote(url)
_write_disk_cache(data)
except (urllib.error.URLError, TimeoutError, OSError, json.JSONDecodeError) as err:
disk = _read_disk_cache()
if disk is not None:
console.print(
f"[yellow][cpa] Could not refresh catalog ({err}); using disk cache.[/yellow]"
)
data = disk
else:
fixture = _read_fixture()
if fixture.get("templates"):
console.print(
f"[yellow][cpa] Could not refresh catalog ({err}); using fixture.[/yellow]"
)
data = fixture
else:
raise RuntimeError(f"Failed to load template catalog: {err}") from err

_memory_cache = data
_memory_ts = time.time()
return data


def reset_catalog_cache_for_tests() -> None:
global _memory_cache, _memory_ts
_memory_cache = None
_memory_ts = 0.0


def list_templates() -> None:
data = _load()
data = get_catalog_data()
table = Table(title="Templates")
table.add_column("slug")
table.add_column("category")
table.add_column("type")
for t in data.get("templates", []):
table.add_row(t.get("slug", ""), t.get("category", ""))
table.add_row(
str(t.get("slug", "")),
str(t.get("category", "")),
str(t.get("type", "")),
)
console.print(table)


def list_addons(template_slug: str | None = None) -> None:
_ = template_slug
data = _load()
table = Table(title="Addons")
data = get_catalog_data()
template_type: str | None = None
if template_slug:
for t in data.get("templates", []):
if t.get("slug") == template_slug:
template_type = str(t.get("type", ""))
break

table = Table(title="Extensions")
table.add_column("slug")
table.add_column("category")
for t in data.get("addons", []):
table.add_row(t.get("slug", ""), t.get("category", ""))
table.add_column("type")
for ext in data.get("extensions", data.get("addons", [])):
ext_types = ext.get("type", [])
if isinstance(ext_types, str):
ext_types = [ext_types]
if template_type and template_type not in ext_types:
continue
type_label = ", ".join(ext_types) if isinstance(ext_types, list) else str(ext_types)
table.add_row(
str(ext.get("slug", "")),
str(ext.get("category", "")),
type_label,
)
console.print(table)
86 changes: 86 additions & 0 deletions packages/create-awesome-python-app/tests/test_catalog_fetch.py
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
"""Catalog fetch tests."""

from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import patch

import pytest

from create_awesome_python_app.catalog import (
DEFAULT_CATALOG_URL,
catalog_cache_path,
catalog_url,
get_catalog_data,
reset_catalog_cache_for_tests,
)

FIXTURE_PATH = (
Path(__file__).resolve().parents[3] / "fixtures" / "catalog" / "templates.json"
)


@pytest.fixture(autouse=True)
def _reset_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
reset_catalog_cache_for_tests()
monkeypatch.setenv("CPA_CACHE_DIR", str(tmp_path / "cache"))


def test_default_catalog_url(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("CPA_CATALOG_URL", raising=False)
assert "Create-Python-App/cpa-templates" in catalog_url()
monkeypatch.setenv("CPA_CATALOG_URL", "https://example.com/templates.json")
assert catalog_url() == "https://example.com/templates.json"


def test_get_catalog_data_fetches_and_caches(tmp_path: Path) -> None:
payload = json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))

class FakeResponse:
def read(self) -> bytes:
return json.dumps(payload).encode("utf-8")

def __enter__(self) -> "FakeResponse":
return self

def __exit__(self, *args: object) -> None:
return None

with patch(
"create_awesome_python_app.catalog.urllib.request.urlopen",
return_value=FakeResponse(),
):
data = get_catalog_data(force_refresh=True)

assert any(t["slug"] == "fastapi-starter" for t in data["templates"])
assert catalog_cache_path().is_file()


def test_get_catalog_data_fixture_fallback(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("CPA_CATALOG_FIXTURE", "1")
data = get_catalog_data(force_refresh=True)
assert data["templates"]


def test_get_catalog_data_disk_fallback_on_network_error(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
payload = {"templates": [{"slug": "cached"}], "extensions": [], "categories": []}
cache_file = catalog_cache_path()
cache_file.parent.mkdir(parents=True, exist_ok=True)
cache_file.write_text(json.dumps(payload), encoding="utf-8")

with patch(
"create_awesome_python_app.catalog._fetch_remote",
side_effect=OSError("network down"),
):
data = get_catalog_data(force_refresh=True)

assert data["templates"][0]["slug"] == "cached"


def test_default_url_points_to_cpa_templates() -> None:
assert DEFAULT_CATALOG_URL.endswith("/cpa-templates/main/templates.json")
Loading