feat: Ansible source manager UI — replace raw JSON textarea

- Replace the raw JSON textarea in Settings → Ansible with a proper
  per-source card UI: add, inline-edit, remove, and git-sync each
  source independently via HTMX without a page reload
- Each source card shows type badge (git/local), URL or path, branch,
  and pull interval; git sources get a Sync button that triggers
  git pull and returns inline status
- Add source form uses a type-aware select that shows URL/branch/interval
  fields for git or path field for local, collapsed in a details accordion
- Add new settings routes: sources/add, sources/<idx>/remove,
  sources/<idx>/save, sources/<idx>/sync
- Update browse.html and run_detail.html to use CSS vars throughout
  (removed hardcoded hex colors); link empty state to Settings page

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-23 17:24:35 -04:00
parent edc9b752d2
commit 56283df5a1
5 changed files with 338 additions and 48 deletions
+103 -16
View File
@@ -136,30 +136,117 @@ async def save_notifications():
# ── Ansible Sources ───────────────────────────────────────────────────────────
@settings_bp.get("/ansible/")
@require_role(UserRole.admin)
async def ansible():
async def _get_ansible_sources() -> list[dict]:
async with current_app.db_sessionmaker() as db:
settings = await get_all_settings(db)
return await render_template("settings/ansible.html", settings=settings)
sources = settings.get("ansible.sources", [])
return sources if isinstance(sources, list) else []
@settings_bp.post("/ansible/")
@require_role(UserRole.admin)
async def save_ansible():
form = await request.form
sources_raw = form.get("ansible.sources", "[]")
try:
sources = json.loads(sources_raw)
if not isinstance(sources, list):
sources = []
except json.JSONDecodeError:
sources = []
async def _save_ansible_sources(sources: list[dict]) -> None:
async with current_app.db_sessionmaker() as db:
async with db.begin():
await set_setting(db, "ansible.sources", sources)
await _reload_app_config()
return redirect(url_for("settings.ansible"))
async def _ansible_sources_partial(sources: list[dict]):
return await render_template("settings/_ansible_sources.html", sources=sources)
@settings_bp.get("/ansible/")
@require_role(UserRole.admin)
async def ansible():
sources = await _get_ansible_sources()
return await render_template("settings/ansible.html", sources=sources)
@settings_bp.post("/ansible/sources/add")
@require_role(UserRole.admin)
async def ansible_add_source():
form = await request.form
name = form.get("name", "").strip()
src_type = form.get("type", "local").strip()
if not name:
sources = await _get_ansible_sources()
return await _ansible_sources_partial(sources)
entry: dict = {"name": name, "type": src_type}
if src_type == "git":
entry["url"] = form.get("url", "").strip()
entry["branch"] = form.get("branch", "main").strip() or "main"
try:
entry["pull_interval_seconds"] = int(form.get("pull_interval_seconds", 3600))
except ValueError:
entry["pull_interval_seconds"] = 3600
else:
entry["path"] = form.get("path", "").strip()
sources = await _get_ansible_sources()
sources.append(entry)
await _save_ansible_sources(sources)
return await _ansible_sources_partial(sources)
@settings_bp.post("/ansible/sources/<int:idx>/remove")
@require_role(UserRole.admin)
async def ansible_remove_source(idx: int):
sources = await _get_ansible_sources()
if 0 <= idx < len(sources):
sources.pop(idx)
await _save_ansible_sources(sources)
return await _ansible_sources_partial(sources)
@settings_bp.post("/ansible/sources/<int:idx>/save")
@require_role(UserRole.admin)
async def ansible_save_source(idx: int):
sources = await _get_ansible_sources()
if not (0 <= idx < len(sources)):
return await _ansible_sources_partial(sources)
form = await request.form
src_type = form.get("type", "local").strip()
entry: dict = {
"name": form.get("name", sources[idx].get("name", "")).strip(),
"type": src_type,
}
if src_type == "git":
entry["url"] = form.get("url", "").strip()
entry["branch"] = form.get("branch", "main").strip() or "main"
try:
entry["pull_interval_seconds"] = int(form.get("pull_interval_seconds", 3600))
except ValueError:
entry["pull_interval_seconds"] = 3600
else:
entry["path"] = form.get("path", "").strip()
sources[idx] = entry
await _save_ansible_sources(sources)
return await _ansible_sources_partial(sources)
@settings_bp.post("/ansible/sources/<int:idx>/sync")
@require_role(UserRole.admin)
async def ansible_sync_source(idx: int):
"""Trigger a git pull for a git source and return a status badge."""
sources = await _get_ansible_sources()
if not (0 <= idx < len(sources)):
return ('<span style="color:var(--red);font-size:0.8rem;">Source not found</span>', 404)
source = sources[idx]
if source.get("type") != "git":
return '<span style="color:var(--text-muted);font-size:0.8rem;">Not a git source</span>'
from fabledscryer.ansible.sources import git_pull
from fabledscryer.core.settings import to_ansible_cfg
async with current_app.db_sessionmaker() as db:
settings = await get_all_settings(db)
ansible_cfg = to_ansible_cfg(settings)
from fabledscryer.ansible.sources import get_sources
resolved = next((s for s in get_sources(ansible_cfg) if s["name"] == source["name"]), None)
if resolved is None:
return '<span style="color:var(--red);font-size:0.8rem;">Could not resolve source path</span>'
try:
await git_pull(resolved)
return '<span style="color:var(--green);font-size:0.8rem;">Synced</span>'
except Exception as exc:
logger.exception("git pull failed for source %r", source["name"])
return f'<span style="color:var(--red);font-size:0.8rem;">Sync failed: {exc}</span>'
# ── Auth (OIDC / LDAP) ────────────────────────────────────────────────────────