bvandeusen
524fd1f509
The schedule and host run forms opened their Source/Playbook <select>s on a "— choose … —" placeholder, forcing an extra click. Default them to the first item and cascade the dependent fields so the initial state is real: - Remove the placeholder options (schedules source+playbook, host source+ playbook, and the shared _playbook_options fragment). - schedules(): pass sel_source (first source, or the edited schedule's) and that source's pre-rendered playbooks so the playbook dropdown starts populated (create mode too, no longer the all-sources union). - Cascade on load/change: the source <select> fires playbook-options then (hx-on::after-settle) re-triggers the playbook <select>, which loads that playbook's variable fields. Host form drives the whole chain from a load trigger; schedule create mode loads vars via the playbook <select>'s own load trigger, while edit mode keeps its server-rendered prefill untouched. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
619 lines
23 KiB
Python
619 lines
23 KiB
Python
# steward/ansible/routes.py
|
|
from __future__ import annotations
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from quart import (
|
|
Blueprint, Response, current_app, redirect, render_template,
|
|
request, session, url_for,
|
|
)
|
|
from sqlalchemy import select
|
|
|
|
from steward.ansible import executor, sources as src_module
|
|
from steward.auth.middleware import require_role
|
|
from steward.models.ansible import AnsibleRun, AnsibleRunStatus
|
|
from steward.models.ansible_schedule import AnsibleSchedule
|
|
from steward.models.users import User, UserRole
|
|
|
|
INTERVAL_PRESETS = [
|
|
(300, "Every 5 minutes"),
|
|
(3600, "Hourly"),
|
|
(21600, "Every 6 hours"),
|
|
(43200, "Every 12 hours"),
|
|
(86400, "Daily"),
|
|
(604800, "Weekly"),
|
|
]
|
|
|
|
ansible_bp = Blueprint("ansible", __name__, url_prefix="/ansible")
|
|
|
|
|
|
def _get_sources() -> list[dict]:
|
|
ansible_cfg = current_app.config.get("ANSIBLE", {})
|
|
return src_module.get_sources(ansible_cfg)
|
|
|
|
|
|
@ansible_bp.get("/")
|
|
@require_role(UserRole.viewer)
|
|
async def index():
|
|
async with current_app.db_sessionmaker() as db:
|
|
result = await db.execute(
|
|
select(AnsibleRun).order_by(AnsibleRun.started_at.desc()).limit(50)
|
|
)
|
|
runs = result.scalars().all()
|
|
return await render_template("ansible/index.html", runs=runs)
|
|
|
|
|
|
@ansible_bp.get("/browse")
|
|
@require_role(UserRole.viewer)
|
|
async def browse():
|
|
from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup
|
|
|
|
all_sources = _get_sources()
|
|
source_data = []
|
|
for source in all_sources:
|
|
playbooks = []
|
|
for path in src_module.discover_playbooks(source["path"]):
|
|
contents = src_module.read_playbook(source["path"], path)
|
|
meta = (src_module.discover_playbook_meta(contents) if contents is not None
|
|
else {"description": "", "category": ""})
|
|
playbooks.append({
|
|
"path": path,
|
|
"category": meta["category"],
|
|
"description": meta["description"],
|
|
})
|
|
inventories = src_module.discover_inventories(source["path"])
|
|
source_data.append({
|
|
"source": source,
|
|
"playbooks": playbooks,
|
|
"inventories": inventories,
|
|
"editable": src_module.is_editable_source(source),
|
|
})
|
|
|
|
async with current_app.db_sessionmaker() as db:
|
|
targets = (await db.execute(
|
|
select(AnsibleTarget).order_by(AnsibleTarget.name)
|
|
)).scalars().all()
|
|
groups = (await db.execute(
|
|
select(AnsibleGroup).order_by(AnsibleGroup.name)
|
|
)).scalars().all()
|
|
|
|
return await render_template(
|
|
"ansible/browse.html",
|
|
source_data=source_data,
|
|
targets=targets,
|
|
groups=groups,
|
|
)
|
|
|
|
|
|
@ansible_bp.get("/browse/<source_name>/<path:playbook_path>")
|
|
@require_role(UserRole.viewer)
|
|
async def view_playbook(source_name: str, playbook_path: str):
|
|
all_sources = _get_sources()
|
|
source = next((s for s in all_sources if s["name"] == source_name), None)
|
|
if source is None:
|
|
return "Source not found", 404
|
|
contents = src_module.read_playbook(source["path"], playbook_path)
|
|
if contents is None:
|
|
return "Playbook not found", 404
|
|
return await render_template(
|
|
"ansible/browse.html",
|
|
source_data=[],
|
|
view_source=source_name,
|
|
view_path=playbook_path,
|
|
view_contents=contents,
|
|
view_editable=src_module.is_editable_source(source),
|
|
)
|
|
|
|
|
|
def _parse_run_params(form) -> tuple[dict | None, dict | None, str | None]:
|
|
"""Build executor params + the unpersisted secret-vars dict from form fields.
|
|
|
|
Returns (params, secret_vars, error). Discovered variable fields are named
|
|
``var__<name>``; a sibling hidden ``secret__<name>`` marks the sensitive
|
|
ones, which go into secret_vars (never persisted) instead of params. A
|
|
free-form ``extra_vars`` textarea (key=value per line) is also folded in.
|
|
All run-time values flow through a JSON extra-vars file downstream, so
|
|
values may contain spaces/quotes safely.
|
|
"""
|
|
extra_vars_map: dict[str, str] = {}
|
|
secret_vars: dict[str, str] = {}
|
|
|
|
for key in set(form.keys()):
|
|
if not key.startswith("var__"):
|
|
continue
|
|
name = key[len("var__"):]
|
|
val = (form.get(key) or "").strip()
|
|
if not val:
|
|
continue # untouched → fall through to inventory/play default
|
|
if f"secret__{name}" in form:
|
|
secret_vars[name] = val
|
|
else:
|
|
extra_vars_map[name] = val
|
|
|
|
for line in (form.get("extra_vars", "") or "").splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if "=" not in line:
|
|
return None, None, f"Invalid extra var (expected key=value): {line!r}"
|
|
k, v = line.split("=", 1)
|
|
extra_vars_map[k.strip()] = v
|
|
|
|
params: dict = {}
|
|
if extra_vars_map:
|
|
params["extra_vars_map"] = extra_vars_map
|
|
limit = (form.get("limit", "") or "").strip()
|
|
if limit:
|
|
params["limit"] = limit
|
|
tags = (form.get("tags", "") or "").strip()
|
|
if tags:
|
|
params["tags"] = tags
|
|
if "check" in form:
|
|
params["check"] = True
|
|
return (params or None), (secret_vars or None), None
|
|
|
|
|
|
@ansible_bp.get("/playbook-options")
|
|
@require_role(UserRole.operator)
|
|
async def playbook_options():
|
|
"""HTMX fragment: <option>s of playbooks in a source, to populate a playbook
|
|
dropdown when a source is chosen. `selected` pre-selects one (edit forms)."""
|
|
source_name = (request.args.get("source_name", "") or "").strip()
|
|
selected = (request.args.get("selected", "") or "").strip()
|
|
source = next((s for s in _get_sources() if s["name"] == source_name), None)
|
|
playbooks = src_module.discover_playbooks(source["path"]) if source else []
|
|
return await render_template(
|
|
"ansible/_playbook_options.html", playbooks=playbooks, selected=selected)
|
|
|
|
|
|
@ansible_bp.get("/playbook-vars")
|
|
@require_role(UserRole.operator)
|
|
async def playbook_vars():
|
|
"""HTMX fragment: fill-in fields for a playbook's declared variables
|
|
(vars/vars_prompt), loaded when a playbook is chosen from a dropdown."""
|
|
source_name = (request.args.get("source_name", "") or "").strip()
|
|
playbook_path = (request.args.get("playbook_path", "") or "").strip()
|
|
# Schedule form: schedules can't prompt, so secret vars render disabled.
|
|
schedule = (request.args.get("schedule", "") or "").strip() == "1"
|
|
source = next((s for s in _get_sources() if s["name"] == source_name), None)
|
|
variables: list = []
|
|
meta: dict = {"description": "", "category": "", "confirm": False}
|
|
if source and playbook_path:
|
|
contents = src_module.read_playbook(source["path"], playbook_path)
|
|
if contents is not None:
|
|
variables = src_module.discover_playbook_variables(contents)
|
|
meta = src_module.discover_playbook_meta(contents)
|
|
return await render_template(
|
|
"ansible/_playbook_vars.html", variables=variables,
|
|
description=meta["description"], category=meta["category"],
|
|
confirm=meta["confirm"], schedule=schedule)
|
|
|
|
|
|
@ansible_bp.get("/run-form/<source_name>/<path:playbook_path>")
|
|
@require_role(UserRole.operator)
|
|
async def run_form(source_name: str, playbook_path: str):
|
|
"""HTMX fragment: the run form for one playbook, with a field per declared
|
|
variable (vars/vars_prompt) discovered from the playbook itself."""
|
|
from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup
|
|
|
|
source = next((s for s in _get_sources() if s["name"] == source_name), None)
|
|
if source is None:
|
|
return "Source not found", 404
|
|
contents = src_module.read_playbook(source["path"], playbook_path)
|
|
if contents is None:
|
|
return "Playbook not found", 404
|
|
variables = src_module.discover_playbook_variables(contents)
|
|
meta = src_module.discover_playbook_meta(contents)
|
|
inventories = src_module.discover_inventories(source["path"])
|
|
|
|
async with current_app.db_sessionmaker() as db:
|
|
targets = (await db.execute(
|
|
select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all()
|
|
groups = (await db.execute(
|
|
select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all()
|
|
|
|
return await render_template(
|
|
"ansible/_run_form.html",
|
|
source_name=source_name, playbook_path=playbook_path,
|
|
variables=variables, description=meta["description"], category=meta["category"],
|
|
confirm=meta["confirm"], targets=targets, groups=groups,
|
|
inventories=inventories,
|
|
)
|
|
|
|
|
|
@ansible_bp.post("/runs")
|
|
@require_role(UserRole.operator)
|
|
async def create_run():
|
|
from steward.ansible.runner import trigger_run
|
|
|
|
form = await request.form
|
|
playbook_path = form.get("playbook_path", "").strip()
|
|
source_name = form.get("source_name", "").strip()
|
|
inventory_scope = form.get("inventory_scope", "steward:all").strip()
|
|
|
|
if not playbook_path:
|
|
return "playbook_path is required", 400
|
|
|
|
params_or_none, secret_vars, err = _parse_run_params(form)
|
|
if err:
|
|
return err, 400
|
|
|
|
run, source, err = await trigger_run(
|
|
current_app._get_current_object(), # type: ignore[attr-defined]
|
|
source_name=source_name,
|
|
playbook_path=playbook_path,
|
|
inventory_scope=inventory_scope,
|
|
params=params_or_none,
|
|
secret_vars=secret_vars,
|
|
triggered_by=session["user_id"],
|
|
)
|
|
if err == "Source not found":
|
|
return err, 404
|
|
if err:
|
|
return err, 400
|
|
|
|
return await render_template(
|
|
"ansible/run_started.html",
|
|
run=run,
|
|
source=source,
|
|
)
|
|
|
|
|
|
@ansible_bp.get("/runs/<run_id>/stream")
|
|
@require_role(UserRole.viewer)
|
|
async def stream_run(run_id: str):
|
|
async with current_app.db_sessionmaker() as db:
|
|
result = await db.execute(select(AnsibleRun).where(AnsibleRun.id == run_id))
|
|
run = result.scalar_one_or_none()
|
|
if run is None:
|
|
return "Not found", 404
|
|
|
|
async def generate():
|
|
if run.status not in (AnsibleRunStatus.running, AnsibleRunStatus.queued):
|
|
yield f"event: done\ndata: {run.status.value}\n\n"
|
|
return
|
|
|
|
q = executor.register_listener(run_id)
|
|
try:
|
|
while True:
|
|
msg = await q.get()
|
|
if isinstance(msg, executor._Done):
|
|
yield f"event: done\ndata: {msg.status}\n\n"
|
|
break
|
|
# Escape newlines in SSE data field
|
|
safe = msg.replace("\n", " ")
|
|
yield f"event: output\ndata: {safe}\n\n"
|
|
finally:
|
|
executor.deregister_listener(run_id, q)
|
|
|
|
return Response(
|
|
generate(),
|
|
content_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
)
|
|
|
|
|
|
@ansible_bp.get("/runs/<run_id>")
|
|
@require_role(UserRole.viewer)
|
|
async def run_detail(run_id: str):
|
|
async with current_app.db_sessionmaker() as db:
|
|
result = await db.execute(select(AnsibleRun).where(AnsibleRun.id == run_id))
|
|
run = result.scalar_one_or_none()
|
|
if run is None:
|
|
return "Not found", 404
|
|
# NULL triggered_by = an automated (system) run; otherwise resolve the user.
|
|
triggered_label = "system"
|
|
if run.triggered_by:
|
|
res = await db.execute(
|
|
select(User.username).where(User.id == run.triggered_by))
|
|
triggered_label = res.scalar_one_or_none() or "(deleted user)"
|
|
return await render_template(
|
|
"ansible/run_detail.html", run=run, triggered_label=triggered_label)
|
|
|
|
|
|
@ansible_bp.post("/runs/<run_id>/cancel")
|
|
@require_role(UserRole.operator)
|
|
async def cancel_run(run_id: str):
|
|
async with current_app.db_sessionmaker() as db:
|
|
run = (await db.execute(
|
|
select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one_or_none()
|
|
if run is None:
|
|
return "Not found", 404
|
|
if run.status in (AnsibleRunStatus.running, AnsibleRunStatus.queued):
|
|
executor.cancel_run(run_id)
|
|
return redirect(url_for("ansible.run_detail", run_id=run_id))
|
|
|
|
|
|
@ansible_bp.get("/runs/<run_id>/log")
|
|
@require_role(UserRole.viewer)
|
|
async def run_log(run_id: str):
|
|
"""Download the full run log (persistent artifact; falls back to DB output)."""
|
|
import os
|
|
from steward.ansible.executor import artifact_path
|
|
|
|
async with current_app.db_sessionmaker() as db:
|
|
run = (await db.execute(
|
|
select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one_or_none()
|
|
if run is None:
|
|
return "Not found", 404
|
|
|
|
path = artifact_path(run_id)
|
|
if os.path.exists(path):
|
|
with open(path, encoding="utf-8", errors="replace") as f:
|
|
body = f.read()
|
|
else:
|
|
body = run.output or ""
|
|
return Response(
|
|
body, mimetype="text/plain",
|
|
headers={"Content-Disposition": f'attachment; filename="ansible-run-{run_id[:8]}.log"'},
|
|
)
|
|
|
|
|
|
# ── Scheduled recurring runs ──────────────────────────────────────────────────
|
|
|
|
@ansible_bp.get("/schedules")
|
|
@require_role(UserRole.viewer)
|
|
async def schedules():
|
|
from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup
|
|
|
|
editing_id = request.args.get("edit")
|
|
async with current_app.db_sessionmaker() as db:
|
|
scheds = (await db.execute(
|
|
select(AnsibleSchedule).order_by(AnsibleSchedule.name))).scalars().all()
|
|
groups = (await db.execute(
|
|
select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all()
|
|
targets = (await db.execute(
|
|
select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all()
|
|
run_ids = [s.last_run_id for s in scheds if s.last_run_id]
|
|
runs = {
|
|
r.id: r for r in (await db.execute(
|
|
select(AnsibleRun).where(AnsibleRun.id.in_(run_ids)))).scalars().all()
|
|
} if run_ids else {}
|
|
|
|
source_data = [
|
|
{"name": s["name"], "playbooks": src_module.discover_playbooks(s["path"])}
|
|
for s in _get_sources()
|
|
]
|
|
|
|
rows = []
|
|
editing = None
|
|
for s in scheds:
|
|
rows.append({
|
|
"s": s,
|
|
"last_run": runs.get(s.last_run_id) if s.last_run_id else None,
|
|
"next_run": (s.last_run_at + timedelta(seconds=s.interval_seconds)) if s.last_run_at else None,
|
|
})
|
|
if editing_id and s.id == editing_id:
|
|
editing = s
|
|
|
|
# Default the source dropdown to the first source (or the edited schedule's
|
|
# source), and pre-render that source's playbooks so the playbook dropdown
|
|
# starts on a real item rather than a placeholder.
|
|
sel_source = editing.source_name if editing else (source_data[0]["name"] if source_data else "")
|
|
sel_playbooks = next((sd["playbooks"] for sd in source_data if sd["name"] == sel_source), [])
|
|
|
|
# Edit mode: pre-render the selected playbook's declared variables so their
|
|
# saved values show (fresh loads fetch these via HTMX from /playbook-vars).
|
|
edit_variables: list = []
|
|
edit_meta: dict = {"description": "", "category": "", "confirm": False}
|
|
if editing:
|
|
src = next((s for s in _get_sources() if s["name"] == editing.source_name), None)
|
|
if src:
|
|
contents = src_module.read_playbook(src["path"], editing.playbook_path)
|
|
if contents is not None:
|
|
edit_variables = src_module.discover_playbook_variables(contents)
|
|
edit_meta = src_module.discover_playbook_meta(contents)
|
|
|
|
return await render_template(
|
|
"ansible/schedules.html",
|
|
rows=rows, groups=groups, targets=targets,
|
|
source_data=source_data, sel_source=sel_source, sel_playbooks=sel_playbooks,
|
|
editing=editing, interval_presets=INTERVAL_PRESETS,
|
|
edit_variables=edit_variables, edit_description=edit_meta["description"],
|
|
edit_category=edit_meta["category"], edit_confirm=edit_meta["confirm"],
|
|
)
|
|
|
|
|
|
def _schedule_form_fields(form) -> tuple[dict | None, str | None]:
|
|
"""Validate + extract schedule fields from a form. Returns (fields, error)."""
|
|
name = (form.get("name", "") or "").strip()
|
|
source_name = (form.get("source_name", "") or "").strip()
|
|
playbook_path = (form.get("playbook_path", "") or "").strip()
|
|
inventory_scope = (form.get("inventory_scope", "steward:all") or "steward:all").strip()
|
|
try:
|
|
interval = int(form.get("interval_seconds", "0"))
|
|
except (TypeError, ValueError):
|
|
interval = 0
|
|
if not name or not source_name or not playbook_path or interval <= 0:
|
|
return None, "name, source, playbook, and a positive interval are required"
|
|
# Scheduled runs can't prompt, so secret vars are intentionally dropped —
|
|
# automation should rely on global creds / inventory vars, not run-time secrets.
|
|
params, _secret_vars, err = _parse_run_params(form)
|
|
if err:
|
|
return None, err
|
|
return {
|
|
"name": name, "source_name": source_name, "playbook_path": playbook_path,
|
|
"inventory_scope": inventory_scope, "interval_seconds": interval, "params": params,
|
|
}, None
|
|
|
|
|
|
@ansible_bp.post("/schedules")
|
|
@require_role(UserRole.operator)
|
|
async def create_schedule():
|
|
form = await request.form
|
|
fields, err = _schedule_form_fields(form)
|
|
if err:
|
|
return err, 400
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
db.add(AnsibleSchedule(enabled=True, **fields))
|
|
return redirect(url_for("ansible.schedules"))
|
|
|
|
|
|
@ansible_bp.post("/schedules/<schedule_id>")
|
|
@require_role(UserRole.operator)
|
|
async def update_schedule(schedule_id: str):
|
|
form = await request.form
|
|
fields, err = _schedule_form_fields(form)
|
|
if err:
|
|
return err, 400
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
sched = await db.get(AnsibleSchedule, schedule_id)
|
|
if sched is None:
|
|
return "Not found", 404
|
|
for k, v in fields.items():
|
|
setattr(sched, k, v)
|
|
return redirect(url_for("ansible.schedules"))
|
|
|
|
|
|
@ansible_bp.post("/schedules/<schedule_id>/toggle")
|
|
@require_role(UserRole.operator)
|
|
async def toggle_schedule(schedule_id: str):
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
sched = await db.get(AnsibleSchedule, schedule_id)
|
|
if sched is None:
|
|
return "Not found", 404
|
|
sched.enabled = not sched.enabled
|
|
return redirect(url_for("ansible.schedules"))
|
|
|
|
|
|
@ansible_bp.post("/schedules/<schedule_id>/delete")
|
|
@require_role(UserRole.operator)
|
|
async def delete_schedule(schedule_id: str):
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
sched = await db.get(AnsibleSchedule, schedule_id)
|
|
if sched is not None:
|
|
await db.delete(sched)
|
|
return redirect(url_for("ansible.schedules"))
|
|
|
|
|
|
@ansible_bp.post("/schedules/<schedule_id>/run-now")
|
|
@require_role(UserRole.operator)
|
|
async def run_schedule_now(schedule_id: str):
|
|
from steward.ansible.runner import trigger_run
|
|
|
|
async with current_app.db_sessionmaker() as db:
|
|
sched = await db.get(AnsibleSchedule, schedule_id)
|
|
if sched is None:
|
|
return "Not found", 404
|
|
source_name, playbook_path = sched.source_name, sched.playbook_path
|
|
inventory_scope, params = sched.inventory_scope, sched.params
|
|
|
|
run, _source, err = await trigger_run(
|
|
current_app._get_current_object(), # type: ignore[attr-defined]
|
|
source_name=source_name, playbook_path=playbook_path,
|
|
inventory_scope=inventory_scope, params=params,
|
|
triggered_by=session["user_id"],
|
|
)
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
fresh = await db.get(AnsibleSchedule, schedule_id)
|
|
if fresh:
|
|
fresh.last_run_at = datetime.now(timezone.utc)
|
|
fresh.last_run_id = run.id if run else None
|
|
fresh.last_error = err
|
|
if run:
|
|
return redirect(url_for("ansible.run_detail", run_id=run.id))
|
|
return redirect(url_for("ansible.schedules"))
|
|
|
|
|
|
# ── In-app playbook authoring/editor (admin only) ─────────────────────────────
|
|
|
|
def _editable_sources() -> list[dict]:
|
|
return [s for s in _get_sources() if src_module.is_editable_source(s)]
|
|
|
|
|
|
def _editable_source(name: str) -> dict | None:
|
|
return next((s for s in _editable_sources() if s["name"] == name), None)
|
|
|
|
|
|
@ansible_bp.get("/playbooks/new")
|
|
@require_role(UserRole.admin)
|
|
async def new_playbook():
|
|
sources = _editable_sources()
|
|
return await render_template(
|
|
"ansible/playbook_editor.html",
|
|
editable_sources=sources, editing=False,
|
|
source_name=(sources[0]["name"] if sources else ""),
|
|
playbook_path="", content=_NEW_PLAYBOOK_TEMPLATE, error=None,
|
|
)
|
|
|
|
|
|
@ansible_bp.get("/playbooks/edit/<source_name>/<path:playbook_path>")
|
|
@require_role(UserRole.admin)
|
|
async def edit_playbook(source_name: str, playbook_path: str):
|
|
source = _editable_source(source_name)
|
|
if source is None:
|
|
return "Source is not editable", 404
|
|
content = src_module.read_playbook(source["path"], playbook_path)
|
|
if content is None:
|
|
return "Playbook not found", 404
|
|
return await render_template(
|
|
"ansible/playbook_editor.html",
|
|
editable_sources=_editable_sources(), editing=True,
|
|
source_name=source_name, playbook_path=playbook_path,
|
|
content=content, error=None,
|
|
)
|
|
|
|
|
|
@ansible_bp.post("/playbooks/save")
|
|
@require_role(UserRole.admin)
|
|
async def save_playbook():
|
|
form = await request.form
|
|
source_name = (form.get("source_name", "") or "").strip()
|
|
playbook_path = (form.get("playbook_path", "") or "").strip()
|
|
content = form.get("content", "") or ""
|
|
editing = (form.get("editing", "") == "1")
|
|
|
|
source = _editable_source(source_name)
|
|
err = None
|
|
if source is None:
|
|
err = "Source is not editable"
|
|
elif not playbook_path:
|
|
err = "Filename is required"
|
|
else:
|
|
ok, verr = src_module.validate_playbook_yaml(content)
|
|
if not ok:
|
|
err = verr
|
|
if err is None:
|
|
ok, werr = src_module.write_playbook(source["path"], playbook_path, content)
|
|
if not ok:
|
|
err = werr
|
|
|
|
if err is not None:
|
|
return await render_template(
|
|
"ansible/playbook_editor.html",
|
|
editable_sources=_editable_sources(), editing=editing,
|
|
source_name=source_name, playbook_path=playbook_path,
|
|
content=content, error=err,
|
|
), 400
|
|
|
|
return redirect(url_for(
|
|
"ansible.view_playbook", source_name=source_name, playbook_path=playbook_path))
|
|
|
|
|
|
@ansible_bp.post("/playbooks/delete")
|
|
@require_role(UserRole.admin)
|
|
async def delete_playbook():
|
|
form = await request.form
|
|
source = _editable_source((form.get("source_name", "") or "").strip())
|
|
if source is None:
|
|
return "Source is not editable", 400
|
|
ok, err = src_module.delete_playbook(source["path"], (form.get("playbook_path", "") or "").strip())
|
|
if not ok:
|
|
return err, 400
|
|
return redirect(url_for("ansible.browse"))
|
|
|
|
|
|
_NEW_PLAYBOOK_TEMPLATE = """\
|
|
---
|
|
- name: My playbook
|
|
hosts: all
|
|
gather_facts: false
|
|
tasks:
|
|
- name: Ping
|
|
ansible.builtin.ping:
|
|
"""
|