ctfy.sdk.admin_resources.tasks

 1"""``client.admin.tasks`` + ``client.admin.scheduled_jobs`` — the
 2background-task system (admin)."""
 3
 4from __future__ import annotations
 5
 6import builtins
 7from typing import Any
 8
 9from ctfy.sdk._helpers import _raise_for_status
10from ctfy.sdk.base import BaseHttpClient
11from ctfy.server.models import (
12    AdminTaskInfo,
13    AdminTaskListResponse,
14    AdminTaskLogsResponse,
15    ScheduledJobInfo,
16)
17
18
19class AdminTasksResource:
20    """One-shot background jobs: list / view / submit / cancel."""
21
22    def __init__(self, http: BaseHttpClient) -> None:
23        self._http = http
24
25    def list(
26        self, *, status: str = "", kind: str = "", offset: int = 0, limit: int = 50
27    ) -> AdminTaskListResponse:
28        """Paginated task list, newest first; filter by ``status`` / ``kind``."""
29        resp = self._http.request(
30            "GET",
31            "/admin/tasks",
32            params={"status": status, "kind": kind, "offset": offset, "limit": limit},
33        )
34        _raise_for_status(resp)
35        return AdminTaskListResponse.model_validate(resp.json())
36
37    def get(self, task_id: str) -> AdminTaskInfo:
38        resp = self._http.request("GET", f"/admin/tasks/{task_id}")
39        _raise_for_status(resp)
40        return AdminTaskInfo.model_validate(resp.json())
41
42    def logs(self, task_id: str, *, after: int = 0, limit: int = 1000) -> AdminTaskLogsResponse:
43        """Cursor page of a task's verbose logs (lines with ``seq > after``).
44        Poll with the returned ``next_after`` to tail a running task."""
45        resp = self._http.request(
46            "GET", f"/admin/tasks/{task_id}/logs", params={"after": after, "limit": limit}
47        )
48        _raise_for_status(resp)
49        return AdminTaskLogsResponse.model_validate(resp.json())
50
51    def submit(self, kind: str, params: dict[str, Any] | None = None) -> AdminTaskInfo:
52        """Queue a one-shot task (``challenge_build_all`` / ``_pull_all`` /
53        ``_rescan`` / ``scheduled_job_run``). Returns the pending row; the
54        dispatcher runs it within a tick."""
55        resp = self._http.request(
56            "POST", "/admin/tasks", json={"kind": kind, "params": params or {}}
57        )
58        _raise_for_status(resp)
59        return AdminTaskInfo.model_validate(resp.json())
60
61    def cancel(self, task_id: str) -> AdminTaskInfo:
62        """Cancel a task. Pending → cancelled immediately; running → a
63        cooperative-cancel flag the handler observes between node calls."""
64        resp = self._http.request("POST", f"/admin/tasks/{task_id}/cancel")
65        _raise_for_status(resp)
66        return AdminTaskInfo.model_validate(resp.json())
67
68
69class AdminScheduledJobsResource:
70    """Recurring jobs: list config + heartbeat, pause/resume/retune, run-now."""
71
72    def __init__(self, http: BaseHttpClient) -> None:
73        self._http = http
74
75    def list(self) -> builtins.list[ScheduledJobInfo]:
76        resp = self._http.request("GET", "/admin/scheduled-jobs")
77        _raise_for_status(resp)
78        return [ScheduledJobInfo.model_validate(row) for row in resp.json()]
79
80    def update(
81        self, name: str, *, enabled: bool | None = None, interval_s: int | None = None
82    ) -> ScheduledJobInfo:
83        """Pause / resume (``enabled``) or re-tune the cadence
84        (``interval_s``, clamped server-side to [5, 86400])."""
85        body: dict[str, Any] = {}
86        if enabled is not None:
87            body["enabled"] = enabled
88        if interval_s is not None:
89            body["interval_s"] = interval_s
90        resp = self._http.request("PATCH", f"/admin/scheduled-jobs/{name}", json=body)
91        _raise_for_status(resp)
92        return ScheduledJobInfo.model_validate(resp.json())
93
94    def run(self, name: str) -> AdminTaskInfo:
95        """Trigger an ad-hoc run now — submits a ``scheduled_job_run`` task."""
96        resp = self._http.request("POST", f"/admin/scheduled-jobs/{name}/run")
97        _raise_for_status(resp)
98        return AdminTaskInfo.model_validate(resp.json())