ctfy.sdk.admin

Operator / admin client for the ctfy platform.

AdminClient is the operator-facing counterpart to ~ctfy.sdk.client.PlatformClient, mirroring the ctfy / ctfy-admin CLI split: the player client never surfaces admin methods, so agents and players reading the SDK reference aren't distracted by operator endpoints they can't call.

Reach it two ways:

  • client.admin on an existing PlatformClient — shares that client's transport (connection pool + bearer)::

    from ctfy.sdk import PlatformClient

client = PlatformClient("https://ctfy.example", token="pf_xxx")
client.admin.users.set_role(user_id, "admin")

  • AdminClient.connect() for operator tooling that doesn't already hold a player client::

    from ctfy.sdk import AdminClient

with AdminClient.connect("https://ctfy.example", token="pf_xxx") as admin:
    admin.observability.overview()

Every endpoint here is role-gated server-side (admin / super-admin); the SDK only relays — the token's role decides access.

  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.competition_invites import AdminCompetitionInvitesResource
 43from ctfy.sdk.admin_resources.competitions import AdminCompetitionsResource
 44from ctfy.sdk.admin_resources.instances import AdminInstancesResource
 45from ctfy.sdk.admin_resources.nodes import AdminNodesResource
 46from ctfy.sdk.admin_resources.observability import AdminObservabilityResource
 47from ctfy.sdk.admin_resources.records import AdminRecordsResource
 48from ctfy.sdk.admin_resources.settings import AdminSettingsResource
 49from ctfy.sdk.admin_resources.tasks import (
 50    AdminScheduledJobsResource,
 51    AdminTasksResource,
 52)
 53from ctfy.sdk.admin_resources.users import AdminUsersResource
 54from ctfy.sdk.base import BaseHttpClient
 55
 56__all__ = ["AdminClient"]
 57
 58
 59class AdminClient:
 60    """Admin / operator surface, grouped into resource namespaces.
 61
 62    Namespaces: :attr:`users`, :attr:`competitions`, :attr:`announcements`,
 63    :attr:`achievements`, :attr:`challenges`, :attr:`instances`,
 64    :attr:`records`, :attr:`nodes`, :attr:`observability`, :attr:`settings`,
 65    :attr:`competition_admins`.
 66    """
 67
 68    def __init__(self, http: BaseHttpClient) -> None:
 69        #: Shared transport (the parent PlatformClient when reached via
 70        #: ``client.admin``, or an owned client when built via ``connect``).
 71        self._http = http
 72
 73    @classmethod
 74    def connect(
 75        cls,
 76        server_url: str,
 77        token: str = "",
 78        *,
 79        max_retries: int = 3,
 80        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 81    ) -> AdminClient:
 82        """Build an ``AdminClient`` that owns its own transport.
 83
 84        For operator tooling that doesn't already hold a
 85        :class:`PlatformClient`. The returned client is a context manager
 86        that closes its transport on exit.
 87        """
 88        base = server_url.rstrip("/")
 89        http = BaseHttpClient(f"{base}/api/v1", token, timeout=timeout, max_retries=max_retries)
 90        return cls(http)
 91
 92    @cached_property
 93    def users(self) -> AdminUsersResource:
 94        return AdminUsersResource(self._http)
 95
 96    @cached_property
 97    def competitions(self) -> AdminCompetitionsResource:
 98        return AdminCompetitionsResource(self._http)
 99
