Skip to content

Audit (campus.audit)

The /audit/v1 service: trace ingestion and queries, plus management of the audit service's own API keys. The audit root authenticates with an audit_v1_... key (AUDIT_API_KEY) as its Bearer — client-secret Basic auth is rejected by the service.

Ingestion feeds the client-side circuit breaker before raising: a 429 trips the bucket (Retry-After honored), 2xx clears expired trips, and a breaker failure never breaks ingestion. See Tracing for how spans attribute to journeys.

campus.model.* types referenced here are documented in the campus repo.

Traces

campus_python.audit.v1.traces.Traces

Bases: ResourceCollection

Campus Audit traces resource.

Source code in campus_python/audit/v1/traces.py
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class Traces(ResourceCollection):
    """Campus Audit traces resource."""
    path = "traces"

    def __getitem__(self, trace_id: str) -> "Traces.Trace":
        """Get a specific trace resource by ID."""
        return Traces.Trace(trace_id, parent=self)

    def ingest(self, spans: "list[dict[str, Any]]") -> JsonDict:
        """Ingest a batch of trace spans.

        Args:
            spans: Span dicts shaped like the server's TraceSpan
                resource (trace_id, span_id, method, path, ...)

        Returns:
            201 full success: {"created": [...]}; 207 partial failure
            adds {"failed": [...]} with per-span statuses.
        """
        resp = self.client.post(self.make_path(), json={"spans": spans})
        # Feed the ingest circuit breaker before raising: 429 trips the
        # bucket (Retry-After + tripped key from the body), 2xx clears
        # expired trips (#831). The error still raises as normal, and a
        # breaker failure must never break ingestion.
        with contextlib.suppress(Exception):
            ratelimit.breaker.observe(resp.status_code, resp.headers, _safe_json(resp))
        resp.raise_for_status()
        return resp.json()

    def list(
            self,
            *,
            since: "str | None" = None,
            until: "str | None" = None,
            limit: "int | None" = None,
            cursor: "str | None" = None,
    ) -> JsonDict:
        """List recent traces, newest first.

        Args:
            since: ISO 8601 lower bound
            until: ISO 8601 upper bound
            limit: Page size (server clamps to its max)
            cursor: Opaque token from a previous page's cursor.next

        Returns:
            {"traces": [...], "cursor": {"next": ..., "has_more": bool}}
        """
        query: JsonDict = {}
        if since is not None:
            query["since"] = since
        if until is not None:
            query["until"] = until
        if limit is not None:
            query["limit"] = limit
        if cursor is not None:
            query["cursor"] = cursor
        return self._get_json(self.make_path(), query=query or None)

    def search(
            self,
            *,
            path: "str | None" = None,
            status: "int | str | None" = None,
            api_key_id: "str | None" = None,
            client_id: "str | None" = None,
            user_id: "str | None" = None,
            journey_id: "str | None" = None,
            since: "str | None" = None,
            until: "str | None" = None,
            limit: "int | None" = None,
            cursor: "str | None" = None,
    ) -> JsonDict:
        """Filter and search traces.

        Args:
            path: Filter by endpoint path
            status: Filter by HTTP status code
            api_key_id: Filter by audit API key
            client_id: Filter by OAuth client
            user_id: Filter by user
            journey_id: Filter by login-journey tag (campus#803)
            since: ISO 8601 lower bound (as in list())
            until: ISO 8601 upper bound (as in list())
            limit: Page size (as in list())
            cursor: Page token (as in list())

        Returns:
            {"traces": [...], "cursor": {"next": ..., "has_more": bool}}
        """
        query: JsonDict = {}
        if path is not None:
            query["path"] = path
        if status is not None:
            query["status"] = status
        if api_key_id is not None:
            query["api_key_id"] = api_key_id
        if client_id is not None:
            query["client_id"] = client_id
        if user_id is not None:
            query["user_id"] = user_id
        if journey_id is not None:
            query["journey_id"] = journey_id
        if since is not None:
            query["since"] = since
        if until is not None:
            query["until"] = until
        if limit is not None:
            query["limit"] = limit
        if cursor is not None:
            query["cursor"] = cursor
        return self._get_json(
            self.make_path("search", end_slash=False), query=query or None
        )

    class Trace(Resource):
        """Single campus audit trace resource."""

        @property
        def spans(self) -> "Traces.Trace.Spans":
            """Get the spans resource for this trace."""
            return Traces.Trace.Spans("spans", parent=self)

        def get(self) -> JsonDict:
            """Get the full trace tree.

            Returns:
                {"trace_id": ..., "root_span": <nested span tree>}
            """
            return self._get_json(self.make_path(end_slash=True))

        class Spans(Resource):
            """Spans of a single trace."""

            def __getitem__(self, span_id: str) -> "Traces.Trace.Span":
                """Get a specific span resource by ID."""
                return Traces.Trace.Span(span_id, parent=self)

            def list(self) -> "list[JsonDict]":
                """List all spans in this trace (flat)."""
                return self._get_json(self.make_path(end_slash=True))["spans"]

        class Span(Resource):
            """Single span of a trace, with full request/response detail."""

            def get(self) -> JsonDict:
                """Get this span with full request/response detail."""
                return self._get_json(self.make_path(end_slash=True))

