From 56d619432497c90c711e32fb66acb39a64908e84 Mon Sep 17 00:00:00 2001 From: ajp_anton Date: Sat, 8 Aug 2026 17:21:50 +0000 Subject: [PATCH] Integrate task runner into python tools --- .gitea/workflows/build-images.yaml | 6 +- README.md | 21 ++- python-tools/.dockerignore | 6 + python-tools/Dockerfile | 18 +- python-tools/app/__init__.py | 1 + python-tools/app/__main__.py | 17 ++ python-tools/app/config.py | 58 ++++++ python-tools/app/discovery.py | 107 +++++++++++ python-tools/app/runner.py | 219 ++++++++++++++++++++++ python-tools/app/store.py | 271 ++++++++++++++++++++++++++++ python-tools/app/web.py | 136 ++++++++++++++ python-tools/requirements-dev.txt | 2 + python-tools/requirements.txt | 2 + python-tools/static/app.js | 19 ++ python-tools/static/style.css | 83 +++++++++ python-tools/templates/base.html | 18 ++ python-tools/templates/history.html | 18 ++ python-tools/templates/index.html | 48 +++++ python-tools/templates/run.html | 29 +++ python-tools/tests/__init__.py | 0 python-tools/tests/conftest.py | 44 +++++ python-tools/tests/test_store.py | 21 +++ 22 files changed, 1137 insertions(+), 7 deletions(-) create mode 100644 python-tools/.dockerignore create mode 100644 python-tools/app/__init__.py create mode 100644 python-tools/app/__main__.py create mode 100644 python-tools/app/config.py create mode 100644 python-tools/app/discovery.py create mode 100644 python-tools/app/runner.py create mode 100644 python-tools/app/store.py create mode 100644 python-tools/app/web.py create mode 100644 python-tools/requirements-dev.txt create mode 100644 python-tools/requirements.txt create mode 100644 python-tools/static/app.js create mode 100644 python-tools/static/style.css create mode 100644 python-tools/templates/base.html create mode 100644 python-tools/templates/history.html create mode 100644 python-tools/templates/index.html create mode 100644 python-tools/templates/run.html create mode 100644 python-tools/tests/__init__.py create mode 100644 python-tools/tests/conftest.py create mode 100644 python-tools/tests/test_store.py diff --git a/.gitea/workflows/build-images.yaml b/.gitea/workflows/build-images.yaml index 25cbc91..02da27d 100644 --- a/.gitea/workflows/build-images.yaml +++ b/.gitea/workflows/build-images.yaml @@ -70,8 +70,8 @@ jobs: base_image: python:3 primary_tag: git.ajpanton.se/ajp_anton/python-tools:3 oci_labels: | - org.opencontainers.image.title=Python 3 tools with ExifTool - org.opencontainers.image.description=Python 3 utility image with ExifTool for local helper workflows. + org.opencontainers.image.title=Python 3 tools with ExifTool and task runner + org.opencontainers.image.description=Python 3 utility image with ExifTool and an optional web control panel for mounted task scripts. org.opencontainers.image.documentation=https://git.ajpanton.se/ajp_anton/docker-images#python-tools org.opencontainers.image.source=https://git.ajpanton.se/ajp_anton/docker-images org.opencontainers.image.url=https://git.ajpanton.se/ajp_anton/-/packages/container/python-tools @@ -81,7 +81,7 @@ jobs: build_args: | PYTHON_VERSION=3 fingerprint_command: | - dpkg-query -W -f='${binary:Package}=${Version}\n' | LC_ALL=C sort + { dpkg-query -W -f='${binary:Package}=${Version}\n' | LC_ALL=C sort; pip freeze | LC_ALL=C sort; } steps: - name: Check out repository diff --git a/README.md b/README.md index c289f2b..2d6c352 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ a few commonly useful database/runtime extensions. | --- | --- | | `git.ajpanton.se/ajp_anton/postgres-custom:18` | PostgreSQL 18 with selected extension packages installed and `pg_cron`/VectorChord preloaded. | | `git.ajpanton.se/ajp_anton/php8-pgsql:8` | PHP 8 FPM on Alpine with PostgreSQL client runtime support and the `pdo_pgsql` PHP extension enabled. | -| `git.ajpanton.se/ajp_anton/python-tools:3` | Python 3 utility image with ExifTool for AI/helper workflows. | +| `git.ajpanton.se/ajp_anton/python-tools:3` | Python 3 task runner with ExifTool and a web control panel for mounted task scripts. | ## Included Components @@ -78,10 +78,25 @@ Installed from Debian apt repositories: - `ca-certificates` - `exiftool` +Pinned Python packages: + +- `Flask==3.1.0` +- `waitress==3.0.2` + +Included application: + +- A server-rendered web control panel for discovering and running reviewed, + zero-input Python scripts mounted at runtime. +- Per-group task queues, SQLite run history, captured output, and task-process + cancellation. +- The default command starts the control panel with `python -m app`. + Intended use: -- Run small helper scripts mounted at runtime, for example media metadata - extraction scripts used by local AI tools. +- Run reviewed, autonomous task scripts mounted at runtime, including metadata + extraction scripts that need ExifTool. +- Use the included task control panel. Its generic deployment example and + application documentation are in `python-tools/`. - Keep host systems clean by avoiding host-level package installs. ## Tags diff --git a/python-tools/.dockerignore b/python-tools/.dockerignore new file mode 100644 index 0000000..ea507b4 --- /dev/null +++ b/python-tools/.dockerignore @@ -0,0 +1,6 @@ +.git +.pytest_cache +__pycache__ +*.py[cod] +tests +tasks diff --git a/python-tools/Dockerfile b/python-tools/Dockerfile index f7b7eb1..2e37886 100644 --- a/python-tools/Dockerfile +++ b/python-tools/Dockerfile @@ -1,6 +1,9 @@ ARG PYTHON_VERSION=3 FROM python:${PYTHON_VERSION} +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + RUN set -eux; \ apt-get update; \ DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \ @@ -8,6 +11,19 @@ RUN set -eux; \ exiftool; \ rm -rf /var/lib/apt/lists/* +WORKDIR /opt/server-maintenance + +COPY requirements.txt ./ +RUN pip install --no-cache-dir -r requirements.txt + +COPY app ./app +COPY static ./static +COPY templates ./templates + +ENV PYTHONPATH=/opt/server-maintenance + WORKDIR /work -CMD ["python3", "--version"] +EXPOSE 8080 + +CMD ["python", "-m", "app"] diff --git a/python-tools/app/__init__.py b/python-tools/app/__init__.py new file mode 100644 index 0000000..d7fd8ff --- /dev/null +++ b/python-tools/app/__init__.py @@ -0,0 +1 @@ +"""Server-maintenance control panel.""" diff --git a/python-tools/app/__main__.py b/python-tools/app/__main__.py new file mode 100644 index 0000000..59156a1 --- /dev/null +++ b/python-tools/app/__main__.py @@ -0,0 +1,17 @@ +"""Production entry point.""" + +from __future__ import annotations + +import os + +from waitress import serve + +from .web import create_app + + +if __name__ == "__main__": + serve( + create_app(), + host=os.environ.get("SERVER_MAINTENANCE_HOST", "0.0.0.0"), + port=int(os.environ.get("SERVER_MAINTENANCE_PORT", "8080")), + ) diff --git a/python-tools/app/config.py b/python-tools/app/config.py new file mode 100644 index 0000000..e9df242 --- /dev/null +++ b/python-tools/app/config.py @@ -0,0 +1,58 @@ +"""Runtime configuration.""" + +from __future__ import annotations + +import os +import sys +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class Settings: + task_root: Path + state_root: Path + credentials_root: Path + python_executable: str + timezone_name: str | None + allowed_proxy_ips: frozenset[str] + cancel_grace_seconds: float + + @property + def database_path(self) -> Path: + return self.state_root / "server-maintenance.sqlite3" + + @classmethod + def from_environment(cls) -> "Settings": + allowed_proxy_ips = frozenset( + value.strip() + for value in os.environ.get( + "SERVER_MAINTENANCE_ALLOWED_PROXY_IPS", "" + ).split(",") + if value.strip() + ) + return cls( + task_root=Path( + os.environ.get("SERVER_MAINTENANCE_TASK_ROOT", "/opt/tasks") + ), + state_root=Path( + os.environ.get( + "SERVER_MAINTENANCE_STATE_ROOT", + "/var/lib/server-maintenance", + ) + ), + credentials_root=Path( + os.environ.get( + "SERVER_MAINTENANCE_CREDENTIALS_ROOT", + "/opt/credentials", + ) + ), + python_executable=os.environ.get( + "SERVER_MAINTENANCE_PYTHON", sys.executable + ), + timezone_name=os.environ.get("TZ") or None, + allowed_proxy_ips=allowed_proxy_ips, + cancel_grace_seconds=float( + os.environ.get("SERVER_MAINTENANCE_CANCEL_GRACE_SECONDS", "10") + ), + ) diff --git a/python-tools/app/discovery.py b/python-tools/app/discovery.py new file mode 100644 index 0000000..b103e5d --- /dev/null +++ b/python-tools/app/discovery.py @@ -0,0 +1,107 @@ +"""Safe, non-executing task discovery.""" + +from __future__ import annotations + +import ast +from dataclasses import dataclass +from pathlib import Path + + +def display_name(value: str) -> str: + return value.replace("_", " ") + + +@dataclass(frozen=True) +class Task: + id: str + group_id: str + filename: str + path: Path + display_name: str + description: str | None + + +@dataclass(frozen=True) +class Group: + id: str + path: Path + display_name: str + tasks: tuple[Task, ...] + + +class TaskCatalog: + """Discovers only direct Python children of non-hidden group directories.""" + + def __init__(self, task_root: Path) -> None: + self.task_root = task_root + + def groups(self) -> tuple[Group, ...]: + if not self.task_root.is_dir(): + return () + + groups = [] + for group_path in sorted(self.task_root.iterdir(), key=lambda path: path.name): + if ( + group_path.name.startswith((".", "_")) + or group_path.is_symlink() + or not group_path.is_dir() + ): + continue + tasks = tuple(self._tasks_in_group(group_path)) + groups.append( + Group( + id=group_path.name, + path=group_path, + display_name=display_name(group_path.name), + tasks=tasks, + ) + ) + return tuple(groups) + + def group(self, group_id: str) -> Group | None: + return next((group for group in self.groups() if group.id == group_id), None) + + def task(self, task_id: str) -> Task | None: + for group in self.groups(): + for task in group.tasks: + if task.id == task_id: + return task + return None + + def _tasks_in_group(self, group_path: Path) -> list[Task]: + tasks = [] + for path in sorted(group_path.iterdir(), key=lambda candidate: candidate.name): + if ( + path.name.startswith((".", "_")) + or path.is_symlink() + or not path.is_file() + or path.suffix != ".py" + ): + continue + tasks.append( + Task( + id=f"{group_path.name}/{path.name}", + group_id=group_path.name, + filename=path.name, + path=path, + display_name=display_name(path.stem), + description=read_description(path), + ) + ) + return tasks + + +def read_description(path: Path) -> str | None: + """Return the first docstring line without importing or executing a task.""" + + try: + module = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + except (OSError, SyntaxError, UnicodeDecodeError): + return None + docstring = ast.get_docstring(module, clean=True) + if not docstring: + return None + return next( + (line.strip() for line in docstring.splitlines() if line.strip()), + None, + ) diff --git a/python-tools/app/runner.py b/python-tools/app/runner.py new file mode 100644 index 0000000..5f14a36 --- /dev/null +++ b/python-tools/app/runner.py @@ -0,0 +1,219 @@ +"""Per-group task queues and subprocess execution.""" + +from __future__ import annotations + +import os +import signal +import subprocess +import threading +from pathlib import Path + +from .config import Settings +from .discovery import Group, Task, TaskCatalog +from .store import Run, RunStore + + +class RunManager: + def __init__(self, settings: Settings, catalog: TaskCatalog, store: RunStore) -> None: + self.settings = settings + self.catalog = catalog + self.store = store + self._wake_events: dict[str, threading.Event] = {} + self._workers: dict[str, threading.Thread] = {} + self._processes: dict[int, subprocess.Popen[bytes]] = {} + self._lock = threading.Lock() + + def run_task(self, task: Task) -> tuple[Run, bool]: + run, created = self.store.enqueue( + target_kind="task", + target_path=task.id, + queue_group=task.group_id, + ) + if created: + self._wake_worker(task.group_id) + return run, created + + def run_group(self, group: Group) -> tuple[Run, bool]: + run, created = self.store.enqueue( + target_kind="group", + target_path=group.id, + queue_group=group.id, + ) + if created: + self._wake_worker(group.id) + return run, created + + def request_cancel(self, run_id: int) -> Run | None: + run = self.store.request_cancel(run_id) + if run is None: + return None + if run.parent_run_id is None and run.target_kind == "group": + child = self.store.active_child(run.id) + if child is not None: + self.request_cancel(child.id) + self._terminate_active_process(run.id) + return self.store.get(run.id) + + def _wake_worker(self, group_id: str) -> None: + with self._lock: + event = self._wake_events.setdefault(group_id, threading.Event()) + worker = self._workers.get(group_id) + if worker is None or not worker.is_alive(): + worker = threading.Thread( + target=self._worker_loop, + args=(group_id, event), + name=f"task-group-{group_id}", + daemon=True, + ) + self._workers[group_id] = worker + worker.start() + event.set() + + def _worker_loop(self, group_id: str, event: threading.Event) -> None: + while True: + event.clear() + run = self.store.claim_next(group_id) + if run is None: + event.wait() + continue + self._execute_root_run(run) + + def _execute_root_run(self, run: Run) -> None: + if run.target_kind == "task": + task = self.catalog.task(run.target_path) + if task is None: + self.store.finish( + run.id, + status="failed", + message="Task is no longer available.", + ) + return + self._execute_task(run, task) + return + + group = self.catalog.group(run.target_path) + if group is None: + self.store.finish( + run.id, + status="failed", + message="Group is no longer available.", + ) + return + self._execute_group(run, group) + + def _execute_group(self, parent_run: Run, group: Group) -> None: + for task in group.tasks: + if self.store.is_cancel_requested(parent_run.id): + self.store.finish( + parent_run.id, + status="cancelled", + message="Group cancellation requested.", + ) + return + child = self.store.create_child(parent_run, task.id) + completed_child = self._execute_task(child, task, parent_run_id=parent_run.id) + if completed_child.status != "succeeded": + parent_status = ( + "cancelled" + if completed_child.status == "cancelled" + or self.store.is_cancel_requested(parent_run.id) + else "failed" + ) + self.store.finish( + parent_run.id, + status=parent_status, + message=f"Stopped after {task.id}: {completed_child.status}.", + ) + return + self.store.finish(parent_run.id, status="succeeded") + + def _execute_task( + self, + run: Run, + task: Task, + *, + parent_run_id: int | None = None, + ) -> Run: + if self.store.is_cancel_requested(run.id) or ( + parent_run_id is not None and self.store.is_cancel_requested(parent_run_id) + ): + return self.store.finish(run.id, status="cancelled", message="Cancelled before execution.") + + task_state = self.settings.state_root / "tasks" / task.group_id + task_state.mkdir(parents=True, exist_ok=True) + log_path = self.settings.state_root / "runs" / f"{run.id}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + environment = { + "PATH": "/usr/local/bin:/usr/bin:/bin", + "PYTHONUNBUFFERED": "1", + "SERVER_MAINTENANCE_CREDENTIALS": str(self.settings.credentials_root), + "SERVER_MAINTENANCE_STATE": str(task_state), + } + if self.settings.timezone_name: + environment["TZ"] = self.settings.timezone_name + + try: + with log_path.open("wb") as log_file: + process = subprocess.Popen( + [self.settings.python_executable, str(task.path)], + cwd=task.path.parent, + stdin=subprocess.DEVNULL, + stdout=log_file, + stderr=subprocess.STDOUT, + env=environment, + start_new_session=True, + ) + with self._lock: + self._processes[run.id] = process + exit_code = process.wait() + except OSError as error: + return self.store.finish( + run.id, + status="failed", + message=f"Could not start task: {error}", + log_path=str(log_path), + ) + finally: + with self._lock: + self._processes.pop(run.id, None) + + if self.store.is_cancel_requested(run.id) or ( + parent_run_id is not None and self.store.is_cancel_requested(parent_run_id) + ): + return self.store.finish( + run.id, + status="cancelled", + exit_code=exit_code, + log_path=str(log_path), + ) + return self.store.finish( + run.id, + status="succeeded" if exit_code == 0 else "failed", + exit_code=exit_code, + log_path=str(log_path), + ) + + def _terminate_active_process(self, run_id: int) -> None: + with self._lock: + process = self._processes.get(run_id) + if process is None or process.poll() is not None: + return + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + return + threading.Thread( + target=self._kill_after_grace, + args=(process,), + name=f"task-cancel-{run_id}", + daemon=True, + ).start() + + def _kill_after_grace(self, process: subprocess.Popen[bytes]) -> None: + try: + process.wait(timeout=self.settings.cancel_grace_seconds) + except subprocess.TimeoutExpired: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass diff --git a/python-tools/app/store.py b/python-tools/app/store.py new file mode 100644 index 0000000..3fed05b --- /dev/null +++ b/python-tools/app/store.py @@ -0,0 +1,271 @@ +"""SQLite persistence for runs.""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path + + +ACTIVE_STATUSES = ("queued", "running") + + +@dataclass(frozen=True) +class Run: + id: int + target_kind: str + target_path: str + queue_group: str + trigger: str + status: str + created_at: str + started_at: str | None + finished_at: str | None + exit_code: int | None + cancel_requested_at: str | None + parent_run_id: int | None + log_path: str | None + message: str | None + + +def now() -> str: + return datetime.now(UTC).isoformat() + + +class RunStore: + def __init__(self, database_path: Path) -> None: + self.database_path = database_path + + def initialize(self) -> None: + self.database_path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as connection: + connection.execute( + """ + CREATE TABLE IF NOT EXISTS runs ( + id INTEGER PRIMARY KEY, + target_kind TEXT NOT NULL CHECK(target_kind IN ('task', 'group')), + target_path TEXT NOT NULL, + queue_group TEXT NOT NULL, + trigger TEXT NOT NULL CHECK(trigger IN ('manual', 'schedule')), + status TEXT NOT NULL, + created_at TEXT NOT NULL, + started_at TEXT, + finished_at TEXT, + exit_code INTEGER, + cancel_requested_at TEXT, + parent_run_id INTEGER REFERENCES runs(id), + log_path TEXT, + message TEXT + ) + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS runs_queue_index + ON runs(queue_group, status, created_at) + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS runs_target_index + ON runs(target_kind, target_path, created_at DESC) + """ + ) + + def mark_interrupted(self) -> None: + with self._connect() as connection: + connection.execute( + """ + UPDATE runs + SET status = 'interrupted', + finished_at = ?, + message = 'Application restarted before this run completed.' + WHERE status IN ('queued', 'running') + """, + (now(),), + ) + + def enqueue( + self, + *, + target_kind: str, + target_path: str, + queue_group: str, + trigger: str = "manual", + ) -> tuple[Run, bool]: + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + """ + SELECT * FROM runs + WHERE target_kind = ? AND target_path = ? + AND parent_run_id IS NULL + AND status IN ('queued', 'running') + ORDER BY id DESC + LIMIT 1 + """, + (target_kind, target_path), + ).fetchone() + if existing is not None: + return self._run(existing), False + cursor = connection.execute( + """ + INSERT INTO runs ( + target_kind, target_path, queue_group, trigger, status, created_at + ) VALUES (?, ?, ?, ?, 'queued', ?) + """, + (target_kind, target_path, queue_group, trigger, now()), + ) + return self.get(cursor.lastrowid, connection=connection), True + + def claim_next(self, queue_group: str) -> Run | None: + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + """ + SELECT * FROM runs + WHERE queue_group = ? AND parent_run_id IS NULL AND status = 'queued' + ORDER BY created_at, id + LIMIT 1 + """, + (queue_group,), + ).fetchone() + if row is None: + return None + run_id = row["id"] + connection.execute( + "UPDATE runs SET status = 'running', started_at = ? WHERE id = ?", + (now(), run_id), + ) + return self.get(run_id, connection=connection) + + def create_child(self, parent_run: Run, task_path: str) -> Run: + with self._connect() as connection: + cursor = connection.execute( + """ + INSERT INTO runs ( + target_kind, target_path, queue_group, trigger, status, created_at, + started_at, parent_run_id + ) VALUES ('task', ?, ?, ?, 'running', ?, ?, ?) + """, + ( + task_path, + parent_run.queue_group, + parent_run.trigger, + now(), + now(), + parent_run.id, + ), + ) + return self.get(cursor.lastrowid, connection=connection) + + def finish( + self, + run_id: int, + *, + status: str, + exit_code: int | None = None, + message: str | None = None, + log_path: str | None = None, + ) -> Run: + with self._connect() as connection: + connection.execute( + """ + UPDATE runs + SET status = ?, exit_code = ?, finished_at = ?, message = ?, + log_path = COALESCE(?, log_path) + WHERE id = ? + """, + (status, exit_code, now(), message, log_path, run_id), + ) + return self.get(run_id, connection=connection) + + def request_cancel(self, run_id: int) -> Run | None: + with self._connect() as connection: + run = self.get(run_id, connection=connection) + if run is None or run.status not in ACTIVE_STATUSES: + return run + if run.status == "queued": + connection.execute( + """ + UPDATE runs + SET status = 'cancelled', cancel_requested_at = ?, finished_at = ?, + message = 'Cancelled before execution.' + WHERE id = ? + """, + (now(), now(), run_id), + ) + elif run.cancel_requested_at is None: + connection.execute( + "UPDATE runs SET cancel_requested_at = ? WHERE id = ?", + (now(), run_id), + ) + return self.get(run_id, connection=connection) + + def is_cancel_requested(self, run_id: int) -> bool: + run = self.get(run_id) + return run is not None and run.cancel_requested_at is not None + + def active_child(self, parent_run_id: int) -> Run | None: + with self._connect() as connection: + row = connection.execute( + """ + SELECT * FROM runs + WHERE parent_run_id = ? AND status = 'running' + ORDER BY id DESC + LIMIT 1 + """, + (parent_run_id,), + ).fetchone() + return self._run(row) if row else None + + def get(self, run_id: int, *, connection: sqlite3.Connection | None = None) -> Run | None: + if connection is not None: + row = connection.execute("SELECT * FROM runs WHERE id = ?", (run_id,)).fetchone() + return self._run(row) if row else None + with self._connect() as owned_connection: + return self.get(run_id, connection=owned_connection) + + def last_run(self, target_kind: str, target_path: str) -> Run | None: + with self._connect() as connection: + row = connection.execute( + """ + SELECT * FROM runs + WHERE target_kind = ? AND target_path = ? AND parent_run_id IS NULL + ORDER BY id DESC + LIMIT 1 + """, + (target_kind, target_path), + ).fetchone() + return self._run(row) if row else None + + def history(self, target_kind: str, target_path: str) -> list[Run]: + with self._connect() as connection: + rows = connection.execute( + """ + SELECT * FROM runs + WHERE target_kind = ? AND target_path = ? AND parent_run_id IS NULL + ORDER BY id DESC + LIMIT 50 + """, + (target_kind, target_path), + ).fetchall() + return [self._run(row) for row in rows] + + def children(self, parent_run_id: int) -> list[Run]: + with self._connect() as connection: + rows = connection.execute( + "SELECT * FROM runs WHERE parent_run_id = ? ORDER BY id", + (parent_run_id,), + ).fetchall() + return [self._run(row) for row in rows] + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect(self.database_path, timeout=30) + connection.row_factory = sqlite3.Row + return connection + + @staticmethod + def _run(row: sqlite3.Row) -> Run: + return Run(**dict(row)) diff --git a/python-tools/app/web.py b/python-tools/app/web.py new file mode 100644 index 0000000..396ad54 --- /dev/null +++ b/python-tools/app/web.py @@ -0,0 +1,136 @@ +"""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, url_for + +from .config import Settings +from .discovery import TaskCatalog +from .runner import RunManager +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) + + 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.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") + + @app.get("/") + def index(): + 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) + + @app.post("/tasks//run") + def run_task(task_id: str): + task = catalog.task(task_id) + if task is None: + abort(404) + run, _ = manager.run_task(task) + 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: + 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), + ) + + @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) + if not run.log_path: + return {"output": ""} + try: + return {"output": Path(run.log_path).read_text(encoding="utf-8", errors="replace")} + except FileNotFoundError: + return {"output": ""} + + @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 diff --git a/python-tools/requirements-dev.txt b/python-tools/requirements-dev.txt new file mode 100644 index 0000000..31c406a --- /dev/null +++ b/python-tools/requirements-dev.txt @@ -0,0 +1,2 @@ +-r requirements.txt +pytest==8.3.5 diff --git a/python-tools/requirements.txt b/python-tools/requirements.txt new file mode 100644 index 0000000..f5f8f0f --- /dev/null +++ b/python-tools/requirements.txt @@ -0,0 +1,2 @@ +Flask==3.1.0 +waitress==3.0.2 diff --git a/python-tools/static/app.js b/python-tools/static/app.js new file mode 100644 index 0000000..223a2b3 --- /dev/null +++ b/python-tools/static/app.js @@ -0,0 +1,19 @@ +const output = document.querySelector("#run-output"); + +if (output) { + const refresh = async () => { + const response = await fetch(output.dataset.outputUrl, { cache: "no-store" }); + if (response.ok) { + output.textContent = (await response.json()).output; + } + const status = document.querySelector("#run-status"); + if (status) { + const stateResponse = await fetch(status.dataset.stateUrl, { cache: "no-store" }); + if (stateResponse.ok) { + status.textContent = (await stateResponse.json()).status; + } + } + }; + refresh(); + window.setInterval(refresh, 2000); +} diff --git a/python-tools/static/style.css b/python-tools/static/style.css new file mode 100644 index 0000000..63f9af7 --- /dev/null +++ b/python-tools/static/style.css @@ -0,0 +1,83 @@ +:root { + color-scheme: light dark; + font-family: system-ui, sans-serif; +} + +body { + margin: 0; +} + +header, +main { + max-width: 72rem; + margin: auto; + padding: 1rem; +} + +header { + border-bottom: 1px solid color-mix(in srgb, currentColor 20%, transparent); + font-size: 1.2rem; + font-weight: 700; +} + +a { + color: inherit; +} + +.group { + border: 1px solid color-mix(in srgb, currentColor 20%, transparent); + border-radius: .5rem; + margin: 1rem 0; + padding: 1rem; +} + +.group-heading, +.tasks li { + align-items: center; + display: flex; + gap: 1rem; + justify-content: space-between; +} + +.tasks, +.history { + list-style: none; + padding: 0; +} + +.tasks li { + border-top: 1px solid color-mix(in srgb, currentColor 15%, transparent); + padding: .8rem 0; +} + +.tasks h3, +.tasks p { + margin: .2rem 0; +} + +.run-summary { + font-size: .9rem; + opacity: .75; +} + +button { + cursor: pointer; + font: inherit; + padding: .4rem .7rem; +} + +dt { + font-weight: 700; +} + +dd { + margin: 0 0 .5rem; +} + +pre { + background: color-mix(in srgb, currentColor 8%, transparent); + max-height: 60vh; + overflow: auto; + padding: 1rem; + white-space: pre-wrap; +} diff --git a/python-tools/templates/base.html b/python-tools/templates/base.html new file mode 100644 index 0000000..6655286 --- /dev/null +++ b/python-tools/templates/base.html @@ -0,0 +1,18 @@ + + + + + + {% block title %}Server Maintenance{% endblock %} + + + +
+ Server Maintenance +
+
+ {% block content %}{% endblock %} +
+ + + diff --git a/python-tools/templates/history.html b/python-tools/templates/history.html new file mode 100644 index 0000000..5704c88 --- /dev/null +++ b/python-tools/templates/history.html @@ -0,0 +1,18 @@ +{% extends "base.html" %} +{% block title %}History · Server Maintenance{% endblock %} +{% block content %} +

