ctfy.sdk.resources.tasks

 1"""``client.tasks`` -- the background tasks you may see (#2098), the same read
 2as the web 「我的任务」 page."""
 3
 4from __future__ import annotations
 5
 6from typing import Any
 7
 8from ctfy.sdk._helpers import _raise_for_status
 9from ctfy.sdk.base import BaseHttpClient
10from ctfy.server.models import MyTaskDetail, TaskCard, TaskCardList
11
12
13class TasksResource:
14    """Smoke runs, evaluation runs, report grading and the rest of the work the
15    platform does for you in the background: when each waited, ran and ended,
16    and how it ended.
17
18    What comes back is decided by the server: yours and your team's, those
19    about an upload you wrote, a competition's for its administrators, and for
20    a platform admin everything (``scope="all"``).
21    """
22
23    def __init__(self, http: BaseHttpClient) -> None:
24        self._http = http
25
26    def list(
27        self,
28        *,
29        scope: str = "visible",
30        type: str = "",
31        status: str = "",
32        range: str = "",
33        subject: str = "",
34        offset: int = 0,
35        limit: int = 50,
36    ) -> TaskCardList:
37        """One page of tasks, newest first, with the ``total`` and the status
38        ``counts`` of the same scope.
39
40        ``type`` is ``smoke`` / ``eval`` / ``grading`` / ``build`` / ``system``;
41        ``status`` is ``active`` / ``failed`` or a stored status; ``range`` is
42        ``24h`` / ``7d`` / ``30d`` (empty = no window).
43        """
44        params: dict[str, Any] = {"scope": scope, "offset": offset, "limit": limit}
45        wanted = {"type": type, "status": status, "range": range, "subject": subject}
46        params.update({key: value for key, value in wanted.items() if value})
47        resp = self._http.request("GET", "/me/tasks", params=params)
48        _raise_for_status(resp)
49        return TaskCardList.model_validate(resp.json())
50
51    def get(self, task_id: str) -> TaskCard:
52        """One task as a card; a task that is not yours to see is a 404."""
53        resp = self._http.request("GET", f"/me/tasks/{task_id}")
54        _raise_for_status(resp)
55        return TaskCard.model_validate(resp.json())
56
57    def detail(self, task_id: str) -> MyTaskDetail:
58        """Visible task and the owner's personal evaluation, when applicable."""
59        resp = self._http.request("GET", f"/me/tasks/{task_id}/detail")
60        _raise_for_status(resp)
61        return MyTaskDetail.model_validate(resp.json())
62
63    def cancel(self, task_id: str) -> MyTaskDetail:
64        """Request cancellation of your own personal evaluation only."""
65        resp = self._http.request("POST", f"/me/tasks/{task_id}/cancel")
66        _raise_for_status(resp)
67        return MyTaskDetail.model_validate(resp.json())