path = 'traces' class-attribute instance-attribute

Trace

Bases: Resource

Single campus audit trace resource.

Source code in campus_python/audit/v1/traces.py
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class Trace(Resource):
    """Single campus audit trace resource."""

    @property
    def spans(self) -> "Traces.Trace.Spans":
        """Get the spans resource for this trace."""
        return Traces.Trace.Spans("spans", parent=self)

    def get(self) -> JsonDict:
        """Get the full trace tree.

        Returns:
            {"trace_id": ..., "root_span": <nested span tree>}
        """
        return self._get_json(self.make_path(end_slash=True))

    class Spans(Resource):
        """Spans of a single trace."""

        def __getitem__(self, span_id: str) -> "Traces.Trace.Span":
            """Get a specific span resource by ID."""
            return Traces.Trace.Span(span_id, parent=self)

        def list(self) -> "list[JsonDict]":
            """List all spans in this trace (flat)."""
            return self._get_json(self.make_path(end_slash=True))["spans"]

    class Span(Resource):
        """Single span of a trace, with full request/response detail."""

        def get(self) -> JsonDict:
            """Get this span with full request/response detail."""
            return self._get_json(self.make_path(end_slash=True))

spans: Traces.Trace.Spans property

Get the spans resource for this trace.

Spans

Bases: Resource

Spans of a single trace.

Source code in campus_python/audit/v1/traces.py
155
156
157
158
159
160
161
162
163
164
class Spans(Resource):
    """Spans of a single trace."""

    def __getitem__(self, span_id: str) -> "Traces.Trace.Span":
        """Get a specific span resource by ID."""
        return Traces.Trace.Span(span_id, parent=self)

    def list(self) -> "list[JsonDict]":
        """List all spans in this trace (flat)."""
        return self._get_json(self.make_path(end_slash=True))["spans"]
__getitem__(span_id: str) -> Traces.Trace.Span

Get a specific span resource by ID.

Source code in campus_python/audit/v1/traces.py
158
159
160
def __getitem__(self, span_id: str) -> "Traces.Trace.Span":
    """Get a specific span resource by ID."""
    return Traces.Trace.Span(span_id, parent=self)
list() -> list[JsonDict]

List all spans in this trace (flat).

Source code in campus_python/audit/v1/traces.py
162
163
164
def list(self) -> "list[JsonDict]":
    """List all spans in this trace (flat)."""
    return self._get_json(self.make_path(end_slash=True))["spans"]

Span

Bases: Resource

Single span of a trace, with full request/response detail.

Source code in campus_python/audit/v1/traces.py
166
167
168
169
170
171
class Span(Resource):
    """Single span of a trace, with full request/response detail."""

    def get(self) -> JsonDict:
        """Get this span with full request/response detail."""
        return self._get_json(self.make_path(end_slash=True))
