From 6d28257b54dabbcf450308bf87e3ea1bd5da9748 Mon Sep 17 00:00:00 2001 From: ajp_anton Date: Sun, 9 Aug 2026 02:31:54 +0000 Subject: [PATCH] Update Python tools --- python-tools/app/discovery.py | 65 ++++++++++++++- python-tools/app/inputs.py | 50 ++++++++++++ python-tools/app/runner.py | 127 +++++++++++++++++++++++++++++- python-tools/app/scheduler.py | 4 + python-tools/app/store.py | 15 ++++ python-tools/app/web.py | 54 ++++++++++++- python-tools/requirements.txt | 2 + python-tools/static/app.js | 45 +++++++++++ python-tools/templates/index.html | 8 +- python-tools/templates/run.html | 4 + 10 files changed, 362 insertions(+), 12 deletions(-) diff --git a/python-tools/app/discovery.py b/python-tools/app/discovery.py index aff2590..c4e3d14 100644 --- a/python-tools/app/discovery.py +++ b/python-tools/app/discovery.py @@ -11,7 +11,7 @@ from typing import Any FIELD_NAME = re.compile(r"[a-z][a-z0-9_]*\Z") -FIELD_TYPES = frozenset({"text", "integer", "date", "datetime", "choice", "multi_choice"}) +FIELD_TYPES = frozenset({"text", "integer", "date", "datetime", "choice", "multi_choice", "file"}) def display_name(value: str) -> str: @@ -28,6 +28,7 @@ class Task: description: str | None inputs: tuple["InputField", ...] wait_for_result: bool = False + download_artifacts: bool = False @dataclass(frozen=True) @@ -58,6 +59,8 @@ class InputField: step: int | None pattern: str | None options: tuple[InputOption, ...] + accept: tuple[str, ...] + maximum_bytes: int | None @dataclass(frozen=True) @@ -154,6 +157,7 @@ class TaskCatalog: description=read_description(path), inputs=declaration.inputs if declaration else (), wait_for_result=declaration.wait_for_result if declaration else False, + download_artifacts=declaration.download_artifacts if declaration else False, ) ) return tasks @@ -275,6 +279,7 @@ def load_input_manifest( class TaskDeclaration: inputs: tuple[InputField, ...] wait_for_result: bool + download_artifacts: bool def parse_task_declaration( @@ -282,7 +287,11 @@ def parse_task_declaration( filename: str, declaration: Any, ) -> TaskDeclaration: - if not isinstance(declaration, dict) or not {"inputs"} <= set(declaration) or set(declaration) - {"inputs", "wait_for_result"}: + if ( + not isinstance(declaration, dict) + or not {"inputs"} <= set(declaration) + or set(declaration) - {"inputs", "wait_for_result", "download_artifacts"} + ): raise ValueError(f"{manifest_path} task {filename} has an invalid declaration.") raw_inputs = declaration["inputs"] if not isinstance(raw_inputs, list): @@ -294,7 +303,16 @@ def parse_task_declaration( wait_for_result = declaration.get("wait_for_result", False) if not isinstance(wait_for_result, bool): raise ValueError(f"{manifest_path} task {filename} wait_for_result must be true or false.") - return TaskDeclaration(inputs=fields, wait_for_result=wait_for_result) + download_artifacts = declaration.get("download_artifacts", False) + if not isinstance(download_artifacts, bool): + raise ValueError(f"{manifest_path} task {filename} download_artifacts must be true or false.") + if download_artifacts and not wait_for_result: + raise ValueError(f"{manifest_path} task {filename} download_artifacts requires wait_for_result.") + return TaskDeclaration( + inputs=fields, + wait_for_result=wait_for_result, + download_artifacts=download_artifacts, + ) def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputField: @@ -302,7 +320,7 @@ def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputFiel raise ValueError(f"{manifest_path} task {filename} has a non-object input field.") allowed = { "name", "label", "type", "required", "default", "minimum", "maximum", - "step", "pattern", "options", + "step", "pattern", "options", "accept", "maximum_bytes", } unknown = set(raw) - allowed if unknown: @@ -344,6 +362,20 @@ def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputFiel except re.error as error: raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid pattern: {error}.") from error + accept = parse_file_accept(manifest_path, filename, name, field_type, raw.get("accept")) + maximum_bytes = raw.get("maximum_bytes") + if field_type == "file": + if pattern is not None or raw.get("default") is not None: + raise ValueError(f"{manifest_path} task {filename} file input {name} cannot define a pattern or default.") + if maximum_bytes is not None and ( + not isinstance(maximum_bytes, int) + or isinstance(maximum_bytes, bool) + or maximum_bytes < 1 + ): + raise ValueError(f"{manifest_path} task {filename} file input {name} maximum_bytes must be positive.") + elif maximum_bytes is not None: + raise ValueError(f"{manifest_path} task {filename} input {name} only file fields accept maximum_bytes.") + options = parse_options(manifest_path, filename, name, field_type, raw.get("options")) default = parse_default(manifest_path, filename, name, field_type, options, raw.get("default")) validate_default( @@ -368,9 +400,34 @@ def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputFiel step=numeric_values["step"], pattern=pattern, options=options, + accept=accept, + maximum_bytes=maximum_bytes, ) +def parse_file_accept( + manifest_path: Path, + filename: str, + name: str, + field_type: str, + raw_accept: Any, +) -> tuple[str, ...]: + if field_type != "file": + if raw_accept is not None: + raise ValueError(f"{manifest_path} task {filename} input {name} only file fields accept accept.") + return () + if raw_accept is None: + return () + if ( + not isinstance(raw_accept, list) + or not raw_accept + or not all(isinstance(value, str) and re.fullmatch(r"\.[a-z0-9]{1,10}", value) for value in raw_accept) + or len(set(raw_accept)) != len(raw_accept) + ): + raise ValueError(f"{manifest_path} task {filename} file input {name} has invalid accept values.") + return tuple(raw_accept) + + def parse_options( manifest_path: Path, filename: str, diff --git a/python-tools/app/inputs.py b/python-tools/app/inputs.py index c34466c..eebd41b 100644 --- a/python-tools/app/inputs.py +++ b/python-tools/app/inputs.py @@ -4,11 +4,28 @@ from __future__ import annotations import re from collections.abc import Mapping, Sequence +from dataclasses import dataclass from datetime import date, datetime +from pathlib import Path +from typing import BinaryIO, Protocol from .discovery import InputField +class Upload(Protocol): + filename: str | None + content_length: int | None + stream: BinaryIO + + +@dataclass(frozen=True) +class PendingUpload: + field_name: str + extension: str + maximum_bytes: int | None + stream: BinaryIO + + def validate_inputs( fields: Sequence[InputField], values: Mapping[str, list[str]], @@ -16,6 +33,8 @@ def validate_inputs( result: dict[str, str | int | list[str]] = {} errors: dict[str, str] = {} for field in fields: + if field.type == "file": + continue raw_values = values.get(field.name, []) value, error = validate_field(field, raw_values) if error: @@ -25,6 +44,37 @@ def validate_inputs( return result, errors +def validate_uploads( + fields: Sequence[InputField], + uploads: Mapping[str, Upload], +) -> tuple[dict[str, PendingUpload], dict[str, str]]: + result: dict[str, PendingUpload] = {} + errors: dict[str, str] = {} + for field in fields: + if field.type != "file": + continue + upload = uploads.get(field.name) + filename = upload.filename if upload is not None else None + if upload is None or not filename: + if field.required: + errors[field.name] = "Choose a file." + continue + extension = Path(filename).suffix.lower() + if field.accept and extension not in field.accept: + errors[field.name] = f"Choose a file with one of: {', '.join(field.accept)}." + continue + if upload.content_length is not None and field.maximum_bytes is not None and upload.content_length > field.maximum_bytes: + errors[field.name] = f"Choose a file no larger than {field.maximum_bytes} bytes." + continue + result[field.name] = PendingUpload( + field_name=field.name, + extension=extension, + maximum_bytes=field.maximum_bytes, + stream=upload.stream, + ) + return result, errors + + def validate_field( field: InputField, raw_values: list[str], diff --git a/python-tools/app/runner.py b/python-tools/app/runner.py index e3501ed..a2d60bb 100644 --- a/python-tools/app/runner.py +++ b/python-tools/app/runner.py @@ -2,15 +2,19 @@ from __future__ import annotations -import os import json +import os import signal +import shutil import subprocess import threading +import time +from collections.abc import Mapping from pathlib import Path from .config import Settings from .discovery import BackgroundTask, Group, Task, TaskCatalog +from .inputs import PendingUpload from .store import Run, RunStore @@ -28,6 +32,7 @@ class RunManager: self, task: Task, input_data: dict[str, str | int | list[str]] | None = None, + uploads: Mapping[str, PendingUpload] | None = None, ) -> tuple[Run, bool]: run, created = self.store.enqueue( target_kind="task", @@ -36,6 +41,17 @@ class RunManager: input_data=input_data, ) if created: + try: + uploaded_inputs = self._save_uploads(run.id, uploads or {}) + except (OSError, ValueError) as error: + self._delete_uploads(run.id) + return self.store.finish( + run.id, + status="failed", + message=f"Could not save uploaded file: {error}", + ), True + if uploaded_inputs: + run = self.store.set_input(run.id, {**(input_data or {}), **uploaded_inputs}) self._wake_worker(task.group_id) return run, created @@ -58,7 +74,55 @@ class RunManager: if child is not None: self.request_cancel(child.id) self._terminate_active_process(run.id) - return self.store.get(run.id) + completed = self.store.get(run.id) + if completed is not None and completed.status == "cancelled": + self._delete_uploads(run.id) + return completed + + def artifacts(self, run_id: int) -> list[str]: + root = self._artifact_root(run_id) + if not root.is_dir(): + return [] + return [ + path.name + for path in sorted(root.iterdir(), key=lambda item: item.name) + if path.is_file() and not path.is_symlink() + ] + + def artifact_path(self, run_id: int, filename: str) -> Path | None: + path = self._artifact_root(run_id) / filename + if Path(filename).name != filename or path.is_symlink() or not path.is_file(): + return None + return path + + def delete_artifact(self, run_id: int, filename: str) -> None: + path = self.artifact_path(run_id, filename) + if path is None: + return + path.unlink(missing_ok=True) + try: + path.parent.rmdir() + except OSError: + pass + + def cleanup_transient_files(self, *, artifact_max_age_seconds: int = 86_400) -> None: + """Remove uploads from interrupted runs and expired undownloaded artifacts.""" + + runs_root = self.settings.state_root / "runs" + if not runs_root.is_dir(): + return + cutoff = time.time() - artifact_max_age_seconds + for path in runs_root.iterdir(): + if not path.is_dir(): + continue + run_id_text, separator, kind = path.name.partition(".") + if not separator or not run_id_text.isdigit() or kind not in {"uploads", "artifacts"}: + continue + run = self.store.get(int(run_id_text)) + if kind == "uploads" and (run is None or run.status not in {"queued", "running"}): + shutil.rmtree(path, ignore_errors=True) + elif kind == "artifacts" and path.stat().st_mtime < cutoff: + shutil.rmtree(path, ignore_errors=True) def run_background_task(self, task: BackgroundTask) -> None: """Run a declared internal task without creating high-volume run history.""" @@ -171,10 +235,14 @@ class RunManager: log_path = self.settings.state_root / "runs" / f"{run.id}.log" log_path.parent.mkdir(parents=True, exist_ok=True) input_path = self.settings.state_root / "runs" / f"{run.id}.input.json" + artifact_root = self._artifact_root(run.id) + artifact_root.mkdir(parents=True, exist_ok=True) try: input_path.write_text(run.input_json + "\n", encoding="utf-8") input_path.chmod(0o600) except OSError as error: + self._delete_uploads(run.id) + shutil.rmtree(artifact_root, ignore_errors=True) return self.store.finish( run.id, status="failed", @@ -182,6 +250,7 @@ class RunManager: ) environment = self._environment(task.group_id) environment["SERVER_MAINTENANCE_INPUT"] = str(input_path) + environment["SERVER_MAINTENANCE_ARTIFACTS"] = str(artifact_root) try: with log_path.open("wb") as log_file: @@ -198,6 +267,8 @@ class RunManager: self._processes[run.id] = process exit_code = process.wait() except OSError as error: + self._delete_uploads(run.id) + shutil.rmtree(artifact_root, ignore_errors=True) return self.store.finish( run.id, status="failed", @@ -207,22 +278,33 @@ class RunManager: finally: with self._lock: self._processes.pop(run.id, None) + self._delete_uploads(run.id) if self.store.is_cancel_requested(run.id) or ( parent_run_id is not None and self.store.is_cancel_requested(parent_run_id) ): + shutil.rmtree(artifact_root, ignore_errors=True) return self.store.finish( run.id, status="cancelled", exit_code=exit_code, log_path=str(log_path), ) - return self.store.finish( + completed = self.store.finish( run.id, status="succeeded" if exit_code == 0 else "failed", exit_code=exit_code, log_path=str(log_path), ) + if completed.status != "succeeded": + shutil.rmtree(artifact_root, ignore_errors=True) + else: + try: + artifact_root.rmdir() + except OSError: + pass + self.cleanup_transient_files() + return completed def _environment(self, group_id: str) -> dict[str, str]: task_state = self.settings.state_root / "tasks" / group_id @@ -237,6 +319,45 @@ class RunManager: environment["TZ"] = self.settings.timezone_name return environment + def _upload_root(self, run_id: int) -> Path: + return self.settings.state_root / "runs" / f"{run_id}.uploads" + + def _artifact_root(self, run_id: int) -> Path: + return self.settings.state_root / "runs" / f"{run_id}.artifacts" + + def _save_uploads( + self, + run_id: int, + uploads: Mapping[str, PendingUpload], + ) -> dict[str, str]: + if not uploads: + return {} + root = self._upload_root(run_id) + root.mkdir(parents=True, exist_ok=False) + saved = {} + try: + for field_name, upload in uploads.items(): + suffix = upload.extension or ".upload" + destination = root / f"{field_name}{suffix}" + temporary = destination.with_suffix(destination.suffix + ".part") + total = 0 + with temporary.open("xb") as output: + while chunk := upload.stream.read(64 * 1024): + total += len(chunk) + if upload.maximum_bytes is not None and total > upload.maximum_bytes: + raise ValueError(f"{field_name} exceeds its size limit") + output.write(chunk) + temporary.replace(destination) + destination.chmod(0o600) + saved[field_name] = str(destination) + except Exception: + shutil.rmtree(root, ignore_errors=True) + raise + return saved + + def _delete_uploads(self, run_id: int) -> None: + shutil.rmtree(self._upload_root(run_id), ignore_errors=True) + def _terminate_active_process(self, run_id: int) -> None: with self._lock: process = self._processes.get(run_id) diff --git a/python-tools/app/scheduler.py b/python-tools/app/scheduler.py index 0a8a4fd..acbd22c 100644 --- a/python-tools/app/scheduler.py +++ b/python-tools/app/scheduler.py @@ -19,6 +19,7 @@ class BackgroundScheduler: self._stopped = threading.Event() self._lock = threading.Lock() self._next_due: dict[str, float] = {} + self._next_cleanup = 0.0 self._running: set[str] = set() self._thread: threading.Thread | None = None @@ -40,6 +41,9 @@ class BackgroundScheduler: def _loop(self) -> None: while not self._stopped.is_set(): current = time.monotonic() + if current >= self._next_cleanup: + self._next_cleanup = current + 60 + self.manager.cleanup_transient_files() for task in self.tasks: due = self._next_due.setdefault(task.id, current) if current >= due: diff --git a/python-tools/app/store.py b/python-tools/app/store.py index 5730aba..2bfc392 100644 --- a/python-tools/app/store.py +++ b/python-tools/app/store.py @@ -133,6 +133,21 @@ class RunStore: ) return self.get(cursor.lastrowid, connection=connection), True + def set_input( + self, + run_id: int, + input_data: Mapping[str, str | int | list[str]], + ) -> Run: + input_json = json.dumps(input_data, sort_keys=True, separators=(",", ":")) + with self._connect() as connection: + connection.execute( + "UPDATE runs SET input_json = ? WHERE id = ?", + (input_json, run_id), + ) + run = self.get(run_id, connection=connection) + assert run is not None + return run + def claim_next(self, queue_group: str) -> Run | None: with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") diff --git a/python-tools/app/web.py b/python-tools/app/web.py index e8b94d3..0a4257d 100644 --- a/python-tools/app/web.py +++ b/python-tools/app/web.py @@ -6,11 +6,11 @@ import ipaddress from datetime import datetime from pathlib import Path -from flask import Flask, abort, redirect, render_template, request, url_for +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 +from .inputs import validate_inputs, validate_uploads from .runner import RunManager from .scheduler import BackgroundScheduler from .store import RunStore @@ -23,6 +23,7 @@ def create_app(settings: Settings | None = None) -> Flask: store.initialize() store.mark_interrupted() manager = RunManager(settings, catalog, store) + manager.cleanup_transient_files() scheduler = BackgroundScheduler(catalog, manager) scheduler.start() @@ -95,6 +96,8 @@ def create_app(settings: Settings | None = None) -> Flask: 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 @@ -103,9 +106,10 @@ def create_app(settings: Settings | None = None) -> Flask: input_errors={task.id: errors}, status=400, ) - run, _ = manager.run_task(task, input_data) + 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, @@ -151,6 +155,50 @@ def create_app(settings: Settings | None = None) -> Flask: except FileNotFoundError: return {"output": ""} + @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) diff --git a/python-tools/requirements.txt b/python-tools/requirements.txt index f5f8f0f..d719401 100644 --- a/python-tools/requirements.txt +++ b/python-tools/requirements.txt @@ -1,2 +1,4 @@ Flask==3.1.0 waitress==3.0.2 +numpy==2.5.1 +psycopg[binary]==3.3.4 diff --git a/python-tools/static/app.js b/python-tools/static/app.js index fa6e3f8..512b2e1 100644 --- a/python-tools/static/app.js +++ b/python-tools/static/app.js @@ -18,6 +18,31 @@ if (output) { window.setInterval(refresh, 2000); } +const refreshArtifacts = async (container) => { + const response = await fetch(container.dataset.artifactsUrl, { cache: "no-store" }); + if (!response.ok) { + return []; + } + const artifacts = (await response.json()).artifacts; + const list = container.querySelector("ul"); + list.replaceChildren(); + for (const artifact of artifacts) { + const item = document.createElement("li"); + const link = document.createElement("a"); + link.href = artifact.url; + link.textContent = artifact.name; + item.append(link); + list.append(item); + } + container.hidden = artifacts.length === 0; + return artifacts; +}; + +for (const container of document.querySelectorAll("#run-artifacts")) { + refreshArtifacts(container); + window.setInterval(() => refreshArtifacts(container), 2000); +} + for (const form of document.querySelectorAll("form[data-wait-for-result]")) { const result = form.querySelector(".task-result"); const controls = [...form.querySelectorAll("input, select, button")]; @@ -53,6 +78,23 @@ for (const form of document.querySelectorAll("form[data-wait-for-result]")) { } }; + const downloadArtifacts = async (run) => { + const response = await fetch(run.artifacts_url, { cache: "no-store" }); + if (!response.ok) { + return; + } + const artifacts = (await response.json()).artifacts; + if (!artifacts.length) { + return; + } + const link = document.createElement("a"); + link.href = artifacts[0].url; + link.download = ""; + document.body.append(link); + link.click(); + link.remove(); + }; + const waitForRun = async (run) => { while (true) { const response = await fetch(run.state_url, { cache: "no-store" }); @@ -63,6 +105,9 @@ for (const form of document.querySelectorAll("form[data-wait-for-result]")) { if (state.status !== "queued" && state.status !== "running") { if (state.status === "succeeded") { showResult("Completed successfully.", "succeeded", run.detail_url); + if (form.dataset.downloadArtifacts !== undefined) { + await downloadArtifacts(run); + } } else { showResult(`Finished with status: ${state.status}.`, "failed", run.detail_url); } diff --git a/python-tools/templates/index.html b/python-tools/templates/index.html index 6c10fc7..765504e 100644 --- a/python-tools/templates/index.html +++ b/python-tools/templates/index.html @@ -37,12 +37,16 @@

{% endif %} -
+ {% set values = form_values.get(task.id, {}) %} {% set errors = input_errors.get(task.id, {}) %} {% for field in task.inputs %}
- {% if field.type == "multi_choice" %} + {% if field.type == "file" %} + + {% elif field.type == "multi_choice" %}
{{ field.label }}{% if field.required %} (required){% endif %} {% set selected = values.get(field.name, field.default or ()) %} diff --git a/python-tools/templates/run.html b/python-tools/templates/run.html index a657aa4..d2a8618 100644 --- a/python-tools/templates/run.html +++ b/python-tools/templates/run.html @@ -24,6 +24,10 @@ {% endfor %} {% endif %} +

Output


 {% endblock %}