100    @cached_property
101    def announcements(self) -> AdminAnnouncementsResource:
102        return AdminAnnouncementsResource(self._http)
103
104    @cached_property
105    def achievements(self) -> AdminAchievementsResource:
106        return AdminAchievementsResource(self._http)
107
108    @cached_property
109    def challenges(self) -> AdminChallengesResource:
110        return AdminChallengesResource(self._http)
111
112    @cached_property
113    def instances(self) -> AdminInstancesResource:
114        return AdminInstancesResource(self._http)
115
116    @cached_property
117    def records(self) -> AdminRecordsResource:
118        return AdminRecordsResource(self._http)
119
120    @cached_property
121    def nodes(self) -> AdminNodesResource:
122        return AdminNodesResource(self._http)
123
124    @cached_property
125    def observability(self) -> AdminObservabilityResource:
126        return AdminObservabilityResource(self._http)
127
128    @cached_property
129    def settings(self) -> AdminSettingsResource:
130        return AdminSettingsResource(self._http)
131
132    @cached_property
133    def competition_admins(self) -> AdminCompetitionAdminsResource:
134        return AdminCompetitionAdminsResource(self._http)
135
136    @cached_property
137    def competition_invites(self) -> AdminCompetitionInvitesResource:
138        return AdminCompetitionInvitesResource(self._http)
139
140    @cached_property
141    def tasks(self) -> AdminTasksResource:
142        return AdminTasksResource(self._http)
143
144    @cached_property
145    def scheduled_jobs(self) -> AdminScheduledJobsResource:
146        return AdminScheduledJobsResource(self._http)
147
148    def close(self) -> None:
149        """Close the underlying transport.
150
151        Only call this on a standalone client built via :meth:`connect`;
152        when reached via ``PlatformClient.admin`` the transport is shared
153        with — and closed by — the parent player client.
154        """
155        self._http.close()
156
157    def __enter__(self) -> Self:
158        return self
159
160    def __exit__(
161        self,
162        exc_type: type[BaseException] | None,
163        exc: BaseException | None,
164        tb: TracebackType | None,
165    ) -> None:
166        self.close()
class AdminClient: 
 60class AdminClient:
 61    """Admin / operator surface, grouped into resource namespaces.
 62
 63    Namespaces: :attr:`users`, :attr:`competitions`, :attr:`announcements`,
 64    :attr:`achievements`, :attr:`challenges`, :attr:`instances`,
 65    :attr:`records`, :attr:`nodes`, :attr:`observability`, :attr:`settings`,
 66    :attr:`competition_admins`.
 67    """
 68
 69    def __init__(self, http: BaseHttpClient) -> None:
 70        #: Shared transport (the parent PlatformClient when reached via
 71        #: ``client.admin``, or an owned client when built via ``connect``).
 72        self._http = http
 73
 74    @classmethod
 75    def connect(
 76        cls,
 77        server_url: str,
 78        token: str = "",
 79        *,
 80        max_retries: int = 3,
 81        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 82    ) -> AdminClient:
 83        """Build an ``AdminClient`` that owns its own transport.
 84
 85        For operator tooling that doesn't already hold a
 86        :class:`PlatformClient`. The returned client is a context manager
 87        that closes its transport on exit.
 88        """
 89        base = server_url.rstrip("/")
 90        http = BaseHttpClient(f"{base}/api/v1", token, timeout=timeout, max_retries=max_retries)
 91        return cls(http)
 92
 93    @cached_property
 94    def users(self) -> AdminUsersResource:
 95        return AdminUsersResource(self._http)
 96
 97    @cached_property
 98    def competitions(self) -> AdminCompetitionsResource:
 99        return AdminCompetitionsResource(self._http)
100
101    @cached_property
102    def announcements(self) -> AdminAnnouncementsResource:
103        return AdminAnnouncementsResource(self._http)
104
105    @cached_property
106    def achievements(self) -> AdminAchievementsResource:
107        return AdminAchievementsResource(self._http)
108
109    @cached_property
110    def challenges(self) -> AdminChallengesResource:
111        return AdminChallengesResource(self._http)
112
113    @cached_property
114    def instances(self) -> AdminInstancesResource:
115        return AdminInstancesResource(self._http)
116
117    @cached_property
118    def records(self) -> AdminRecordsResource:
119        return AdminRecordsResource(self._http)
120
121    @cached_property
122    def nodes(self) -> AdminNodesResource:
123        return AdminNodesResource(self._http)
124
125    @cached_property
126    def observability(self) -> AdminObservabilityResource:
127        return AdminObservabilityResource(self._http)
128
129    @cached_property
130    def settings(self) -> AdminSettingsResource:
131        return AdminSettingsResource(self._http)
132
133    @cached_property
134    def competition_admins(self) -> AdminCompetitionAdminsResource:
135        return AdminCompetitionAdminsResource(self._http)
136
137    @cached_property
138    def competition_invites(self) -> AdminCompetitionInvitesResource:
139        return AdminCompetitionInvitesResource(self._http)
140
141    @cached_property
142    def tasks(self) -> AdminTasksResource:
143        return AdminTasksResource(self._http)
144
145    @cached_property
146    def scheduled_jobs(self) -> AdminScheduledJobsResource:
147        return AdminScheduledJobsResource(self._http)
148
149    def close(self) -> None:
150        """Close the underlying transport.
151
152        Only call this on a standalone client built via :meth:`connect`;
153        when reached via ``PlatformClient.admin`` the transport is shared
154        with — and closed by — the parent player client.
155        """
156        self._http.close()
157
158    def __enter__(self) -> Self:
159        return self
160
161    def __exit__(
162        self,
163        exc_type: type[BaseException] | None,
164        exc: BaseException | None,
165        tb: TracebackType | None,
166    ) -> None:
167        self.close()

Admin / operator surface, grouped into resource namespaces.

Namespaces: users, competitions, announcements, achievements, challenges, instances, records, nodes, observability, settings, competition_admins.