get() -> JsonDict

Get this span with full request/response detail.

Source code in campus_python/audit/v1/traces.py
169
170
171
def get(self) -> JsonDict:
    """Get this span with full request/response detail."""
    return self._get_json(self.make_path(end_slash=True))

get() -> JsonDict

Get the full trace tree.

Returns:

Type Description
JsonDict

{"trace_id": ..., "root_span": }

Source code in campus_python/audit/v1/traces.py
147
148
149
150
151
152
153
def get(self) -> JsonDict:
    """Get the full trace tree.

    Returns:
        {"trace_id": ..., "root_span": <nested span tree>}
    """
    return self._get_json(self.make_path(end_slash=True))

__getitem__(trace_id: str) -> Traces.Trace

Get a specific trace resource by ID.

Source code in campus_python/audit/v1/traces.py
28
29
30
def __getitem__(self, trace_id: str) -> "Traces.Trace":
    """Get a specific trace resource by ID."""
    return Traces.Trace(trace_id, parent=self)

ingest(spans: list[dict[str, Any]]) -> JsonDict

Ingest a batch of trace spans.

Parameters:

Name Type Description Default
spans list[dict[str, Any]]

Span dicts shaped like the server's TraceSpan resource (trace_id, span_id, method, path, ...)

required

Returns:

Type Description
JsonDict

201 full success: {"created": [...]}; 207 partial failure

JsonDict

adds {"failed": [...]} with per-span statuses.

Source code in campus_python/audit/v1/traces.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def ingest(self, spans: "list[dict[str, Any]]") -> JsonDict:
    """Ingest a batch of trace spans.

    Args:
        spans: Span dicts shaped like the server's TraceSpan
            resource (trace_id, span_id, method, path, ...)

    Returns:
        201 full success: {"created": [...]}; 207 partial failure
        adds {"failed": [...]} with per-span statuses.
    """
    resp = self.client.post(self.make_path(), json={"spans": spans})
    # Feed the ingest circuit breaker before raising: 429 trips the
    # bucket (Retry-After + tripped key from the body), 2xx clears
    # expired trips (#831). The error still raises as normal, and a
    # breaker failure must never break ingestion.
    with contextlib.suppress(Exception):
        ratelimit.breaker.observe(resp.status_code, resp.headers, _safe_json(resp))
    resp.raise_for_status()
    return resp.json()

list(*, since: str | None = None, until: str | None = None, limit: int | None = None, cursor: str | None = None) -> JsonDict

List recent traces, newest first.

Parameters:

Name Type Description Default
since str | None

ISO 8601 lower bound

None
until str | None

ISO 8601 upper bound

None
limit int | None

Page size (server clamps to its max)

None
cursor str | None

Opaque token from a previous page's cursor.next

None

Returns:

Type Description
JsonDict

{"traces": [...], "cursor": {"next": ..., "has_more": bool}}

Source code in campus_python/audit/v1/traces.py
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def list(
        self,
        *,
        since: "str | None" = None,
        until: "str | None" = None,
        limit: "int | None" = None,
        cursor: "str | None" = None,
) -> JsonDict:
    """List recent traces, newest first.

    Args:
        since: ISO 8601 lower bound
        until: ISO 8601 upper bound
        limit: Page size (server clamps to its max)
        cursor: Opaque token from a previous page's cursor.next

    Returns:
        {"traces": [...], "cursor": {"next": ..., "has_more": bool}}
    """
    query: JsonDict = {}
    if since is not None:
        query["since"] = since
    if until is not None:
        query["until"] = until
    if limit is not None:
        query["limit"] = limit
    if cursor is not None:
        query["cursor"] = cursor
    return self._get_json(self.make_path(), query=query or None)