← Tasks

+

History: {{ target_path }}

+
    + {% for run in runs %} +
  • + Run #{{ run.id }} + · {{ run.status }} + · {{ run.created_at|timestamp }} + {% if run.exit_code is not none %}· exit {{ run.exit_code }}{% endif %} +
    • + {% else %} +
  • No runs yet.
    • + {% endfor %} +
+{% endblock %} diff --git a/python-tools/templates/index.html b/python-tools/templates/index.html new file mode 100644 index 0000000..5297593 --- /dev/null +++ b/python-tools/templates/index.html @@ -0,0 +1,48 @@ +{% extends "base.html" %} +{% block content %} +

Tasks

+ {% if not groups %} +

No task groups found.

+ {% endif %} + {% for group in groups %} + {% set group_run = latest[("group", group.id)] %} +
+
+

{{ group.display_name }}

+
+ +
+
+ {% if group_run %} +

+ Group: {{ group_run.status }} + · {{ group_run.finished_at|timestamp if group_run.finished_at else "in progress" }} +

+ {% endif %} +
    + {% for task in group.tasks %} + {% set task_run = latest[("task", task.id)] %} +
  • +
    +

    {{ task.display_name }}

    + {% if task.description %}

    {{ task.description }}

    {% endif %} + {% if task_run %} +

    + Last run: + {{ task_run.status }} + · {{ task_run.finished_at|timestamp if task_run.finished_at else "in progress" }} + · history +

    + {% endif %} +
    +
    + +
    +
    • + {% else %} +
  • No exposed tasks in this group.
    • + {% endfor %} +
