src/genro_asgi/applications/server_sections/tasks_section.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """The ``_server/tasks`` section: SUPERADMIN-gated task backbone endpoints.16 17 A ``RoutingClass`` the ``ServerApplication`` attaches (``attach_section``), so18 its routes live at ``/_server/tasks/...``. JSON endpoints ONLY — no HTML/JS19 panel ships with the core (ratified). Two surfaces over the server's20 ``TaskManager``:21 22 - **schedules** (the recurring scheduler's store): ``list``/``create``/23 ``update``/``enable``/``disable``/``run_now``/``delete``/``logs``;24 - **spool** (the batch folders): ``spool_list`` (by ``owner`` or ``status``),25 ``progress``, ``cancel``, ``result``.26 27 Every route is gated ``auth_rule="SUPERADMIN"``. A server composed without the28 task backbone — or with ``tasks=False`` — answers every endpoint with the29 ``{"error": ...}`` document (HTTP 200): the section is ALWAYS declared, fixed30 structure (D26), the payload states the availability.31 32 Parent (dual relationship): the section holds its ``ServerApplication`` as33 ``self.application`` and reaches the manager via ``self.application.server.tasks``34 (guarded by ``tasks_enabled`` — the property raises when disabled).35 """36 37 from __future__ import annotations38 39 import time40 from typing import TYPE_CHECKING, Any41 42 from genro_routes import RoutingClass, route43 44 from ...tasks.schedule import TaskCadence45 from ...tasks.spool import STATUSES46 47 if TYPE_CHECKING:48 from ...tasks.manager import TaskManager49 from ..server_app import ServerApplication50 51 __all__ = ["TasksSection"]52 53 TASKS_DISABLED_ERROR = {"error": "Tasks are disabled"}54 55 # The record fields a client may set through create/update (the run outcome56 # fields — last_*, next_run_ts — belong to the scheduler, never to the wire).57 EDITABLE_FIELDS = ("task_name", "kwargs", "kind", "spec", "enabled")58 59 60 class TasksSection(RoutingClass):61 """The ``/_server/tasks`` endpoints over the server's task backbone.62 63 Note:64 Bound to its ``ServerApplication`` (dual relationship:65 ``self.application``); the manager, store, scheduler and spool are66 reached through ``self.application.server.tasks``.67 """68 69 def __init__(self, application: ServerApplication) -> None:70 """Bind the section to its ServerApplication (dual relationship)."""71 self.application = application72 73 @property74 def manager(self) -> TaskManager | None:75 """The server's TaskManager, or ``None`` when tasks are off/absent."""76 server = self.application.server77 if not getattr(server, "tasks_enabled", False):78 return None79 return server.tasks80 81 # -- schedules (the recurring scheduler's store) --82 83 @route(auth_rule="SUPERADMIN")84 def list(self) -> dict[str, Any]:85 """Every schedule record."""86 manager = self.manager87 if manager is None:88 return TASKS_DISABLED_ERROR89 return {"schedules": manager.task_store.load_all()}90 91 @route(auth_rule="SUPERADMIN", openapi_method="post")92 def create(self, body_data: dict | None = None) -> dict[str, Any]:93 """Create a schedule: ``code``, ``kind``, ``spec`` required.94 95 ``task_name`` defaults to ``code``; ``kwargs`` and ``enabled`` are96 optional. ``next_run_ts`` is computed here (an invalid spec is the97 ``{"error": ...}`` answer, not a record).98 """99 manager = self.manager100 if manager is None:101 return TASKS_DISABLED_ERROR102 body = body_data or {}103 code = body.get("code") or ""104 kind, spec = body.get("kind"), body.get("spec")105 if not code or not kind or spec is None:106 return {"error": "code, kind and spec are required"}107 if manager.task_store.get(code) is not None:108 return {"error": f"schedule already exists: {code}"}109 try:110 first_run = TaskCadence(kind, spec).get_next_run(time.time())111 except ValueError as exc:112 return {"error": str(exc)}113 record = {114 "code": code,115 "task_name": body.get("task_name") or code,116 "target_kind": "task",117 "kwargs": body.get("kwargs") or {},118 "kind": kind,119 "spec": spec,120 "enabled": bool(body.get("enabled", True)),121 "next_run_ts": first_run,122 "last_run_ts": None,123 "last_outcome": None,124 "last_error": None,125 "last_duration": None,126 }127 manager.task_store.save(record)128 return {"schedule": record}129 130 @route(auth_rule="SUPERADMIN", openapi_method="post")131 def update(self, body_data: dict | None = None) -> dict[str, Any]:132 """Merge the editable fields into a schedule (``code`` names it).133 134 A changed ``kind``/``spec`` recomputes ``next_run_ts``; the run-outcome135 fields are the scheduler's and never settable from the wire.136 """137 manager = self.manager138 if manager is None:139 return TASKS_DISABLED_ERROR140 body = body_data or {}141 code = body.get("code") or ""142 record = manager.task_store.get(code)143 if record is None:144 return {"error": f"schedule not found: {code}"}145 record.update({field: body[field] for field in EDITABLE_FIELDS if field in body})146 if "kind" in body or "spec" in body:147 try:148 record["next_run_ts"] = TaskCadence(record["kind"], record["spec"]).get_next_run(149 time.time()150 )151 except ValueError as exc:152 return {"error": str(exc)}153 manager.task_store.save(record)154 return {"schedule": record}155 156 @route(auth_rule="SUPERADMIN", openapi_method="post")157 def enable(self, code: str = "") -> dict[str, Any]:158 """Arm a schedule."""159 return self._set_enabled(code, value=True)160 161 @route(auth_rule="SUPERADMIN", openapi_method="post")162 def disable(self, code: str = "") -> dict[str, Any]:163 """Disarm a schedule (the record stays)."""164 return self._set_enabled(code, value=False)165 166 def _set_enabled(self, code: str, *, value: bool) -> dict[str, Any]:167 """Flip a schedule's enabled flag; unknown code is the error shape."""168 manager = self.manager169 if manager is None:170 return TASKS_DISABLED_ERROR171 record = manager.task_store.set_enabled(code, value)172 if record is None:173 return {"error": f"schedule not found: {code}"}174 return {"schedule": record}175 176 @route(auth_rule="SUPERADMIN", openapi_method="post")177 def run_now(self, code: str = "") -> dict[str, Any]:178 """Fire a schedule immediately (same no-overlap guard as the loop)."""179 manager = self.manager180 if manager is None:181 return TASKS_DISABLED_ERROR182 try:183 outcome = manager.scheduler.run_now(code)184 except LookupError as exc:185 return {"error": str(exc)}186 return {"code": code, "run": outcome}187 188 @route(auth_rule="SUPERADMIN", openapi_method="post")189 def delete(self, code: str = "") -> dict[str, Any]:190 """Remove a schedule record. ``deleted`` reports whether it existed."""191 manager = self.manager192 if manager is None:193 return TASKS_DISABLED_ERROR194 return {"code": code, "deleted": manager.task_store.delete(code)}195 196 @route(auth_rule="SUPERADMIN")197 def logs(self, task_name: str = "", limit: str = "") -> dict[str, Any]:198 """A task's capped JSONL run log, oldest first."""199 manager = self.manager200 if manager is None:201 return TASKS_DISABLED_ERROR202 if not task_name:203 return {"error": "task_name is required"}204 store = manager.task_store205 entries = store.read_log(task_name, int(limit)) if limit else store.read_log(task_name)206 return {"task_name": task_name, "log": entries}207 208 # -- spool (the batch folders) --209 210 @route(auth_rule="SUPERADMIN")211 def spool_list(self, owner: str = "", status: str = "") -> dict[str, Any]:212 """Task descriptors by ``owner`` OR by ``status`` (one filter required)."""213 manager = self.manager214 if manager is None:215 return TASKS_DISABLED_ERROR216 if owner:217 return {"tasks": manager.spool.list_by_owner(owner)}218 if status:219 if status not in STATUSES:220 return {"error": f"unknown status: {status} (want one of {', '.join(STATUSES)})"}221 return {"tasks": manager.spool.list_by_status(status)}222 return {"error": "owner or status is required"}223 224 @route(auth_rule="SUPERADMIN")225 def progress(self, task_id: str = "") -> dict[str, Any]:226 """A task's latest progress snapshot (``None`` until the worker writes)."""227 manager = self.manager228 if manager is None:229 return TASKS_DISABLED_ERROR230 if manager.spool.get(task_id) is None:231 return {"error": f"task not found: {task_id}"}232 return {"task_id": task_id, "progress": manager.spool.read_progress(task_id)}233 234 @route(auth_rule="SUPERADMIN", openapi_method="post")235 def cancel(self, task_id: str = "") -> dict[str, Any]:236 """Drop the cancel marker (the worker honors it at its own pace)."""237 manager = self.manager238 if manager is None:239 return TASKS_DISABLED_ERROR240 try:241 manager.spool.request_cancel(task_id)242 except LookupError as exc:243 return {"error": str(exc)}244 return {"task_id": task_id, "cancelled": True}245 246 @route(auth_rule="SUPERADMIN")247 def result(self, task_id: str = "") -> dict[str, Any]:248 """A task's result (``None`` until written; JSON-encodable results only)."""249 manager = self.manager250 if manager is None:251 return TASKS_DISABLED_ERROR252 if manager.spool.get(task_id) is None:253 return {"error": f"task not found: {task_id}"}254 return {"task_id": task_id, "result": manager.spool.read_result(task_id)}