ctfy.sdk.admin

  1"""Operator / admin client for the ctfy platform.
  2
  3``AdminClient`` is the operator-facing counterpart to
  4:class:`~ctfy.sdk.client.PlatformClient`, mirroring the ``ctfy`` /
  5``ctfy-admin`` CLI split: the player client never surfaces admin methods, so
  6agents and players reading the SDK reference aren't distracted by operator
  7endpoints they can't call.
  8
  9Reach it two ways:
 10
 11* ``client.admin`` on an existing :class:`PlatformClient` — shares that
 12  client's transport (connection pool + bearer)::
 13
 14      from ctfy.sdk import PlatformClient
 15
 16      client = PlatformClient("https://ctfy.example", token="pf_xxx")
 17      client.admin.users.set_role(user_id, "admin")
 18
 19* :meth:`AdminClient.connect` for operator tooling that doesn't already hold
 20  a player client::
 21
 22      from ctfy.sdk import AdminClient
 23
 24      with AdminClient.connect("https://ctfy.example", token="pf_xxx") as admin:
 25          admin.observability.overview()
 26
 27Every endpoint here is role-gated server-side (admin / super-admin); the SDK
 28only relays — the token's role decides access.
 29"""
 30
 31from __future__ import annotations
 32
 33from functools import cached_property
 34from types import TracebackType
 35from typing import Self
 36
 37from ctfy.core.constants import DEFAULT_CLIENT_TIMEOUT
 38from ctfy.sdk.admin_resources.achievements import AdminAchievementsResource
 39from ctfy.sdk.admin_resources.announcements import AdminAnnouncementsResource
 40from ctfy.sdk.admin_resources.challenges import AdminChallengesResource
 41from ctfy.sdk.admin_resources.competition_admins import AdminCompetitionAdminsResource
 42from ctfy.sdk.admin_resources.competitions import AdminCompetitionsResource
 43from ctfy.sdk.admin_resources.instances import AdminInstancesResource
 44from ctfy.sdk.admin_resources.nodes import AdminNodesResource
 45from ctfy.sdk.admin_resources.observability import AdminObservabilityResource
 46from ctfy.sdk.admin_resources.records import AdminRecordsResource
 47from ctfy.sdk.admin_resources.settings import AdminSettingsResource
 48from ctfy.sdk.admin_resources.users import AdminUsersResource
 49from ctfy.sdk.base import BaseHttpClient
 50
 51__all__ = ["AdminClient"]
 52
 53
 54class AdminClient:
 55    """Admin / operator surface, grouped into resource namespaces.
 56
 57    Namespaces: :attr:`users`, :attr:`competitions`, :attr:`announcements`,
 58    :attr:`achievements`, :attr:`challenges`, :attr:`instances`,
 59    :attr:`records`, :attr:`nodes`, :attr:`observability`, :attr:`settings`,
 60    :attr:`competition_admins`.
 61    """
 62
 63    def __init__(self, http: BaseHttpClient) -> None:
 64        #: Shared transport (the parent PlatformClient when reached via
 65        #: ``client.admin``, or an owned client when built via ``connect``).
 66        self._http = http
 67
 68    @classmethod
 69    def connect(
 70        cls,
 71        server_url: str,
 72        token: str = "",
 73        *,
 74        max_retries: int = 3,
 75        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 76    ) -> AdminClient:
 77        """Build an ``AdminClient`` that owns its own transport.
 78
 79        For operator tooling that doesn't already hold a
 80        :class:`PlatformClient`. The returned client is a context manager
 81        that closes its transport on exit.
 82        """
 83        base = server_url.rstrip("/")
 84        http = BaseHttpClient(f"{base}/api/v1", token, timeout=timeout, max_retries=max_retries)
 85        return cls(http)
 86
 87    @cached_property
 88    def users(self) -> AdminUsersResource:
 89        return AdminUsersResource(self._http)
 90
 91    @cached_property
 92    def competitions(self) -> AdminCompetitionsResource:
 93        return AdminCompetitionsResource(self._http)
 94
 95    @cached_property
 96    def announcements(self) -> AdminAnnouncementsResource:
 97        return AdminAnnouncementsResource(self._http)
 98
 99    @cached_property
100    def achievements(self) -> AdminAchievementsResource:
101        return AdminAchievementsResource(self._http)
102
103    @cached_property
104    def challenges(self) -> AdminChallengesResource:
105        return AdminChallengesResource(self._http)
106
107    @cached_property
108    def instances(self) -> AdminInstancesResource:
109        return AdminInstancesResource(self._http)
110
111    @cached_property
112    def records(self) -> AdminRecordsResource:
113        return AdminRecordsResource(self._http)
114
115    @cached_property
116    def nodes(self) -> AdminNodesResource:
117        return AdminNodesResource(self._http)
118
119    @cached_property
120    def observability(self) -> AdminObservabilityResource:
121        return AdminObservabilityResource(self._http)
122
123    @cached_property
124    def settings(self) -> AdminSettingsResource:
125        return AdminSettingsResource(self._http)
126
127    @cached_property
128    def competition_admins(self) -> AdminCompetitionAdminsResource:
129        return AdminCompetitionAdminsResource(self._http)
130
131    def close(self) -> None:
132        """Close the underlying transport.
133
134        Only call this on a standalone client built via :meth:`connect`;
135        when reached via ``PlatformClient.admin`` the transport is shared
136        with — and closed by — the parent player client.
137        """
138        self._http.close()
139
140    def __enter__(self) -> Self:
141        return self
142
143    def __exit__(
144        self,
145        exc_type: type[BaseException] | None,
146        exc: BaseException | None,
147        tb: TracebackType | None,
148    ) -> None:
149        self.close()