+
+ {% endfor %} +{% endblock %} diff --git a/python-tools/templates/run.html b/python-tools/templates/run.html new file mode 100644 index 0000000..a657aa4 --- /dev/null +++ b/python-tools/templates/run.html @@ -0,0 +1,29 @@ +{% extends "base.html" %} +{% block title %}Run {{ run.id }} · Server Maintenance{% endblock %} +{% block content %} +

← Tasks

+

{{ run.target_kind|title }} run #{{ run.id }}

+
+
Target
{{ run.target_path }}
+
Status
{{ run.status }}
+
Started
{{ run.started_at|timestamp }}
+
Finished
{{ run.finished_at|timestamp }}
+
Exit code
{{ run.exit_code if run.exit_code is not none else "—" }}
+ {% if run.message %}
Message
{{ run.message }}
{% endif %} +
+ {% if run.status in ("queued", "running") %} +
+ +
+ {% endif %} + {% if children %} +

Group tasks

+ + {% endif %} +

Output

+ 

+{% endblock %}
diff --git a/python-tools/tests/__init__.py b/python-tools/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/python-tools/tests/conftest.py b/python-tools/tests/conftest.py
new file mode 100644
index 0000000..83144f8
--- /dev/null
+++ b/python-tools/tests/conftest.py
@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+from app.config import Settings
+from app.discovery import TaskCatalog
+from app.runner import RunManager
+from app.store import RunStore
+
+
+@pytest.fixture
+def settings(tmp_path: Path) -> Settings:
+    return Settings(
+        task_root=tmp_path / "tasks",
+        state_root=tmp_path / "state",
+        credentials_root=tmp_path / "credentials",
+        python_executable=sys.executable,
+        timezone_name="Etc/UTC",
+        allowed_proxy_ips=frozenset(),
+        cancel_grace_seconds=0.1,
+    )
+
+
+@pytest.fixture
+def manager(settings: Settings) -> RunManager:
+    settings.task_root.mkdir(parents=True)
+    store = RunStore(settings.database_path)
+    store.initialize()
+    return RunManager(settings, TaskCatalog(settings.task_root), store)
+
+
+def write_task(
+    task_root: Path,
+    group: str,
+    filename: str,
+    source: str,
+) -> Path:
+    path = task_root / group / filename
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(source, encoding="utf-8")
+    return path
diff --git a/python-tools/tests/test_store.py b/python-tools/tests/test_store.py
new file mode 100644
index 0000000..c1e94a4
--- /dev/null
+++ b/python-tools/tests/test_store.py
@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from app.store import RunStore
+
+
+def test_restart_marks_unfinished_runs_interrupted(settings) -> None:
+    store = RunStore(settings.database_path)
+    store.initialize()
+    run, created = store.enqueue(
+        target_kind="task",
+        target_path="checks/status.py",
+        queue_group="checks",
+    )
+    assert created is True
+
+    store.mark_interrupted()
+
+    interrupted = store.get(run.id)
+    assert interrupted is not None
+    assert interrupted.status == "interrupted"
+    assert interrupted.finished_at is not None