ctfy.sdk.resources.activities

client.activities — platform activity log + histogram.

 1"""``client.activities`` — platform activity log + histogram."""
 2
 3from __future__ import annotations
 4
 5import builtins
 6from typing import Any
 7
 8from ctfy.sdk._helpers import _raise_for_status
 9from ctfy.sdk.base import BaseHttpClient
10from ctfy.server.models import Activity, ActivityTimeseries
11
12
13class ActivitiesResource:
14    """The public activity feed and its bucketed timeseries."""
15
16    def __init__(self, http: BaseHttpClient) -> None:
17        self._http = http
18
19    def list(
20        self,
21        team_id: str = "",
22        challenge_id: str = "",
23        event: str = "",
24        offset: int = 0,
25        limit: int = 100,
26    ) -> builtins.list[Activity]:
27        """Get platform activity log."""
28        params: dict[str, Any] = {"offset": offset, "limit": limit}
29        if team_id:
30            params["team_id"] = team_id
31        if challenge_id:
32            params["challenge_id"] = challenge_id
33        if event:
34            params["event"] = event
35        resp = self._http.request("GET", "/activities", params=params)
36        _raise_for_status(resp)
37        return [Activity.model_validate(a) for a in resp.json()["items"]]
38
39    def timeseries(
40        self,
41        *,
42        team_id: str = "",
43        challenge_id: str = "",
44        event: str = "",
45        window: int = 86400,
46        bucket: int = 3600,
47    ) -> ActivityTimeseries:
48        """Bucketed activity-event counts (the histogram on the activity
49        page). ``window`` + ``bucket`` are in seconds."""
50        params: dict[str, Any] = {"window": window, "bucket": bucket}
51        if team_id:
52            params["team_id"] = team_id
53        if challenge_id:
54            params["challenge_id"] = challenge_id
55        if event:
56            params["event"] = event
57        resp = self._http.request("GET", "/activities/timeseries", params=params)
58        _raise_for_status(resp)
59        return ActivityTimeseries.model_validate(resp.json())
class ActivitiesResource: 
14class ActivitiesResource:
15    """The public activity feed and its bucketed timeseries."""
16
17    def __init__(self, http: BaseHttpClient) -> None:
18        self._http = http
19
20    def list(
21        self,
22        team_id: str = "",
23        challenge_id: str = "",
24        event: str = "",
25        offset: int = 0,
26        limit: int = 100,
27    ) -> builtins.list[Activity]:
28        """Get platform activity log."""
29        params: dict[str, Any] = {"offset": offset, "limit": limit}
30        if team_id:
31            params["team_id"] = team_id
32        if challenge_id:
33            params["challenge_id"] = challenge_id
34        if event:
35            params["event"] = event
36        resp = self._http.request("GET", "/activities", params=params)
37        _raise_for_status(resp)
38        return [Activity.model_validate(a) for a in resp.json()["items"]]
39
40    def timeseries(
41        self,
42        *,
43        team_id: str = "",
44        challenge_id: str = "",
45        event: str = "",
46        window: int = 86400,
47        bucket: int = 3600,
48    ) -> ActivityTimeseries:
49        """Bucketed activity-event counts (the histogram on the activity
50        page). ``window`` + ``bucket`` are in seconds."""
51        params: dict[str, Any] = {"window": window, "bucket": bucket}
52        if team_id:
53            params["team_id"] = team_id
54        if challenge_id:
55            params["challenge_id"] = challenge_id
56        if event:
57            params["event"] = event
58        resp = self._http.request("GET", "/activities/timeseries", params=params)
59        _raise_for_status(resp)
60        return ActivityTimeseries.model_validate(resp.json())

The public activity feed and its bucketed timeseries.

ActivitiesResource(http: ctfy.sdk.base.BaseHttpClient) 
17    def __init__(self, http: BaseHttpClient) -> None:
18        self._http = http
def list( self, team_id: str = '', challenge_id: str = '', event: str = '', offset: int = 0, limit: int = 100) -> list[ctfy.server.models.Activity]: 
20    def list(
21        self,
22        team_id: str = "",
23        challenge_id: str = "",
24        event: str = "",
25        offset: int = 0,
26        limit: int = 100,
27    ) -> builtins.list[Activity]:
28        """Get platform activity log."""
29        params: dict[str, Any] = {"offset": offset, "limit": limit}
30        if team_id:
31            params["team_id"] = team_id
32        if challenge_id:
33            params["challenge_id"] = challenge_id
34        if event:
35            params["event"] = event
36        resp = self._http.request("GET", "/activities", params=params)
37        _raise_for_status(resp)
38        return [Activity.model_validate(a) for a in resp.json()["items"]]

Get platform activity log.

def timeseries( self, *, team_id: str = '', challenge_id: str = '', event: str = '', window: int = 86400, bucket: int = 3600) -> ctfy.server.models.ActivityTimeseries: 
40    def timeseries(
41        self,
42        *,
43        team_id: str = "",
44        challenge_id: str = "",
45        event: str = "",
46        window: int = 86400,
47        bucket: int = 3600,
48    ) -> ActivityTimeseries:
49        """Bucketed activity-event counts (the histogram on the activity
50        page). ``window`` + ``bucket`` are in seconds."""
51        params: dict[str, Any] = {"window": window, "bucket": bucket}
52        if team_id:
53            params["team_id"] = team_id
54        if challenge_id:
55            params["challenge_id"] = challenge_id
56        if event:
57            params["event"] = event
58        resp = self._http.request("GET", "/activities/timeseries", params=params)
59        _raise_for_status(resp)
60        return ActivityTimeseries.model_validate(resp.json())

Bucketed activity-event counts (the histogram on the activity page). window + bucket are in seconds.