search(*, path: str | None = None, status: int | str | None = None, api_key_id: str | None = None, client_id: str | None = None, user_id: str | None = None, journey_id: str | None = None, since: str | None = None, until: str | None = None, limit: int | None = None, cursor: str | None = None) -> JsonDict

Filter and search traces.

Parameters:

Name Type Description Default
path str | None

Filter by endpoint path

None
status int | str | None

Filter by HTTP status code

None
api_key_id str | None

Filter by audit API key

None
client_id str | None

Filter by OAuth client

None
user_id str | None

Filter by user

None
journey_id str | None

Filter by login-journey tag (campus#803)

None
since str | None

ISO 8601 lower bound (as in list())

None
until str | None

ISO 8601 upper bound (as in list())

None
limit int | None

Page size (as in list())

None
cursor str | None

Page token (as in list())

None

Returns:

Type Description
JsonDict

{"traces": [...], "cursor": {"next": ..., "has_more": bool}}

Source code in campus_python/audit/v1/traces.py
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def search(
        self,
        *,
        path: "str | None" = None,
        status: "int | str | None" = None,
        api_key_id: "str | None" = None,
        client_id: "str | None" = None,
        user_id: "str | None" = None,
        journey_id: "str | None" = None,
        since: "str | None" = None,
        until: "str | None" = None,
        limit: "int | None" = None,
        cursor: "str | None" = None,
) -> JsonDict:
    """Filter and search traces.

    Args:
        path: Filter by endpoint path
        status: Filter by HTTP status code
        api_key_id: Filter by audit API key
        client_id: Filter by OAuth client
        user_id: Filter by user
        journey_id: Filter by login-journey tag (campus#803)
        since: ISO 8601 lower bound (as in list())
        until: ISO 8601 upper bound (as in list())
        limit: Page size (as in list())
        cursor: Page token (as in list())

    Returns:
        {"traces": [...], "cursor": {"next": ..., "has_more": bool}}
    """
    query: JsonDict = {}
    if path is not None:
        query["path"] = path
    if status is not None:
        query["status"] = status
    if api_key_id is not None:
        query["api_key_id"] = api_key_id
    if client_id is not None:
        query["client_id"] = client_id
    if user_id is not None:
        query["user_id"] = user_id
    if journey_id is not None:
        query["journey_id"] = journey_id
    if since is not None:
        query["since"] = since
    if until is not None:
        query["until"] = until
    if limit is not None:
        query["limit"] = limit
    if cursor is not None:
        query["cursor"] = cursor
    return self._get_json(
        self.make_path("search", end_slash=False), query=query or None
    )

APIKeys

campus_python.audit.v1.apikeys.APIKeys

Bases: ResourceCollection

Campus Audit API keys resource.

Every write takes a privilege set: either a scope list or a named role the service resolves to one.

Source code in campus_python/audit/v1/apikeys.py
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class APIKeys(ResourceCollection):
    """Campus Audit API keys resource.

    Every write takes a privilege set: either a scope list or a named
    role the service resolves to one.
    """
    path = "apikeys"

    def __getitem__(self, api_key_id: str) -> "APIKeys.APIKey":
        """Get a specific API key resource by ID."""
        return APIKeys.APIKey(api_key_id, parent=self)

    def new(
            self,
            *,
            name: str,
            owner_id: str,
            scopes: "list[str] | None" = None,
            role: "str | None" = None,
            rate_limit: "int | None" = None,
            expires_at: "str | None" = None,
    ) -> JsonDict:
        """Create an API key.

        Give either `scopes` or a named `role` — the service resolves a
        role to exactly one scope list. The two are mutually exclusive
        server-side: sending both is refused (422) rather than one
        silently winning, so this method does not pre-judge them. An
        unknown role is likewise refused, with the server naming the
        roles that exist — the catalogue is the service's, so it is not
        duplicated here.

        Args:
            name: Key name
            owner_id: Owner user ID
            scopes: Scope strings the key grants (e.g. ["traces:read"])
            role: A named role the service resolves to a scope list
            rate_limit: Requests per minute (optional)
            expires_at: ISO 8601 expiry (optional)

        Returns:
            The created key record; its plaintext `api_key` value is
            shown only here.
        """
        return self._post_json(self.make_path(), json=payload(
            name=name,
            owner_id=owner_id,
            scopes=scopes,
            role=role,
            rate_limit=rate_limit,
            expires_at=expires_at,
        ))

    def list(
            self,
            *,
            owner_id: "str | None" = None,
            active_only: bool = True,
            limit: "int | None" = None,
    ) -> JsonDict:
        """List API keys with optional filtering.

        Args:
            owner_id: Filter by owner user ID
            active_only: Only non-expired, non-revoked keys (default True)
            limit: Page size (server default 50)

        Returns:
            {"api_keys": [...], "count": int}
        """
        query: JsonDict = {}
        if owner_id is not None:
            query["owner_id"] = owner_id
        # The server parses bool query values from the lowercase
        # literals true/false (flask_campus _BOOL_LITERALS); Python's
        # str(True) capitalization would 422.
        query["active_only"] = str(active_only).lower()
        if limit is not None:
            query["limit"] = limit
        return self._get_json(self.make_path(), query=query)

    class APIKey(Resource):
        """Single campus audit API key resource.

        update() takes a privilege set too: a scope list or a named
        role.
        """

        def get(self) -> JsonDict:
            """Get this API key's record (no key hash or plaintext)."""
            return self._get_json(self.make_path(end_slash=True))

        def update(
                self,
                *,
                name: "str | None" = None,
                scopes: "list[str] | None" = None,
                role: "str | None" = None,
                rate_limit: "int | None" = None,
        ) -> JsonDict:
            """Update this API key's mutable fields.

            Only name, scopes, and rate_limit are mutable; use
            revoke() to disable the key. The server rejects an empty
            update, so at least one field must be provided.

            A `role` re-scopes the key to the scope list the service
            resolves that role to. It is mutually exclusive with
            `scopes` server-side: sending both is refused (422) rather
            than one silently winning, so this method does not
            pre-judge them.

            Returns:
                The updated key record
            """
            return self._patch_json(self.make_path(end_slash=True), json=updates(
                name=name,
                scopes=scopes,
                role=role,
                rate_limit=rate_limit,
            ))

        def revoke(self) -> None:
            """Revoke this API key (the record is kept for audit)."""
            self._delete_json(self.make_path(end_slash=True))
            return None

        def regenerate(self) -> str:
            """Regenerate this API key's value.

            The old value stops working immediately.

            Returns:
                The new plaintext key value, shown only here.
            """
            return self._post_json(
                self.make_path("regenerate", end_slash=False)
            )["key"]

path = 'apikeys' class-attribute instance-attribute

APIKey

Bases: Resource

Single campus audit API key resource.

update() takes a privilege set too: a scope list or a named role.

Source code in campus_python/audit/v1/apikeys.py
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class APIKey(Resource):
    """Single campus audit API key resource.

    update() takes a privilege set too: a scope list or a named
    role.
    """

    def get(self) -> JsonDict:
        """Get this API key's record (no key hash or plaintext)."""
        return self._get_json(self.make_path(end_slash=True))

    def update(
            self,
            *,
            name: "str | None" = None,
            scopes: "list[str] | None" = None,
            role: "str | None" = None,
            rate_limit: "int | None" = None,
    ) -> JsonDict:
        """Update this API key's mutable fields.

        Only name, scopes, and rate_limit are mutable; use
        revoke() to disable the key. The server rejects an empty
        update, so at least one field must be provided.

        A `role` re-scopes the key to the scope list the service
        resolves that role to. It is mutually exclusive with
        `scopes` server-side: sending both is refused (422) rather
        than one silently winning, so this method does not
        pre-judge them.

        Returns:
            The updated key record
        """
        return self._patch_json(self.make_path(end_slash=True), json=updates(
            name=name,
            scopes=scopes,
            role=role,
            rate_limit=rate_limit,
        ))

    def revoke(self) -> None:
        """Revoke this API key (the record is kept for audit)."""
        self._delete_json(self.make_path(end_slash=True))
        return None

    def regenerate(self) -> str:
        """Regenerate this API key's value.

        The old value stops working immediately.

        Returns:
            The new plaintext key value, shown only here.
        """
        return self._post_json(
            self.make_path("regenerate", end_slash=False)
        )["key"]

get() -> JsonDict

Get this API key's record (no key hash or plaintext).

Source code in campus_python/audit/v1/apikeys.py
117
118
119
def get(self) -> JsonDict:
    """Get this API key's record (no key hash or plaintext)."""
    return self._get_json(self.make_path(end_slash=True))

update(*, name: str | None = None, scopes: list[str] | None = None, role: str | None = None, rate_limit: int | None = None) -> JsonDict

Update this API key's mutable fields.

Only name, scopes, and rate_limit are mutable; use revoke() to disable the key. The server rejects an empty update, so at least one field must be provided.

A role re-scopes the key to the scope list the service resolves that role to. It is mutually exclusive with scopes server-side: sending both is refused (422) rather than one silently winning, so this method does not pre-judge them.

Returns:

Type Description
JsonDict

The updated key record

Source code in campus_python/audit/v1/apikeys.py
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def update(
        self,
        *,
        name: "str | None" = None,
        scopes: "list[str] | None" = None,
        role: "str | None" = None,
        rate_limit: "int | None" = None,
) -> JsonDict:
    """Update this API key's mutable fields.

    Only name, scopes, and rate_limit are mutable; use
    revoke() to disable the key. The server rejects an empty
    update, so at least one field must be provided.

    A `role` re-scopes the key to the scope list the service
    resolves that role to. It is mutually exclusive with
    `scopes` server-side: sending both is refused (422) rather
    than one silently winning, so this method does not
    pre-judge them.

    Returns:
        The updated key record
    """
    return self._patch_json(self.make_path(end_slash=True), json=updates(
        name=name,
        scopes=scopes,
        role=role,
        rate_limit=rate_limit,
    ))

revoke() -> None

Revoke this API key (the record is kept for audit).

Source code in campus_python/audit/v1/apikeys.py
151
152
153
154
def revoke(self) -> None:
    """Revoke this API key (the record is kept for audit)."""
    self._delete_json(self.make_path(end_slash=True))
    return None

regenerate() -> str

Regenerate this API key's value.

The old value stops working immediately.

Returns:

Type Description
str

The new plaintext key value, shown only here.

Source code in campus_python/audit/v1/apikeys.py
156
157
158
159
160
161
162
163
164
165
166
def regenerate(self) -> str:
    """Regenerate this API key's value.

    The old value stops working immediately.

    Returns:
        The new plaintext key value, shown only here.
    """
    return self._post_json(
        self.make_path("regenerate", end_slash=False)
    )["key"]

__getitem__(api_key_id: str) -> APIKeys.APIKey

Get a specific API key resource by ID.

Source code in campus_python/audit/v1/apikeys.py
37
38
39
def __getitem__(self, api_key_id: str) -> "APIKeys.APIKey":
    """Get a specific API key resource by ID."""
    return APIKeys.APIKey(api_key_id, parent=self)

new(*, name: str, owner_id: str, scopes: list[str] | None = None, role: str | None = None, rate_limit: int | None = None, expires_at: str | None = None) -> JsonDict

Create an API key.

Give either scopes or a named role — the service resolves a role to exactly one scope list. The two are mutually exclusive server-side: sending both is refused (422) rather than one silently winning, so this method does not pre-judge them. An unknown role is likewise refused, with the server naming the roles that exist — the catalogue is the service's, so it is not duplicated here.

Parameters:

Name Type Description Default
name str

Key name

required
owner_id str

Owner user ID

required
scopes list[str] | None

Scope strings the key grants (e.g. ["traces:read"])

None
role str | None

A named role the service resolves to a scope list

None
rate_limit int | None

Requests per minute (optional)

None
expires_at str | None

ISO 8601 expiry (optional)

None

Returns:

Type Description
JsonDict

The created key record; its plaintext api_key value is

JsonDict

shown only here.

Source code in campus_python/audit/v1/apikeys.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def new(
        self,
        *,
        name: str,
        owner_id: str,
        scopes: "list[str] | None" = None,
        role: "str | None" = None,
        rate_limit: "int | None" = None,
        expires_at: "str | None" = None,
) -> JsonDict:
    """Create an API key.

    Give either `scopes` or a named `role` — the service resolves a
    role to exactly one scope list. The two are mutually exclusive
    server-side: sending both is refused (422) rather than one
    silently winning, so this method does not pre-judge them. An
    unknown role is likewise refused, with the server naming the
    roles that exist — the catalogue is the service's, so it is not
    duplicated here.

    Args:
        name: Key name
        owner_id: Owner user ID
        scopes: Scope strings the key grants (e.g. ["traces:read"])
        role: A named role the service resolves to a scope list
        rate_limit: Requests per minute (optional)
        expires_at: ISO 8601 expiry (optional)

    Returns:
        The created key record; its plaintext `api_key` value is
        shown only here.
    """
    return self._post_json(self.make_path(), json=payload(
        name=name,
        owner_id=owner_id,
        scopes=scopes,
        role=role,
        rate_limit=rate_limit,
        expires_at=expires_at,
    ))

list(*, owner_id: str | None = None, active_only: bool = True, limit: int | None = None) -> JsonDict

List API keys with optional filtering.

Parameters:

Name Type Description Default
owner_id str | None

Filter by owner user ID

None
active_only bool

Only non-expired, non-revoked keys (default True)

True
limit int | None

Page size (server default 50)

None

Returns:

Type Description
JsonDict

{"api_keys": [...], "count": int}

Source code in campus_python/audit/v1/apikeys.py
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def list(
        self,
        *,
        owner_id: "str | None" = None,
        active_only: bool = True,
        limit: "int | None" = None,
) -> JsonDict:
    """List API keys with optional filtering.

    Args:
        owner_id: Filter by owner user ID
        active_only: Only non-expired, non-revoked keys (default True)
        limit: Page size (server default 50)

    Returns:
        {"api_keys": [...], "count": int}
    """
    query: JsonDict = {}
    if owner_id is not None:
        query["owner_id"] = owner_id
    # The server parses bool query values from the lowercase
    # literals true/false (flask_campus _BOOL_LITERALS); Python's
    # str(True) capitalization would 422.
    query["active_only"] = str(active_only).lower()
    if limit is not None:
        query["limit"] = limit
    return self._get_json(self.make_path(), query=query)

AuditRoot

campus_python.audit.v1.AuditRoot

Bases: ResourceRoot

Campus Audit service resource.

Source code in campus_python/audit/v1/__init__.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class AuditRoot(ResourceRoot):
    """Campus Audit service resource."""
    url_prefix = "/audit/v1"

    def __init__(self, json_client: JsonClient):
        super().__init__(json_client=json_client)
        self._apikeys = None
        self._traces = None

    @property
    def apikeys(self) -> apikeys.APIKeys:
        """Get the API keys resource."""
        if not self._apikeys:
            self._apikeys = apikeys.APIKeys(root=self)
        return self._apikeys

    @property
    def traces(self) -> traces.Traces:
        """Get the traces resource."""
        if not self._traces:
            self._traces = traces.Traces(root=self)
        return self._traces

url_prefix = '/audit/v1' class-attribute instance-attribute

apikeys: apikeys.APIKeys property

Get the API keys resource.

traces: traces.Traces property

Get the traces resource.

__init__(json_client: JsonClient)

Source code in campus_python/audit/v1/__init__.py
20
21
22
23
def __init__(self, json_client: JsonClient):
    super().__init__(json_client=json_client)
    self._apikeys = None
    self._traces = None