alex/manage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tasks): unify saved tasks on ssh_tasks services

Browse Source 
- Add shared task_runner.run_saved_task helper used by routers/tasks.py and
  widgets/sources.py SshTaskWidgetSource.
- Saved tasks now target ssh_tasks service instances via default_service_id;
  the legacy default_machine_id and saved_task_runs are removed.
- Actions page lists ssh_tasks services for default and run-time selection.
- Update types, API client, hooks, tests, docs, and changelog.

Backend tests: 222 passed. Frontend lint/build/test: clean (71 passed).
This commit is contained in:
Developer
2026-06-23 13:55:13 +00:00
parent d7ad933b2a
commit 50eb76a10d
14 changed files with 371 additions and 410 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/media_library_viewer_api/routers/tasks.py
+38 -117
View File
@@ -3,18 +3,15 @@
from __future__ import annotations
import logging
import shlex
import time
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel, Field
from media_library_viewer_api.clients.local import LocalCommandClient
from media_library_viewer_api.clients.ssh import RemoteSSHClient
from media_library_viewer_api.config import get_settings
from media_library_viewer_api.dependencies import get_settings_store
from media_library_viewer_api.services.settings_store import SettingsStore
from media_library_viewer_api.services.task_runner import run_saved_task
from media_library_viewer_api.widgets.sources import ServiceRecord, build_service_record
logger = logging.getLogger(__name__)
@@ -27,7 +24,7 @@ class TaskInput(BaseModel):
task_type: str = Field(default="shell", description="shell or python")
content: str = Field(default="")
enabled: bool = True
default_machine_id: str = ""
default_service_id: str = ""
notes: str = ""
@@ -35,61 +32,31 @@ class RunTaskRequest(BaseModel):
task_id: str
def _machine_label(machine: dict[str, Any] | None) -> str:
if not machine:
def _service_label(service: dict[str, Any] | None) -> str:
if not service:
return ""
return str(machine.get("name") or machine.get("host") or machine.get("id") or "")
return str(service.get("name") or service.get("id") or "")
def _resolve_machine_for_task(
def _resolve_service_for_task(
store: SettingsStore,
task: dict[str, Any],
machine_id: str | None,
service_id: str | None,
) -> dict[str, Any] | None:
if machine_id:
return store.get_machine_config(machine_id) or store.get_machine(machine_id)
default_machine_id = str(task.get("default_machine_id") or "").strip()
if default_machine_id:
return store.get_machine_config(default_machine_id) or store.get_machine(default_machine_id)
machines = [machine for machine in store.list_machines() if machine.get("enabled")]
return machines[0] if machines else None
if service_id:
return store.get_service(service_id)
default_service_id = str(task.get("default_service_id") or "").strip()
if default_service_id:
return store.get_service(default_service_id)
services = [svc for svc in store.list_services("ssh_tasks") if svc.get("enabled")]
return services[0] if services else None
def _client_for_machine(store: SettingsStore, machine: dict[str, Any]):
mode = str(machine.get("mode") or "local").lower()
if mode == "local":
return LocalCommandClient()
def _service_row_to_record(service_row: dict[str, Any]) -> ServiceRecord:
"""Build a ServiceRecord from a raw settings_store service row."""
from media_library_viewer_api.services.settings_store import get_settings_store
host = str(machine.get("host") or "").strip()
username = str(machine.get("username") or "").strip()
if not host or not username:
raise HTTPException(status_code=400, detail="SSH machine is missing host or username")
settings = get_settings()
private_key = str(machine.get("ssh_private_key") or "")
passphrase = str(machine.get("ssh_private_key_passphrase") or "")
ssh_key_id = str(machine.get("ssh_key_id") or "").strip()
if ssh_key_id:
ssh_key = store.get_ssh_key(ssh_key_id)
if ssh_key:
private_key = str(ssh_key.get("private_key") or private_key)
passphrase = str(ssh_key.get("passphrase") or passphrase)
key_filename = ""
if machine.get("key_directory") and machine.get("key_name"):
key_filename = f"{machine.get('key_directory')}/{machine.get('key_name')}"
return RemoteSSHClient(
host=host,
username=username,
port=int(machine.get("port") or 22),
key_filename=key_filename or None,
private_key=private_key or None,
private_key_passphrase=passphrase or None,
password=str(machine.get("password") or "") or None,
known_hosts_path=str(settings.ssh_known_hosts_file),
)
return build_service_record(get_settings_store(), service_row)
@router.get("")
@@ -125,14 +92,14 @@ def list_task_runs(
) -> dict[str, Any]:
if not store.get_task(task_id):
raise HTTPException(status_code=404, detail="Task not found")
runs = store.list_task_runs(task_id, limit=limit)
runs = store.list_service_task_runs(task_id=task_id, limit=limit)
return {"items": runs, "total": len(runs)}
@router.post("/run")
def run_task(
request: RunTaskRequest,
machine_id: str | None = Query(default=None),
service_id: str | None = Query(default=None),
store: SettingsStore = Depends(get_settings_store),
) -> dict[str, Any]:
task = store.get_task(request.task_id)
@@ -141,68 +108,22 @@ def run_task(
if not task.get("enabled", True):
raise HTTPException(status_code=400, detail="Task is disabled")
machine = _resolve_machine_for_task(store, task, machine_id)
if not machine:
raise HTTPException(status_code=400, detail="No machine is available for this action")
service_row = _resolve_service_for_task(store, task, service_id)
if not service_row:
raise HTTPException(status_code=400, detail="No SSH task service is available for this action")
if not service_row.get("enabled", True):
raise HTTPException(status_code=400, detail="Selected SSH task service is disabled")
task_type = str(task.get("task_type") or "shell").lower()
command = str(task.get("content") or "")
if task_type == "python":
command = f"python3 -c {shlex.quote(command)}"
elif task_type != "shell":
raise HTTPException(status_code=400, detail=f"Unknown task type: {task_type}")
service = _service_row_to_record(service_row)
result = run_saved_task(store, task, service)
client = _client_for_machine(store, machine)
start = time.perf_counter()
machine_name = _machine_label(machine)
try:
result = client.run(command, timeout=1200)
stdout = result.stdout or ""
stderr = result.stderr or ""
status_text = "success" if result.exit_status == 0 else "error"
store.record_task_run(
task,
status_text,
machine_id=str(machine.get("id") or ""),
machine_name=machine_name,
task_type=task_type,
duration_ms=int((time.perf_counter() - start) * 1000),
stdout_tail=stdout[-4000:],
stderr_tail=stderr[-4000:],
error="" if result.exit_status == 0 else (stderr or stdout or "Task failed"),
)
return {
"task_id": task["id"],
"task_name": task["name"],
"machine_id": str(machine.get("id") or ""),
"machine_name": machine_name,
"task_type": task_type,
"exit_status": result.exit_status,
"stdout": stdout,
"stderr": stderr,
}
except Exception as exc:
duration_ms = int((time.perf_counter() - start) * 1000)
error_text = str(exc)
store.record_task_run(
task,
"error",
machine_id=str(machine.get("id") or ""),
machine_name=machine_name,
task_type=task_type,
duration_ms=duration_ms,
stdout_tail="",
stderr_tail=error_text[-4000:],
error=error_text,
)
logger.exception("Task execution failed task_id=%s", task["id"])
return {
"task_id": task["id"],
"task_name": task["name"],
"machine_id": str(machine.get("id") or ""),
"machine_name": machine_name,
"task_type": task_type,
"exit_status": 1,
"stdout": "",
"stderr": error_text,
}
return {
"task_id": task["id"],
"task_name": task["name"],
"service_id": service.id,
"service_name": _service_label(service_row),
"task_type": task.get("task_type", "shell"),
"exit_status": result.exit_status,
"stdout": result.stdout,
"stderr": result.stderr,
}
backend/src/media_library_viewer_api/services/settings_store.py
+18 -84
View File
@@ -113,7 +113,7 @@ class SettingsStore:
task_type TEXT NOT NULL,
content TEXT NOT NULL,
enabled INTEGER NOT NULL,
default_machine_id TEXT NOT NULL,
default_service_id TEXT NOT NULL,
notes TEXT NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
@@ -121,28 +121,14 @@ class SettingsStore:
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_saved_tasks_name ON saved_tasks(name)")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS saved_task_runs (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
task_name TEXT NOT NULL,
machine_id TEXT NOT NULL,
machine_name TEXT NOT NULL,
task_type TEXT NOT NULL,
status TEXT NOT NULL,
created_at INTEGER NOT NULL,
duration_ms INTEGER NOT NULL,
request_id TEXT NOT NULL,
stdout_tail TEXT NOT NULL,
stderr_tail TEXT NOT NULL,
error TEXT NOT NULL
)
"""
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_saved_task_runs_task_time ON saved_task_runs(task_id, created_at DESC)"
)
# saved_tasks.default_machine_id → default_service_id (saved tasks now
# target ssh_tasks service instances). Migrate existing columns.
saved_tasks_cols = {row[1] for row in conn.execute("PRAGMA table_info(saved_tasks)").fetchall()}
if "default_service_id" not in saved_tasks_cols and "default_machine_id" in saved_tasks_cols:
conn.execute("ALTER TABLE saved_tasks RENAME COLUMN default_machine_id TO default_service_id")
# Run history for saved tasks now lives in service_task_runs; the
# legacy machine-based table is dropped.
conn.execute("DROP TABLE IF EXISTS saved_task_runs")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS dashboard_shortcuts (
@@ -714,7 +700,7 @@ class SettingsStore:
"task_type": row["task_type"],
"content": row["content"],
"enabled": bool(row["enabled"]),
"default_machine_id": row["default_machine_id"],
"default_service_id": row["default_service_id"],
"notes": row["notes"],
"created_at": row["created_at"],
"updated_at": row["updated_at"],
@@ -731,10 +717,10 @@ class SettingsStore:
payload.get("content") if payload.get("content") is not None else (current or {}).get("content", "") or ""
)
enabled = bool(payload.get("enabled", (current or {}).get("enabled", True)))
default_machine_id = str(
payload.get("default_machine_id")
if payload.get("default_machine_id") is not None
else (current or {}).get("default_machine_id", "") or ""
default_service_id = str(
payload.get("default_service_id")
if payload.get("default_service_id") is not None
else (current or {}).get("default_service_id", "") or ""
).strip()
notes = str(
payload.get("notes") if payload.get("notes") is not None else (current or {}).get("notes", "") or ""
@@ -745,7 +731,7 @@ class SettingsStore:
"task_type": task_type,
"content": content,
"enabled": enabled,
"default_machine_id": default_machine_id,
"default_service_id": default_service_id,
"notes": notes,
}
@@ -773,7 +759,7 @@ class SettingsStore:
conn.execute(
"""
INSERT INTO saved_tasks (
id, name, task_type, content, enabled, default_machine_id,
id, name, task_type, content, enabled, default_service_id,
notes, created_at, updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
@@ -782,7 +768,7 @@ class SettingsStore:
task_type = excluded.task_type,
content = excluded.content,
enabled = excluded.enabled,
default_machine_id = excluded.default_machine_id,
default_service_id = excluded.default_service_id,
notes = excluded.notes,
updated_at = excluded.updated_at
""",
@@ -792,7 +778,7 @@ class SettingsStore:
task["task_type"],
task["content"],
1 if task["enabled"] else 0,
task["default_machine_id"],
task["default_service_id"],
task["notes"],
created_at,
now,
@@ -805,58 +791,6 @@ class SettingsStore:
with self.connect() as conn:
conn.execute("DELETE FROM saved_tasks WHERE id = ?", (task_id,))
def list_task_runs(self, task_id: str, *, limit: int = 10) -> list[dict[str, Any]]:
self.init_schema()
with self.connect() as conn:
rows = conn.execute(
"SELECT * FROM saved_task_runs WHERE task_id = ? ORDER BY created_at DESC LIMIT ?",
(task_id, max(1, min(int(limit), 50))),
).fetchall()
return [dict(row) for row in rows]
def record_task_run(
self,
task: dict[str, Any],
status: str,
*,
machine_id: str,
machine_name: str,
task_type: str,
duration_ms: int,
request_id: str = "",
stdout_tail: str = "",
stderr_tail: str = "",
error: str = "",
) -> None:
self.init_schema()
now = int(time.time())
with self.connect() as conn:
conn.execute(
"""
INSERT INTO saved_task_runs (
id, task_id, task_name, machine_id, machine_name, task_type,
status, created_at, duration_ms, request_id, stdout_tail,
stderr_tail, error
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
uuid.uuid4().hex,
str(task.get("id") or ""),
str(task.get("name") or ""),
machine_id,
machine_name,
task_type,
status,
now,
duration_ms,
request_id,
stdout_tail,
stderr_tail,
error,
),
)
def _row_to_shortcut(self, row: sqlite3.Row) -> dict[str, Any]:
target = json.loads(row["target_json"] or "{}")
return {
backend/src/media_library_viewer_api/services/task_runner.py
+166
View File
@@ -0,0 +1,166 @@
"""Shared runner for saved tasks over SSH task services.
Both the Actions page (``routers/tasks.py``) and the SSH task widget
(``widgets/sources.py``) run saved tasks against ``ssh_tasks`` service instances.
This module is the single execution path: build the client from the service
record, render the command, run it with the service timeout, append a
``service_task_runs`` row, and return the result.
There is intentionally no local execution mode — tasks are SSH-only.
"""
from __future__ import annotations
import logging
import shlex
import time
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from media_library_viewer_api.clients.ssh import RemoteSSHClient
from media_library_viewer_api.config import get_settings
from media_library_viewer_api.services.settings_store import SettingsStore
if TYPE_CHECKING:
from media_library_viewer_api.widgets.sources import ServiceRecord
logger = logging.getLogger(__name__)
@dataclass
class TaskRunResult:
"""Outcome of a single saved-task run."""
exit_status: int
stdout: str
stderr: str
duration_ms: int
status: str # "success" | "failure" | "error"
error: str = ""
def build_ssh_client(store: SettingsStore, service: "ServiceRecord") -> RemoteSSHClient:
"""Build an SSH client from an ssh_tasks service instance + referenced key."""
config = service.config
host = str(config.get("host") or "").strip()
username = str(config.get("username") or "").strip()
if not host or not username:
raise ValueError("SSH task service is missing host or username")
settings = get_settings()
private_key = ""
key_passphrase = ""
ssh_key_id = str(config.get("ssh_key_id") or "").strip()
if ssh_key_id:
ssh_key = store.get_ssh_key(ssh_key_id)
if ssh_key:
private_key = str(ssh_key.get("private_key") or "")
key_passphrase = str(ssh_key.get("passphrase") or "")
# Service-level passphrase secret takes precedence.
key_passphrase = str(service.secrets.get("passphrase") or "") or key_passphrase
return RemoteSSHClient(
host=host,
username=username,
port=int(config.get("port") or 22),
private_key=private_key or None,
private_key_passphrase=key_passphrase or None,
known_hosts_path=str(settings.ssh_known_hosts_file),
timeout=int(config.get("timeout_seconds") or 30),
)
def _render_command(task: dict[str, Any]) -> str:
"""Render a saved task into a shell command (shell or python3 -c)."""
task_type = str(task.get("task_type") or "shell").lower()
command = str(task.get("content") or "")
if task_type == "python":
return f"python3 -c {shlex.quote(command)}"
if task_type == "shell":
return command
raise ValueError(f"Unknown task type: {task_type}")
def run_saved_task(
store: SettingsStore,
task: dict[str, Any],
service: "ServiceRecord",
*,
timeout: int | None = None,
) -> TaskRunResult:
"""Run a saved task on an ssh_tasks service instance and log the run.
The ``timeout`` defaults to the service's ``timeout_seconds`` config. The run
is recorded in ``service_task_runs`` regardless of outcome (success, failure,
error). Raises ``ValueError`` for an unsupported task type or an incomplete
service config (propagated from ``build_ssh_client`` / ``_render_command``).
"""
timeout = int(timeout if timeout is not None else service.config.get("timeout_seconds") or 30)
client = build_ssh_client(store, service)
command = _render_command(task)
start = time.perf_counter()
try:
result = client.run(command, timeout=timeout)
except Exception as exc:
duration_ms = int((time.perf_counter() - start) * 1000)
_record(store, task, service, "error", duration_ms=duration_ms, error=str(exc)[:1000])
logger.exception("saved task run failed task_id=%s", task.get("id"))
return TaskRunResult(
exit_status=1,
stdout="",
stderr=str(exc),
duration_ms=duration_ms,
status="error",
error=str(exc),
)
duration_ms = int((time.perf_counter() - start) * 1000)
stdout = result.stdout or ""
stderr = result.stderr or ""
status = "success" if result.exit_status == 0 else "failure"
_record(
store,
task,
service,
status,
exit_status=result.exit_status,
duration_ms=duration_ms,
stdout_tail=stdout,
stderr_tail=stderr,
error="" if result.exit_status == 0 else (stderr or stdout or "Task failed"),
)
return TaskRunResult(
exit_status=result.exit_status,
stdout=stdout,
stderr=stderr,
duration_ms=duration_ms,
status=status,
)
def _record(
store: SettingsStore,
task: dict[str, Any],
service: "ServiceRecord",
status: str,
*,
exit_status: int | None = None,
duration_ms: int = 0,
stdout_tail: str = "",
stderr_tail: str = "",
error: str = "",
) -> None:
"""Append a service_task_runs row for a saved-task run."""
store.record_service_task_run(
{
"task_id": str(task.get("id") or ""),
"service_id": service.id,
"status": status,
"exit_status": exit_status,
"duration_ms": duration_ms,
"stdout_tail": stdout_tail,
"stderr_tail": stderr_tail,
"error": error,
}
)
backend/src/media_library_viewer_api/widgets/sources.py
+3 -71
View File
@@ -12,21 +12,18 @@ from __future__ import annotations
import asyncio
import logging
import shlex
import time
from dataclasses import dataclass, field
from typing import Any, Protocol
import requests
from media_library_viewer_api.clients.jellyfin import JellyfinClient
from media_library_viewer_api.clients.ssh import RemoteSSHClient
from media_library_viewer_api.config import get_settings
from media_library_viewer_api.domain.dashboard import (
_map_sessions_to_activity_rows,
build_backup_dashboard_summary,
)
from media_library_viewer_api.services.settings_store import SettingsStore, get_settings_store
from media_library_viewer_api.services.task_runner import run_saved_task
logger = logging.getLogger(__name__)
@@ -200,51 +197,17 @@ class SshTaskWidgetSource:
if not task.get("enabled", True):
return {"error": "Task is disabled"}
client = _build_ssh_client(store, service)
timeout = int(service.config.get("timeout_seconds") or 30)
task_type = str(task.get("task_type") or "shell").lower()
command = str(task.get("content") or "")
if task_type == "python":
command = f"python3 -c {shlex.quote(command)}"
elif task_type != "shell":
return {"error": f"Unknown task type: {task_type}"}
start = time.perf_counter()
result = await asyncio.wait_for(
asyncio.to_thread(client.run, command, timeout),
asyncio.to_thread(run_saved_task, store, task, service),
timeout=timeout,
)
duration_ms = int((time.perf_counter() - start) * 1000)
stdout = result.stdout or ""
stderr = result.stderr or ""
store.record_service_task_run(
{
"task_id": task_id,
"service_id": service.id,
"status": "success" if result.exit_status == 0 else "failure",
"exit_status": result.exit_status,
"duration_ms": duration_ms,
"stdout_tail": stdout,
"stderr_tail": stderr,
"error": "" if result.exit_status == 0 else (stderr or stdout or "Task failed"),
}
)
return {"exit_status": result.exit_status, "stdout": stdout, "stderr": stderr}
return {"exit_status": result.exit_status, "stdout": result.stdout, "stderr": result.stderr}
except asyncio.TimeoutError:
_record_timeout(service, config, timeout)
return {"error": "Widget data fetch timed out"}
except Exception as exc:
logger.exception("ssh_task adapter failed")
store = get_settings_store()
store.record_service_task_run(
{
"task_id": str(config.get("task_id") or ""),
"service_id": service.id if service else "",
"status": "error",
"duration_ms": 0,
"error": str(exc)[:1000],
}
)
return {"error": f"SSH task failed: {exc}"}
@@ -264,37 +227,6 @@ def _record_timeout(service: ServiceRecord | None, config: dict[str, Any], timeo
logger.exception("failed to record ssh task timeout")
def _build_ssh_client(store: SettingsStore, service: ServiceRecord) -> RemoteSSHClient:
"""Build an SSH client from an ssh_tasks service instance + referenced key."""
config = service.config
host = str(config.get("host") or "").strip()
username = str(config.get("username") or "").strip()
if not host or not username:
raise ValueError("SSH task service is missing host or username")
settings = get_settings()
private_key = ""
key_passphrase = ""
ssh_key_id = str(config.get("ssh_key_id") or "").strip()
if ssh_key_id:
ssh_key = store.get_ssh_key(ssh_key_id)
if ssh_key:
private_key = str(ssh_key.get("private_key") or "")
key_passphrase = str(ssh_key.get("passphrase") or "")
# Service-level passphrase secret takes precedence.
key_passphrase = str(service.secrets.get("passphrase") or "") or key_passphrase
return RemoteSSHClient(
host=host,
username=username,
port=int(config.get("port") or 22),
private_key=private_key or None,
private_key_passphrase=key_passphrase or None,
known_hosts_path=str(settings.ssh_known_hosts_file),
timeout=int(config.get("timeout_seconds") or 30),
)
# ---------------------------------------------------------------------------
# Registries
# ---------------------------------------------------------------------------
backend/tests/test_widgets.py
+3 -2
View File
@@ -359,7 +359,7 @@ async def test_ssh_task_adapter_records_history_on_run(client):
"task_type": "shell",
"content": "echo hi",
"enabled": True,
"default_machine_id": "",
"default_service_id": "",
}
)
service = store.upsert_service(
@@ -369,6 +369,7 @@ async def test_ssh_task_adapter_records_history_on_run(client):
fake_result = SimpleNamespace(exit_status=0, stdout="hi\n", stderr="")
fake_client = SimpleNamespace(run=lambda *a, **k: fake_result)
from media_library_viewer_api.services.task_runner import build_ssh_client
from media_library_viewer_api.widgets.sources import SshTaskWidgetSource
adapter = SshTaskWidgetSource()
@@ -377,7 +378,7 @@ async def test_ssh_task_adapter_records_history_on_run(client):
)
with (
patch("media_library_viewer_api.widgets.sources.get_settings_store", return_value=store),
patch("media_library_viewer_api.widgets.sources._build_ssh_client", return_value=fake_client),
patch("media_library_viewer_api.services.task_runner.build_ssh_client", return_value=fake_client),
):
result = await adapter.fetch(service_record, "task_output", {"task_id": task["id"]})