"""Flask application.""" from __future__ import annotations import ipaddress from datetime import datetime from pathlib import Path from flask import Flask, abort, redirect, render_template, request, send_file, url_for from .config import Settings from .discovery import TaskCatalog from .inputs import validate_inputs, validate_uploads from .runner import RunManager from .scheduler import BackgroundScheduler from .store import RunStore def create_app(settings: Settings | None = None) -> Flask: settings = settings or Settings.from_environment() catalog = TaskCatalog(settings.task_root) store = RunStore(settings.database_path) store.initialize() store.mark_interrupted() manager = RunManager(settings, catalog, store) manager.cleanup_transient_files() scheduler = BackgroundScheduler(catalog, manager) scheduler.start() project_root = Path(__file__).resolve().parent.parent app = Flask( __name__, template_folder=str(project_root / "templates"), static_folder=str(project_root / "static"), ) app.config["settings"] = settings app.extensions["catalog"] = catalog app.extensions["run_manager"] = manager app.extensions["run_store"] = store app.extensions["background_scheduler"] = scheduler @app.before_request def allow_only_proxy() -> None: if not settings.allowed_proxy_ips: return remote_address = request.remote_addr if remote_address is None: abort(403) try: normalized = str(ipaddress.ip_address(remote_address)) except ValueError: abort(403) if normalized not in settings.allowed_proxy_ips: abort(403) @app.template_filter("timestamp") def timestamp(value: str | None) -> str: if not value: return "—" return datetime.fromisoformat(value).astimezone().strftime("%Y-%m-%d %H:%M:%S %Z") def render_index( *, form_values: dict[str, dict[str, list[str]]] | None = None, input_errors: dict[str, dict[str, str]] | None = None, status: int = 200, ): groups = catalog.groups() latest = { ("group", group.id): store.last_run("group", group.id) for group in groups } for group in groups: for task in group.tasks: latest[("task", task.id)] = store.last_run("task", task.id) return ( render_template( "index.html", groups=groups, latest=latest, form_values=form_values or {}, input_errors=input_errors or {}, ), status, ) @app.get("/") def index(): return render_index() @app.post("/tasks//run") def run_task(task_id: str): task = catalog.task(task_id) if task is None: abort(404) submitted_values = { field.name: request.form.getlist(field.name) for field in task.inputs } input_data, errors = validate_inputs(task.inputs, submitted_values) uploads, upload_errors = validate_uploads(task.inputs, request.files) errors.update(upload_errors) if errors: if request.accept_mimetypes.best == "application/json": return {"errors": errors}, 400 return render_index( form_values={ task.id: { field.name: [] if field.sensitive else submitted_values[field.name] for field in task.inputs } }, input_errors={task.id: errors}, status=400, ) run, _ = manager.run_task(task, input_data, uploads) if request.accept_mimetypes.best == "application/json": return { "artifacts_url": url_for("run_artifacts", run_id=run.id), "detail_url": url_for("run_detail", run_id=run.id), "output_url": url_for("run_output", run_id=run.id), "run_id": run.id, "state_url": url_for("run_state", run_id=run.id), }, 202 return redirect(url_for("run_detail", run_id=run.id)) @app.post("/groups//run") def run_group(group_id: str): group = catalog.group(group_id) if group is None or not group.can_run_as_group: abort(404) run, _ = manager.run_group(group) return redirect(url_for("run_detail", run_id=run.id)) @app.get("/runs/") def run_detail(run_id: int): run = store.get(run_id) if run is None: abort(404) return render_template( "run.html", run=run, children=store.children(run.id), output=read_run_output(run), ) @app.post("/runs//cancel") def cancel_run(run_id: int): run = manager.request_cancel(run_id) if run is None: abort(404) return redirect(url_for("run_detail", run_id=run.id)) @app.get("/runs//output") def run_output(run_id: int): run = store.get(run_id) if run is None: abort(404) return {"output": read_run_output(run)} def read_run_output(run) -> str: if not run.log_path: return "" try: return Path(run.log_path).read_text(encoding="utf-8", errors="replace") except FileNotFoundError: return "" @app.get("/runs//artifacts") def run_artifacts(run_id: int): run = store.get(run_id) if run is None: abort(404) return { "artifacts": [ { "name": name, "url": url_for("download_artifact", run_id=run.id, filename=name), } for name in manager.artifacts(run.id) ] } @app.get("/runs//artifacts/") def download_artifact(run_id: int, filename: str): run = store.get(run_id) if run is None or run.status != "succeeded": abort(404) artifact = manager.artifact_path(run.id, filename) if artifact is None: abort(404) response = send_file( artifact, as_attachment=True, download_name=artifact.name, conditional=False, ) original_response = response.response def delete_after_transfer(): try: yield from original_response finally: close = getattr(original_response, "close", None) if close is not None: close() manager.delete_artifact(run.id, artifact.name) response.response = delete_after_transfer() response.direct_passthrough = False return response @app.get("/runs//state") def run_state(run_id: int): run = store.get(run_id) if run is None: abort(404) return { "status": run.status, "finished_at": run.finished_at, "exit_code": run.exit_code, "cancel_requested_at": run.cancel_requested_at, } @app.get("/history//") def history(target_kind: str, target_path: str): if target_kind not in {"task", "group"}: abort(404) return render_template( "history.html", target_kind=target_kind, target_path=target_path, runs=store.history(target_kind, target_path), ) return app