AdminClient(http: ctfy.sdk.base.BaseHttpClient) 
69    def __init__(self, http: BaseHttpClient) -> None:
70        #: Shared transport (the parent PlatformClient when reached via
71        #: ``client.admin``, or an owned client when built via ``connect``).
72        self._http = http
@classmethod
def connect( cls, server_url: str, token: str = '', *, max_retries: int = 3, timeout: int = 600) -> AdminClient: 
74    @classmethod
75    def connect(
76        cls,
77        server_url: str,
78        token: str = "",
79        *,
80        max_retries: int = 3,
81        timeout: int = DEFAULT_CLIENT_TIMEOUT,
82    ) -> AdminClient:
83        """Build an ``AdminClient`` that owns its own transport.
84
85        For operator tooling that doesn't already hold a
86        :class:`PlatformClient`. The returned client is a context manager
87        that closes its transport on exit.
88        """
89        base = server_url.rstrip("/")
90        http = BaseHttpClient(f"{base}/api/v1", token, timeout=timeout, max_retries=max_retries)
91        return cls(http)

Build an AdminClient that owns its own transport.

For operator tooling that doesn't already hold a PlatformClient. The returned client is a context manager that closes its transport on exit.

users: ctfy.sdk.admin_resources.users.AdminUsersResource 
93    @cached_property
94    def users(self) -> AdminUsersResource:
95        return AdminUsersResource(self._http)
competitions: ctfy.sdk.admin_resources.competitions.AdminCompetitionsResource 
97    @cached_property
98    def competitions(self) -> AdminCompetitionsResource:
99        return AdminCompetitionsResource(self._http)
announcements: ctfy.sdk.admin_resources.announcements.AdminAnnouncementsResource 
101    @cached_property
102    def announcements(self) -> AdminAnnouncementsResource:
103        return AdminAnnouncementsResource(self._http)
achievements: ctfy.sdk.admin_resources.achievements.AdminAchievementsResource 
105    @cached_property
106    def achievements(self) -> AdminAchievementsResource:
107        return AdminAchievementsResource(self._http)
challenges: ctfy.sdk.admin_resources.challenges.AdminChallengesResource 
109    @cached_property
110    def challenges(self) -> AdminChallengesResource:
111        return AdminChallengesResource(self._http)
instances: ctfy.sdk.admin_resources.instances.AdminInstancesResource 
113    @cached_property
114    def instances(self) -> AdminInstancesResource:
115        return AdminInstancesResource(self._http)
records: ctfy.sdk.admin_resources.records.AdminRecordsResource 
117    @cached_property
118    def records(self) -> AdminRecordsResource:
119        return AdminRecordsResource(self._http)
nodes: ctfy.sdk.admin_resources.nodes.AdminNodesResource 
121    @cached_property
122    def nodes(self) -> AdminNodesResource:
123        return AdminNodesResource(self._http)
observability: ctfy.sdk.admin_resources.observability.AdminObservabilityResource 
125    @cached_property
126    def observability(self) -> AdminObservabilityResource:
127        return AdminObservabilityResource(self._http)
settings: ctfy.sdk.admin_resources.settings.AdminSettingsResource 
129    @cached_property
130    def settings(self) -> AdminSettingsResource:
131        return AdminSettingsResource(self._http)
competition_admins: ctfy.sdk.admin_resources.competition_admins.AdminCompetitionAdminsResource 
133    @cached_property
134    def competition_admins(self) -> AdminCompetitionAdminsResource:
135        return AdminCompetitionAdminsResource(self._http)
competition_invites: ctfy.sdk.admin_resources.competition_invites.AdminCompetitionInvitesResource 
137    @cached_property
138    def competition_invites(self) -> AdminCompetitionInvitesResource:
139        return AdminCompetitionInvitesResource(self._http)
tasks: ctfy.sdk.admin_resources.tasks.AdminTasksResource 
141    @cached_property
142    def tasks(self) -> AdminTasksResource:
143        return AdminTasksResource(self._http)
scheduled_jobs: ctfy.sdk.admin_resources.tasks.AdminScheduledJobsResource 
145    @cached_property
146    def scheduled_jobs(self) -> AdminScheduledJobsResource:
147        return AdminScheduledJobsResource(self._http)
def close(self) -> None: 
149    def close(self) -> None:
150        """Close the underlying transport.
151
152        Only call this on a standalone client built via :meth:`connect`;
153        when reached via ``PlatformClient.admin`` the transport is shared
154        with — and closed by — the parent player client.
155        """
156        self._http.close()

Close the underlying transport.

Only call this on a standalone client built via connect(); when reached via PlatformClient.admin the transport is shared with — and closed by — the parent player client.