ctfy.sdk.admin_resources.competitions

 1"""``client.admin.competitions`` — competition CRUD + roster (admin)."""
 2
 3from __future__ import annotations
 4
 5import builtins
 6from datetime import datetime
 7from typing import Any
 8
 9from ctfy.sdk._helpers import _raise_for_status
10from ctfy.sdk.base import BaseHttpClient
11from ctfy.server.models import CompetitionInfo, CompetitionRegistrationInfo
12
13
14class AdminCompetitionsResource:
15    """Create / update / delete competitions and inspect their roster."""
16
17    def __init__(self, http: BaseHttpClient) -> None:
18        self._http = http
19
20    def list(self, offset: int = 0, limit: int = 50) -> builtins.list[CompetitionInfo]:
21        """Every competition — past, current, scheduled."""
22        resp = self._http.request(
23            "GET", "/admin/competitions", params={"offset": offset, "limit": limit}
24        )
25        _raise_for_status(resp)
26        return [CompetitionInfo.model_validate(c) for c in resp.json()["items"]]
27
28    def create(
29        self,
30        *,
31        title: str,
32        description: str = "",
33        starts_at: datetime | None = None,
34        ends_at: datetime | None = None,
35        challenge_ids: builtins.list[str] | None = None,
36        status: str = "draft",
37        access: str = "public",
38    ) -> CompetitionInfo:
39        """Create a new competition. ``None`` time fields = no bound
40        (sent as JSON ``null``). Unknown ``challenge_ids`` → 422.
41        ``status`` defaults to ``draft`` (hidden from players); pass
42        ``"published"`` to make it immediately visible. ``access``
43        defaults to ``public``; pass ``"private_listed"`` /
44        ``"private_hidden"`` for an invite-only competition."""
45        resp = self._http.request(
46            "POST",
47            "/admin/competitions",
48            json={
49                "title": title,
50                "description": description,
51                "starts_at": starts_at.isoformat() if starts_at else None,
52                "ends_at": ends_at.isoformat() if ends_at else None,
53                "challenge_ids": challenge_ids or [],
54                "status": status,
55                "access": access,
56            },
57        )
58        _raise_for_status(resp)
59        return CompetitionInfo.model_validate(resp.json())
60
61    def update(self, competition_id: str, **fields: Any) -> CompetitionInfo:
62        """PATCH any subset of ``title`` / ``description`` / ``starts_at`` /
63        ``ends_at`` / ``challenge_ids``. Omit a field to leave it; pass
64        ``None`` to clear."""
65        resp = self._http.request("PATCH", f"/admin/competitions/{competition_id}", json=fields)
66        _raise_for_status(resp)
67        return CompetitionInfo.model_validate(resp.json())
68
69    def delete(self, competition_id: str) -> None:
70        """Hard-delete a competition. Cascades against per-comp teams,
71        memberships, invites, solve / submission rows."""
72        resp = self._http.request("DELETE", f"/admin/competitions/{competition_id}")
73        _raise_for_status(resp)
74
75    def registrations(
76        self,
77        competition_id: str,
78        offset: int = 0,
79        limit: int = 50,
80    ) -> builtins.list[CompetitionRegistrationInfo]:
81        """Roster for ``competition_id``. Per the per-comp refactor, a
82        team's existence with the comp scope IS the registration."""
83        resp = self._http.request(
84            "GET",
85            f"/admin/competitions/{competition_id}/registrations",
86            params={"offset": offset, "limit": limit},
87        )
88        _raise_for_status(resp)
89        return [CompetitionRegistrationInfo.model_validate(r) for r in resp.json()["items"]]