Custom clients

The SDK talks to Campus through two abstract interfaces in campus_python.json_client. The default implementation is CampusRequest (requests-based, with trace propagation and the optional 401 auto-refresh hook), but any host can bring its own transport — the interfaces were designed to wrap flask.testing.FlaskClient as much as requests.

The JsonClient contract

A client must implement:

• the six verbs — get, post, put, put_data (raw, non-JSON body — file/hosted-app uploads), delete, patch — each returning a JsonResponse;
• authorization control: the authorization property plus set_basic_authorization(), set_bearer_authorization(), and reset_authorization() (the session managers swap tokens through these);
• a base_url.

campus_python.json_client.interface.JsonClient

Bases: ABC

This class describes the public interface required from Client classes, which are used to send JSON requests.

Optional capability, not part of this abstract interface: concrete clients may support a 401 auto-refresh hook (CampusRequest.set_unauthorized_hook, issue #89) that refreshes the bearer token and retries a request once when it comes back 401. Callers must feature-detect (hasattr) rather than assume it.

Source code in campus_python/json_client/interface.py

class JsonClient(ABC):
    """This class describes the public interface required from Client
    classes, which are used to send JSON requests.

    Optional capability, not part of this abstract interface: concrete
    clients may support a 401 auto-refresh hook
    (CampusRequest.set_unauthorized_hook, issue #89) that refreshes the
    bearer token and retries a request once when it comes back 401.
    Callers must feature-detect (hasattr) rather than assume it.
    """
    base_url: str
    # pylint: disable=unnecessary-ellipsis

    @abstractmethod
    def __init__(
            self,
            base_url: str | None = None,
            *,
            auth: Iterable[str] | str | None = None,
            headers: Mapping[str, str] | None = None,
            **kwargs: Any
    ): ...

    @abstractmethod
    def reset_authorization(self) -> None:
        """Reset authorization back to client credentials from env."""

    @property
    @abstractmethod
    def authorization(self) -> str | None:
        """The current Authorization header value, or None when unset."""

    @abstractmethod
    def set_basic_authorization(self, client_id: str, secret: str) -> None:
        """Set Basic Authorization header using a pre-encoded token.

        Args:
            client_id (str): Client ID.
            secret (str): Client Secret.
        """

    @abstractmethod
    def set_bearer_authorization(self, token: str) -> None:
        """Set Bearer Authorization header.

        Args:
            token (str): Bearer token.
        """

    def set_default_header(self, name: str, value: str) -> None:
        """Set a default header sent with every request (#837).

        Optional capability: the ABC default refuses, so callers can
        detect a client that cannot carry default headers instead of
        silently dropping them. Implementations should set the header
        on their underlying transport (e.g. the requests session).
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support default headers"
        )

    @abstractmethod
    def get(self: Self, path: str, query: JsonDict | None = None) -> JsonResponse:
        """Sends a GET request."""

    @abstractmethod
    def post(self: Self, path: str, json: JsonDict | None = None) -> JsonResponse:
        """Sends a POST request."""

    @abstractmethod
    def put(
            self: Self,
            path: str,
            json: JsonDict | None = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a PUT request."""

    @abstractmethod
    def put_data(
            self: Self,
            path: str,
            data: bytes | str,
            content_type: str | None = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a PUT request with a raw (non-JSON) body.

        The body is sent verbatim, with no JSON encoding; the
        Content-Type header is caller-controlled. Used by file
        uploads (campus.api /files by-slug), which carry raw bytes
        with an explicit Content-Type. `query` carries URL query
        parameters (e.g. the files upsert's optional display name).
        """

    @abstractmethod
    def post_data(
            self: Self,
            path: str,
            data: bytes | str,
            content_type: str | None = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a POST request with a raw (non-JSON) body.

        Symmetric with put_data(): the body is sent verbatim, with no
        JSON encoding; the Content-Type header is caller-controlled.
        Used by raw-content create routes that mint server-side ids
        (campus.api timetable attachment uploads), which carry raw
        bytes with an explicit Content-Type. `query` carries URL
        query parameters (e.g. the attachment's original filename).
        """

    @abstractmethod
    def delete(
            self: Self,
            path: str,
            json: JsonDict | None = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a DELETE request."""

    @abstractmethod
    def patch(
            self: Self,
            path: str,
            json: Any = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a PATCH request."""

base_url: str instance-attribute

authorization: str | None abstractmethod property

The current Authorization header value, or None when unset.

__init__(base_url: str | None = None, *, auth: Iterable[str] | str | None = None, headers: Mapping[str, str] | None = None, **kwargs: Any) abstractmethod

Source code in campus_python/json_client/interface.py

@abstractmethod
def __init__(
        self,
        base_url: str | None = None,
        *,
        auth: Iterable[str] | str | None = None,
        headers: Mapping[str, str] | None = None,
        **kwargs: Any
): ...

reset_authorization() -> None abstractmethod

Reset authorization back to client credentials from env.

Source code in campus_python/json_client/interface.py

@abstractmethod
def reset_authorization(self) -> None:
    """Reset authorization back to client credentials from env."""

set_basic_authorization(client_id: str, secret: str) -> None abstractmethod

Set Basic Authorization header using a pre-encoded token.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID.	required
secret	str	Client Secret.	required

Source code in campus_python/json_client/interface.py

@abstractmethod
def set_basic_authorization(self, client_id: str, secret: str) -> None:
    """Set Basic Authorization header using a pre-encoded token.

    Args:
        client_id (str): Client ID.
        secret (str): Client Secret.
    """

set_bearer_authorization(token: str) -> None abstractmethod

Set Bearer Authorization header.

Parameters:

Name	Type	Description	Default
token	str	Bearer token.	required

Source code in campus_python/json_client/interface.py

@abstractmethod
def set_bearer_authorization(self, token: str) -> None:
    """Set Bearer Authorization header.

    Args:
        token (str): Bearer token.
    """

set_default_header(name: str, value: str) -> None

Set a default header sent with every request (#837).

Optional capability: the ABC default refuses, so callers can detect a client that cannot carry default headers instead of silently dropping them. Implementations should set the header on their underlying transport (e.g. the requests session).

Source code in campus_python/json_client/interface.py

def set_default_header(self, name: str, value: str) -> None:
    """Set a default header sent with every request (#837).

    Optional capability: the ABC default refuses, so callers can
    detect a client that cannot carry default headers instead of
    silently dropping them. Implementations should set the header
    on their underlying transport (e.g. the requests session).
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support default headers"
    )

get(path: str, query: JsonDict | None = None) -> JsonResponse abstractmethod

Sends a GET request.

Source code in campus_python/json_client/interface.py

@abstractmethod
def get(self: Self, path: str, query: JsonDict | None = None) -> JsonResponse:
    """Sends a GET request."""

post(path: str, json: JsonDict | None = None) -> JsonResponse abstractmethod

Sends a POST request.

Source code in campus_python/json_client/interface.py

@abstractmethod
def post(self: Self, path: str, json: JsonDict | None = None) -> JsonResponse:
    """Sends a POST request."""

put(path: str, json: JsonDict | None = None, query: JsonDict | None = None) -> JsonResponse abstractmethod

Sends a PUT request.

Source code in campus_python/json_client/interface.py

@abstractmethod
def put(
        self: Self,
        path: str,
        json: JsonDict | None = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a PUT request."""

put_data(path: str, data: bytes | str, content_type: str | None = None, query: JsonDict | None = None) -> JsonResponse abstractmethod

Sends a PUT request with a raw (non-JSON) body.

The body is sent verbatim, with no JSON encoding; the Content-Type header is caller-controlled. Used by file uploads (campus.api /files by-slug), which carry raw bytes with an explicit Content-Type. query carries URL query parameters (e.g. the files upsert's optional display name).

Source code in campus_python/json_client/interface.py

@abstractmethod
def put_data(
        self: Self,
        path: str,
        data: bytes | str,
        content_type: str | None = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a PUT request with a raw (non-JSON) body.

    The body is sent verbatim, with no JSON encoding; the
    Content-Type header is caller-controlled. Used by file
    uploads (campus.api /files by-slug), which carry raw bytes
    with an explicit Content-Type. `query` carries URL query
    parameters (e.g. the files upsert's optional display name).
    """

post_data(path: str, data: bytes | str, content_type: str | None = None, query: JsonDict | None = None) -> JsonResponse abstractmethod

Sends a POST request with a raw (non-JSON) body.

Symmetric with put_data(): the body is sent verbatim, with no JSON encoding; the Content-Type header is caller-controlled. Used by raw-content create routes that mint server-side ids (campus.api timetable attachment uploads), which carry raw bytes with an explicit Content-Type. query carries URL query parameters (e.g. the attachment's original filename).

Source code in campus_python/json_client/interface.py

@abstractmethod
def post_data(
        self: Self,
        path: str,
        data: bytes | str,
        content_type: str | None = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a POST request with a raw (non-JSON) body.

    Symmetric with put_data(): the body is sent verbatim, with no
    JSON encoding; the Content-Type header is caller-controlled.
    Used by raw-content create routes that mint server-side ids
    (campus.api timetable attachment uploads), which carry raw
    bytes with an explicit Content-Type. `query` carries URL
    query parameters (e.g. the attachment's original filename).
    """

delete(path: str, json: JsonDict | None = None, query: JsonDict | None = None) -> JsonResponse abstractmethod

Sends a DELETE request.

Source code in campus_python/json_client/interface.py

@abstractmethod
def delete(
        self: Self,
        path: str,
        json: JsonDict | None = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a DELETE request."""

patch(path: str, json: Any = None, query: JsonDict | None = None) -> JsonResponse abstractmethod

Sends a PATCH request.

Source code in campus_python/json_client/interface.py

@abstractmethod
def patch(
        self: Self,
        path: str,
        json: Any = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a PATCH request."""

The JsonResponse contract

Responses expose status_code, headers, text, content (bytes — file content is binary; callers decode), and json(). Implement raise_for_status() by delegating to the base-class mapping: it parses the Campus error envelope (and the legacy error_code form) into the right APIError subclass, recovering OAuth errors from AUTH_* codes when production strips envelope details.

campus_python.json_client.interface.JsonResponse

Bases: ABC

This class describes the public interface required from Response wrappers for JSON responses.

Source code in campus_python/json_client/interface.py

class JsonResponse(ABC):
    """This class describes the public interface required from Response
    wrappers for JSON responses.
    """
    # pylint: disable=unnecessary-ellipsis

    @abstractmethod
    def __init__(self): ...

    @property
    @abstractmethod
    def status_code(self) -> int:
        """HTTP status code of the response."""
        ...

    @property
    @abstractmethod
    def headers(self) -> dict[str, str]:
        """Returns headers of the response as a dict."""
        ...

    @property
    @abstractmethod
    def text(self) -> str:
        """Returns the response body as a string."""
        ...

    @property
    @abstractmethod
    def content(self) -> bytes:
        """Returns the response body as bytes, undecoded.

        File content (campus.api /files) is binary: callers decode,
        never this property.
        """
        ...

    @abstractmethod
    def json(self) -> Any:
        """Returns the response body as JSON."""
        ...

    def ok(self) -> bool:
        """Returns True if the response status code is 2xx, False otherwise."""
        return 200 <= self.status_code < 300

    def is_client_error(self) -> bool:
        """Returns True if the response status code is 4xx, False otherwise."""
        return 400 <= self.status_code < 500

    def is_server_error(self) -> bool:
        """Returns True if the response status code is 5xx, False otherwise."""
        return 500 <= self.status_code < 600

    def raise_for_status(self) -> None:
        """Raise the mapped APIError subclass on a 4xx/5xx status.

        Parses the Campus error envelope (and the legacy error_code
        form) into the registered error class for the status code;
        OAuth errors are recovered from AUTH_* codes when the server
        strips envelope details. Response headers and body ride along
        as error notes.
        """
        if not (self.is_client_error() or self.is_server_error()):
            return

        status_code = self.status_code

        # Try to extract a useful message/body
        try:
            response_data = self.json()
        except Exception:
            response_data = None

        try:
            body = response_data if response_data is not None else self.text
        except Exception:
            body = f"HTTP {status_code}"

        error = errors.APIError.with_status_code(status_code, response_data)
        if error:
            # Attach notes for debugging (headers, body)
            # Try the notes attribute first (new APIError)
            try:
                error.notes = {"headers": dict(self.headers), "body": body}
            except AttributeError:
                # Fall back to add_note for older implementations
                try:
                    error.add_note(f"Headers: {self.headers}")
                    error.add_note(f"Body: {body}")
                except Exception:
                    # add_note may not exist; ignore if it doesn't
                    pass
            raise error from None

status_code: int abstractmethod property

HTTP status code of the response.

headers: dict[str, str] abstractmethod property

Returns headers of the response as a dict.

text: str abstractmethod property

Returns the response body as a string.

content: bytes abstractmethod property

Returns the response body as bytes, undecoded.

File content (campus.api /files) is binary: callers decode, never this property.

__init__() abstractmethod

Source code in campus_python/json_client/interface.py

@abstractmethod
def __init__(self): ...

json() -> Any abstractmethod

Returns the response body as JSON.

Source code in campus_python/json_client/interface.py

@abstractmethod
def json(self) -> Any:
    """Returns the response body as JSON."""
    ...

ok() -> bool

Returns True if the response status code is 2xx, False otherwise.

Source code in campus_python/json_client/interface.py

def ok(self) -> bool:
    """Returns True if the response status code is 2xx, False otherwise."""
    return 200 <= self.status_code < 300

is_client_error() -> bool

Returns True if the response status code is 4xx, False otherwise.

Source code in campus_python/json_client/interface.py

def is_client_error(self) -> bool:
    """Returns True if the response status code is 4xx, False otherwise."""
    return 400 <= self.status_code < 500

is_server_error() -> bool

Returns True if the response status code is 5xx, False otherwise.

Source code in campus_python/json_client/interface.py

def is_server_error(self) -> bool:
    """Returns True if the response status code is 5xx, False otherwise."""
    return 500 <= self.status_code < 600

raise_for_status() -> None

Raise the mapped APIError subclass on a 4xx/5xx status.

Parses the Campus error envelope (and the legacy error_code form) into the registered error class for the status code; OAuth errors are recovered from AUTH_* codes when the server strips envelope details. Response headers and body ride along as error notes.

Source code in campus_python/json_client/interface.py

def raise_for_status(self) -> None:
    """Raise the mapped APIError subclass on a 4xx/5xx status.

    Parses the Campus error envelope (and the legacy error_code
    form) into the registered error class for the status code;
    OAuth errors are recovered from AUTH_* codes when the server
    strips envelope details. Response headers and body ride along
    as error notes.
    """
    if not (self.is_client_error() or self.is_server_error()):
        return

    status_code = self.status_code

    # Try to extract a useful message/body
    try:
        response_data = self.json()
    except Exception:
        response_data = None

    try:
        body = response_data if response_data is not None else self.text
    except Exception:
        body = f"HTTP {status_code}"

    error = errors.APIError.with_status_code(status_code, response_data)
    if error:
        # Attach notes for debugging (headers, body)
        # Try the notes attribute first (new APIError)
        try:
            error.notes = {"headers": dict(self.headers), "body": body}
        except AttributeError:
            # Fall back to add_note for older implementations
            try:
                error.add_note(f"Headers: {self.headers}")
                error.add_note(f"Body: {body}")
            except Exception:
                # add_note may not exist; ignore if it doesn't
                pass
        raise error from None

Optional capabilities

Two capabilities are deliberately not part of the ABC. Feature-detect them instead of assuming:

• Default headers — client.set_default_header(name, value) (#837). The ABC default raises NotImplementedError, so a client that cannot carry default headers refuses loudly rather than dropping them.
• 401 auto-refresh hook — client.set_unauthorized_hook(fn) (#89). Present on CampusRequest (and used by with_user_session(), which feature-detects via isinstance). The hook runs at most once per request: return the new bearer token to retry with, or None to surface the original 401.

The default transport

CampusRequest is the requests-based client the facade builds; its optional capabilities (set_default_header, set_unauthorized_hook) are what custom clients may mirror.

campus_python.json_client.CampusRequest

Bases: JsonClient

Campus JSON client

This client is pre-configured to handle JSON requests and responses. It provides convenience methods for common HTTP methods with JSON payloads.

Note that the client requires authorization before it can make requests. Use CampusRequest.set_basic_authorization(), CampusRequest.set_bearer_authorization(), or CampusRequest.for_user() to authorize the client, and CampusRequest.reset_authorization() to reset to client credentials

Source code in campus_python/json_client/__init__.py

class CampusRequest(JsonClient):
    """Campus JSON client

    This client is pre-configured to handle JSON requests and responses.
    It provides convenience methods for common HTTP methods with JSON
    payloads.

    Note that the client requires authorization before it can make
    requests. Use CampusRequest.set_basic_authorization(),
    CampusRequest.set_bearer_authorization(), or
    CampusRequest.for_user() to authorize the client, and
    CampusRequest.reset_authorization() to reset to client credentials
    """

    def __init__(
            self,
            base_url: str | None,
            *,
            mode: str = "server",
            headers: Mapping[str, str] | None = None,
            **kwargs: Any,
    ):
        self.base_url = base_url or ""
        self._headers = dict(headers or {})
        # allow optional default timeout via kwargs
        self._timeout = kwargs.get("timeout", 10)
        # Optional 401 auto-refresh hook (issue #89); see
        # set_unauthorized_hook(). Off by default.
        self._unauthorized_hook: Callable[[], str | None] | None = None
        self._in_unauthorized_hook = False
        # Session to persist headers and connection pooling
        self._session = requests.Session()
        self._session.headers.update(self._headers)
        # Propagate campus trace context on calls made while the host app
        # handles a traced request, so receiving campus services record
        # them as child spans (campus#816). No-op outside a request
        # context (startup, background threads).
        from .. import tracing

        tracing.instrument_requests_session(self._session)
        # Only set client credentials in server mode
        # Device mode starts without auth (Bearer token will be set later)
        if mode == "server":
            self.reset_authorization()

    @property
    def headers(self) -> HttpHeader:
        """Get the currently configured headers."""
        headers = cast(Mapping[str, str], self._session.headers)
        return HttpHeader.from_header(headers)

    def _build_url(self, path: str) -> str:
        if not self.base_url:  # relative URL
            return f"http://localhost:{env.get('PORT', '8080')}" + path
        if self.base_url.endswith("/") and path.startswith("/"):
            return self.base_url[:-1] + path
        if not self.base_url.endswith("/") and not path.startswith("/"):
            return self.base_url + "/" + path
        return self.base_url + path

    def reset_authorization(self) -> None:
        """Reset authorization back to client credentials from env."""
        env.require("CLIENT_ID", "CLIENT_SECRET")
        self.set_basic_authorization(
            client_id=env.CLIENT_ID,
            secret=env.CLIENT_SECRET
        )

    @property
    def authorization(self) -> str | None:
        """The current Authorization header value, or None when unset."""
        return self._session.headers.get("Authorization")

    def set_basic_authorization(self, client_id: str, secret: str) -> None:
        """Set Basic Authorization header using a pre-encoded token.

        Args:
            client_id (str): Client ID.
            secret (str): Client Secret.
        """
        credentials = f"{client_id}:{secret}"
        # Encode credentials in base64 following RFC 7617
        encoded_credentials = base64.b64encode(
            credentials.encode('utf-8')
        ).decode('ascii')
        self._session.headers["Authorization"] = "Basic " + encoded_credentials

    def set_bearer_authorization(self, token: str) -> None:
        """Set Bearer Authorization header.

        Args:
            token (str): Bearer token.
        """
        self._session.headers["Authorization"] = "Bearer " + token

    def set_default_header(self, name: str, value: str) -> None:
        """Set a default header sent with every request from this client.

        Non-browser clients use this to present stable identity on API
        calls, e.g. X-Campus-Device (#837) for device attribution of
        spans. Calls made inside a host request context may still add
        per-request headers on top (tracing.propagation_headers).
        """
        self._session.headers[name] = value

    def set_unauthorized_hook(
            self,
            hook: Callable[[], str | None] | None,
    ) -> None:
        """Install a 401 auto-refresh hook (issue #89).

        When a request comes back 401, the hook is invoked once; if it
        returns a bearer token, the Authorization header is refreshed
        and the request is retried once with it. A second 401 on the
        retry is returned to the caller as-is, and a hook returning
        None (refresh failed) leaves the original 401 response in
        place — failures surface exactly as they would without the
        hook.

        The hook runs at most once per request, and never re-enters
        itself: requests made from inside the hook skip the hook, so a
        hook that calls back into this client cannot recurse.

        Args:
            hook: Callable returning the new bearer token, or None to
                signal refresh failure. None clears the hook.
        """
        self._unauthorized_hook = hook

    def _send(self, method: str, url: str, **kwargs: Any) -> JsonResponse:
        """Send a request, with one 401-triggered retry (issue #89).

        The retry is safe even for non-idempotent requests: Campus
        services validate the bearer in a before_request authenticator
        before dispatching, so a 401 response is produced by the auth
        layer before any handler runs — there is no partial work to
        replay.
        """
        try:
            resp = self._session.request(
                method, url, timeout=self._timeout, **kwargs
            )
            if (
                    resp.status_code == 401
                    and self._unauthorized_hook is not None
                    and not self._in_unauthorized_hook
            ):
                self._in_unauthorized_hook = True
                try:
                    token = self._unauthorized_hook()
                finally:
                    self._in_unauthorized_hook = False
                if token:
                    self.set_bearer_authorization(token)
                    resp = self._session.request(
                        method, url, timeout=self._timeout, **kwargs
                    )
        except requests.RequestException as exc:
            raise errors.ServerError(error_description=str(exc)) from None
        return CampusResponse(resp)

    def get(
            self: Self,
            path: str,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a GET request."""
        url = self._build_url(path)
        if query:
            return self._send("GET", url, params=query)
        return self._send("GET", url)

    def post(self: Self, path: str, json: JsonDict | None = None) -> JsonResponse:
        """Sends a POST request."""
        url = self._build_url(path)
        return self._send("POST", url, json=json)

    def put(
            self: Self,
            path: str,
            json: JsonDict | None = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a PUT request."""
        url = self._build_url(path)
        return self._send("PUT", url, json=json, params=query)

    def put_data(
            self: Self,
            path: str,
            data: bytes | str,
            content_type: str | None = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a PUT request with a raw body (e.g. file bytes).

        Unlike put(), the body is sent verbatim with no JSON encoding;
        Content-Type is caller-controlled (requests' own default
        applies when omitted).

        Args:
            path: Request path.
            data: Raw request body.
            content_type: Content-Type header for the body, or None to
                leave it to the transport default.
            query: Optional URL query parameters.
        """
        url = self._build_url(path)
        headers = {"Content-Type": content_type} if content_type else None
        return self._send("PUT", url, data=data, headers=headers,
                          params=query)

    def post_data(
            self: Self,
            path: str,
            data: bytes | str,
            content_type: str | None = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a POST request with a raw body (e.g. file bytes).

        Unlike post(), the body is sent verbatim with no JSON encoding;
        Content-Type is caller-controlled (requests' own default
        applies when omitted).

        Args:
            path: Request path.
            data: Raw request body.
            content_type: Content-Type header for the body, or None to
                leave it to the transport default.
            query: Optional URL query parameters.
        """
        url = self._build_url(path)
        headers = {"Content-Type": content_type} if content_type else None
        return self._send("POST", url, data=data, headers=headers,
                          params=query)

    def delete(
            self: Self,
            path: str,
            json: JsonDict | None = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a DELETE request."""
        url = self._build_url(path)
        return self._send("DELETE", url, json=json, params=query)

    def patch(
            self: Self,
            path: str,
            json: Any = None,
            query: JsonDict | None = None
    ) -> JsonResponse:
        """Sends a PATCH request."""
        url = self._build_url(path)
        return self._send("PATCH", url, json=json, params=query)

base_url = base_url or '' instance-attribute

headers: HttpHeader property

Get the currently configured headers.

authorization: str | None property

The current Authorization header value, or None when unset.

__init__(base_url: str | None, *, mode: str = 'server', headers: Mapping[str, str] | None = None, **kwargs: Any)

Source code in campus_python/json_client/__init__.py

def __init__(
        self,
        base_url: str | None,
        *,
        mode: str = "server",
        headers: Mapping[str, str] | None = None,
        **kwargs: Any,
):
    self.base_url = base_url or ""
    self._headers = dict(headers or {})
    # allow optional default timeout via kwargs
    self._timeout = kwargs.get("timeout", 10)
    # Optional 401 auto-refresh hook (issue #89); see
    # set_unauthorized_hook(). Off by default.
    self._unauthorized_hook: Callable[[], str | None] | None = None
    self._in_unauthorized_hook = False
    # Session to persist headers and connection pooling
    self._session = requests.Session()
    self._session.headers.update(self._headers)
    # Propagate campus trace context on calls made while the host app
    # handles a traced request, so receiving campus services record
    # them as child spans (campus#816). No-op outside a request
    # context (startup, background threads).
    from .. import tracing

    tracing.instrument_requests_session(self._session)
    # Only set client credentials in server mode
    # Device mode starts without auth (Bearer token will be set later)
    if mode == "server":
        self.reset_authorization()

reset_authorization() -> None

Reset authorization back to client credentials from env.

Source code in campus_python/json_client/__init__.py

def reset_authorization(self) -> None:
    """Reset authorization back to client credentials from env."""
    env.require("CLIENT_ID", "CLIENT_SECRET")
    self.set_basic_authorization(
        client_id=env.CLIENT_ID,
        secret=env.CLIENT_SECRET
    )

set_basic_authorization(client_id: str, secret: str) -> None

Set Basic Authorization header using a pre-encoded token.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID.	required
secret	str	Client Secret.	required

Source code in campus_python/json_client/__init__.py

def set_basic_authorization(self, client_id: str, secret: str) -> None:
    """Set Basic Authorization header using a pre-encoded token.

    Args:
        client_id (str): Client ID.
        secret (str): Client Secret.
    """
    credentials = f"{client_id}:{secret}"
    # Encode credentials in base64 following RFC 7617
    encoded_credentials = base64.b64encode(
        credentials.encode('utf-8')
    ).decode('ascii')
    self._session.headers["Authorization"] = "Basic " + encoded_credentials

set_bearer_authorization(token: str) -> None

Set Bearer Authorization header.

Parameters:

Name	Type	Description	Default
token	str	Bearer token.	required

Source code in campus_python/json_client/__init__.py

def set_bearer_authorization(self, token: str) -> None:
    """Set Bearer Authorization header.

    Args:
        token (str): Bearer token.
    """
    self._session.headers["Authorization"] = "Bearer " + token

set_default_header(name: str, value: str) -> None

Set a default header sent with every request from this client.

Non-browser clients use this to present stable identity on API calls, e.g. X-Campus-Device (#837) for device attribution of spans. Calls made inside a host request context may still add per-request headers on top (tracing.propagation_headers).

Source code in campus_python/json_client/__init__.py

def set_default_header(self, name: str, value: str) -> None:
    """Set a default header sent with every request from this client.

    Non-browser clients use this to present stable identity on API
    calls, e.g. X-Campus-Device (#837) for device attribution of
    spans. Calls made inside a host request context may still add
    per-request headers on top (tracing.propagation_headers).
    """
    self._session.headers[name] = value

set_unauthorized_hook(hook: Callable[[], str | None] | None) -> None

Install a 401 auto-refresh hook (issue #89).

When a request comes back 401, the hook is invoked once; if it returns a bearer token, the Authorization header is refreshed and the request is retried once with it. A second 401 on the retry is returned to the caller as-is, and a hook returning None (refresh failed) leaves the original 401 response in place — failures surface exactly as they would without the hook.

The hook runs at most once per request, and never re-enters itself: requests made from inside the hook skip the hook, so a hook that calls back into this client cannot recurse.

Parameters:

Name	Type	Description	Default
hook	Callable[[], str | None] | None	Callable returning the new bearer token, or None to signal refresh failure. None clears the hook.	required

Source code in campus_python/json_client/__init__.py

def set_unauthorized_hook(
        self,
        hook: Callable[[], str | None] | None,
) -> None:
    """Install a 401 auto-refresh hook (issue #89).

    When a request comes back 401, the hook is invoked once; if it
    returns a bearer token, the Authorization header is refreshed
    and the request is retried once with it. A second 401 on the
    retry is returned to the caller as-is, and a hook returning
    None (refresh failed) leaves the original 401 response in
    place — failures surface exactly as they would without the
    hook.

    The hook runs at most once per request, and never re-enters
    itself: requests made from inside the hook skip the hook, so a
    hook that calls back into this client cannot recurse.

    Args:
        hook: Callable returning the new bearer token, or None to
            signal refresh failure. None clears the hook.
    """
    self._unauthorized_hook = hook

get(path: str, query: JsonDict | None = None) -> JsonResponse

Sends a GET request.

Source code in campus_python/json_client/__init__.py

def get(
        self: Self,
        path: str,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a GET request."""
    url = self._build_url(path)
    if query:
        return self._send("GET", url, params=query)
    return self._send("GET", url)

post(path: str, json: JsonDict | None = None) -> JsonResponse

Sends a POST request.

Source code in campus_python/json_client/__init__.py

def post(self: Self, path: str, json: JsonDict | None = None) -> JsonResponse:
    """Sends a POST request."""
    url = self._build_url(path)
    return self._send("POST", url, json=json)

put(path: str, json: JsonDict | None = None, query: JsonDict | None = None) -> JsonResponse

Sends a PUT request.

Source code in campus_python/json_client/__init__.py

def put(
        self: Self,
        path: str,
        json: JsonDict | None = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a PUT request."""
    url = self._build_url(path)
    return self._send("PUT", url, json=json, params=query)

put_data(path: str, data: bytes | str, content_type: str | None = None, query: JsonDict | None = None) -> JsonResponse

Sends a PUT request with a raw body (e.g. file bytes).

Unlike put(), the body is sent verbatim with no JSON encoding; Content-Type is caller-controlled (requests' own default applies when omitted).

Parameters:

Name	Type	Description	Default
path	str	Request path.	required
data	bytes | str	Raw request body.	required
content_type	str | None	Content-Type header for the body, or None to leave it to the transport default.	None
query	JsonDict | None	Optional URL query parameters.	None

Source code in campus_python/json_client/__init__.py

def put_data(
        self: Self,
        path: str,
        data: bytes | str,
        content_type: str | None = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a PUT request with a raw body (e.g. file bytes).

    Unlike put(), the body is sent verbatim with no JSON encoding;
    Content-Type is caller-controlled (requests' own default
    applies when omitted).

    Args:
        path: Request path.
        data: Raw request body.
        content_type: Content-Type header for the body, or None to
            leave it to the transport default.
        query: Optional URL query parameters.
    """
    url = self._build_url(path)
    headers = {"Content-Type": content_type} if content_type else None
    return self._send("PUT", url, data=data, headers=headers,
                      params=query)

post_data(path: str, data: bytes | str, content_type: str | None = None, query: JsonDict | None = None) -> JsonResponse

Sends a POST request with a raw body (e.g. file bytes).

Unlike post(), the body is sent verbatim with no JSON encoding; Content-Type is caller-controlled (requests' own default applies when omitted).

Parameters:

Name	Type	Description	Default
path	str	Request path.	required
data	bytes | str	Raw request body.	required
content_type	str | None	Content-Type header for the body, or None to leave it to the transport default.	None
query	JsonDict | None	Optional URL query parameters.	None

Source code in campus_python/json_client/__init__.py

def post_data(
        self: Self,
        path: str,
        data: bytes | str,
        content_type: str | None = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a POST request with a raw body (e.g. file bytes).

    Unlike post(), the body is sent verbatim with no JSON encoding;
    Content-Type is caller-controlled (requests' own default
    applies when omitted).

    Args:
        path: Request path.
        data: Raw request body.
        content_type: Content-Type header for the body, or None to
            leave it to the transport default.
        query: Optional URL query parameters.
    """
    url = self._build_url(path)
    headers = {"Content-Type": content_type} if content_type else None
    return self._send("POST", url, data=data, headers=headers,
                      params=query)

delete(path: str, json: JsonDict | None = None, query: JsonDict | None = None) -> JsonResponse

Sends a DELETE request.

Source code in campus_python/json_client/__init__.py

def delete(
        self: Self,
        path: str,
        json: JsonDict | None = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a DELETE request."""
    url = self._build_url(path)
    return self._send("DELETE", url, json=json, params=query)

patch(path: str, json: Any = None, query: JsonDict | None = None) -> JsonResponse

Sends a PATCH request.

Source code in campus_python/json_client/__init__.py

def patch(
        self: Self,
        path: str,
        json: Any = None,
        query: JsonDict | None = None
) -> JsonResponse:
    """Sends a PATCH request."""
    url = self._build_url(path)
    return self._send("PATCH", url, json=json, params=query)

Building one

The interfaces exist so hosts can point the SDK at an in-process WSGI app (unit/integration tests without a network) or an exotic transport. The shape:

from campus_python.json_client.interface import JsonClient, JsonResponse

class FlaskTestClientResponse(JsonResponse):
    def __init__(self, response):
        self._response = response          # werkzeug.test.TestResponse
    # implement status_code / headers / text / content / json()
    # and reuse JsonResponse.raise_for_status (do not override)

class FlaskTestClient(JsonClient):
    def __init__(self, test_client, base_url="https://campus.test"):
        self.base_url = base_url
        self._client = test_client
    # implement the six verbs as test_client.open(...) calls,
    # plus the authorization methods

Then construct the service roots directly against it:

from campus_python.api.v1 import ApiRoot

api = ApiRoot(json_client=FlaskTestClient(flask_app.test_client()))
api.assignments.list()

The repo's own test suite stays one level simpler: resource unit tests drive Mock clients through the shared factories in tests/support.py (make_api, make_auth, json_response), pinning recorded calls rather than running a real WSGI stack.