Python API reference

Everything exported from the top-level csiapps Python package. For the R package, see the R reference (pkgdown); for how the two map onto each other, see the parity checklist.

Configuration

set_institute

set_institute(institute: str = 'csipacific') -> None

Set the target institute for all subsequent API calls.

The institute determines the base host every request is sent to and which logo the Shiny chrome renders. It is process-wide global state (a direct mirror of the R package's package_state environment): there is only ever one configured institute per process, so set it once at startup.

Parameters:

Name	Type	Description	Default
institute	str	Which CSI institute to target. Must be one of "csipacific" or "csiontario". Defaults to "csipacific".	'csipacific'

Raises:

Type	Description
ValueError	If institute is not one of the two supported values.

Example

import csiapps

csiapps.set_institute("csiontario")

Note

This only selects the target host; it does not authenticate. See check_secrets for credential setup and is_sandbox_mode for whether requests actually reach the network.

Source code in csiapps/config.py

def set_institute(institute: str = "csipacific") -> None:
    """Set the target institute for all subsequent API calls.

    The institute determines the base host every request is sent to and which
    logo the Shiny chrome renders. It is process-wide global state (a direct
    mirror of the R package's ``package_state`` environment): there is only ever
    one configured institute per process, so set it once at startup.

    Args:
        institute: Which CSI institute to target. Must be one of
            ``"csipacific"`` or ``"csiontario"``. Defaults to ``"csipacific"``.

    Raises:
        ValueError: If ``institute`` is not one of the two supported values.

    Example:
        ```python
        import csiapps

        csiapps.set_institute("csiontario")
        ```

    Note:
        This only selects the target host; it does not authenticate. See
        [`check_secrets`][csiapps.auth.check_secrets] for credential setup and
        [`is_sandbox_mode`][csiapps.config.is_sandbox_mode] for whether requests
        actually reach the network.
    """
    if not (isinstance(institute, str) and institute in _VALID_INSTITUTES):
        raise ValueError(
            f"institute must be one of {_VALID_INSTITUTES!r}, got {institute!r}"
        )
    _state["institute"] = institute

is_sandbox_mode

is_sandbox_mode() -> bool

Report whether sandbox mode is currently enabled.

Sandbox mode is the fail-safe default: when it is on, the fetch_* and make_request helpers route to the local in-memory emulator in the csiapps.sandbox module instead of the network, so no request can reach production by accident. Every helper that hits the API consults this function when its own sandbox argument is left as None.

Resolution order (matching the R package):

1. An explicit override set via set_sandbox_mode always wins. Only the literal True enables it; any other set value is treated as False (mirrors R's isTRUE()).
2. Otherwise the CSIAPPS_ENV environment variable: only the literal "production" disables sandbox; any other non-empty value keeps it on (fail-safe against typos such as "prod").
3. Otherwise True.

Returns:

Name	Type	Description
bool	bool	True if sandbox mode is active (requests are emulated
	bool	locally), False if requests go to the live warehouse.

Example

import os
import csiapps

csiapps.set_sandbox_mode(None)              # use the environment
os.environ["CSIAPPS_ENV"] = "production"
csiapps.is_sandbox_mode()                   # -> False

Source code in csiapps/config.py

def is_sandbox_mode() -> bool:
    """Report whether sandbox mode is currently enabled.

    Sandbox mode is the fail-safe default: when it is on, the ``fetch_*`` and
    [`make_request`][csiapps.client.make_request] helpers route to the local
    in-memory emulator in the `csiapps.sandbox` module instead of the network, so no
    request can reach production by accident. Every helper that hits the API
    consults this function when its own ``sandbox`` argument is left as ``None``.

    Resolution order (matching the R package):

    1. An explicit override set via
       [`set_sandbox_mode`][csiapps.config.set_sandbox_mode] always wins. Only
       the literal ``True`` enables it; any other set value is treated as
       ``False`` (mirrors R's ``isTRUE()``).
    2. Otherwise the ``CSIAPPS_ENV`` environment variable: only the literal
       ``"production"`` disables sandbox; any other non-empty value keeps it on
       (fail-safe against typos such as ``"prod"``).
    3. Otherwise ``True``.

    Returns:
        bool: ``True`` if sandbox mode is active (requests are emulated
        locally), ``False`` if requests go to the live warehouse.

    Example:
        ```python
        import os
        import csiapps

        csiapps.set_sandbox_mode(None)              # use the environment
        os.environ["CSIAPPS_ENV"] = "production"
        csiapps.is_sandbox_mode()                   # -> False
        ```
    """
    override = _state["sandbox_override"]
    if override is not None:
        return override is True
    env = os.environ.get("CSIAPPS_ENV", "")
    if env:
        return env != "production"
    return True

set_sandbox_mode

set_sandbox_mode(enabled: bool | None) -> None

Force sandbox mode on or off, or clear the override.

The Python analog of R's options(csiapps.sandbox = ...). An override set here takes precedence over the CSIAPPS_ENV environment variable when is_sandbox_mode resolves the effective mode.

Parameters:

Name	Type	Description	Default
enabled	bool | None	True to force sandbox mode on, False to force it off, or None to clear the override and fall back to CSIAPPS_ENV.	required

Example

Pin sandbox mode on for a test, then restore environment-based resolution afterwards:

import csiapps

csiapps.set_sandbox_mode(True)
csiapps.is_sandbox_mode()   # -> True

csiapps.set_sandbox_mode(None)   # back to CSIAPPS_ENV

Note

Only the literal True enables sandbox mode; any other truthy value is treated as False (mirrors R's isTRUE()). See is_sandbox_mode for the full resolution order.

Source code in csiapps/config.py

def set_sandbox_mode(enabled: bool | None) -> None:
    """Force sandbox mode on or off, or clear the override.

    The Python analog of R's ``options(csiapps.sandbox = ...)``. An override set
    here takes precedence over the ``CSIAPPS_ENV`` environment variable when
    [`is_sandbox_mode`][csiapps.config.is_sandbox_mode] resolves the effective
    mode.

    Args:
        enabled: ``True`` to force sandbox mode on, ``False`` to force it off, or
            ``None`` to clear the override and fall back to ``CSIAPPS_ENV``.

    Example:
        Pin sandbox mode on for a test, then restore environment-based
        resolution afterwards:

        ```python
        import csiapps

        csiapps.set_sandbox_mode(True)
        csiapps.is_sandbox_mode()   # -> True

        csiapps.set_sandbox_mode(None)   # back to CSIAPPS_ENV
        ```

    Note:
        Only the literal ``True`` enables sandbox mode; any other truthy value is
        treated as ``False`` (mirrors R's ``isTRUE()``). See
        [`is_sandbox_mode`][csiapps.config.is_sandbox_mode] for the full
        resolution order.
    """
    _state["sandbox_override"] = enabled

Authentication

check_secrets

check_secrets(verbose: bool = False, sandbox: bool | None = None) -> bool

Validate that the environment is configured for authentication.

Call this once at app startup to fail fast on a misconfigured deployment. Behaviour depends on the mode:

• Sandbox mode: OAuth secret checks are skipped entirely (the sandbox simulates login and needs no client credentials). Instead the presence or absence of CSIAPPS_ACCESS_TOKEN is reported to stderr, and the function never raises.
• Production mode: the OAuth URLs derived from the configured institute plus CSIAPPS_REDIRECT_URI are checked for a well-formed http(s) scheme, and a :class:ValueError is raised listing any that are missing or malformed.

Parameters:

Name	Type	Description	Default
verbose	bool	If True, dump the resolved CSIAPPS environment (client id, auth/token/userinfo URLs, redirect URI, scope, and whether a client secret is set) to stderr. The client secret value itself is never printed — only whether one is present. Defaults to False.	False
sandbox	bool | None	Force the mode for this check. True or False overrides detection; None (the default) resolves via is_sandbox_mode.	None

Returns:

Name	Type	Description
bool	bool	Always True when the environment is usable. In production the
	bool	function raises rather than returning False, so a True return is
	bool	a positive assurance the required URLs are present.

Raises:

Type	Description
ValueError	In production mode only, if any of CSIAPPS_AUTH_URL, CSIAPPS_TOKEN_URL, or CSIAPPS_REDIRECT_URI is missing or does not start with http:// or https://.

Example

import csiapps

# In sandbox mode: prints token status and returns True.
csiapps.check_secrets(verbose=True)

Note

The auth and token URLs are derived from the institute set via set_institute, not read directly from the environment, so set the institute before calling this.

Source code in csiapps/auth.py

def check_secrets(verbose: bool = False, sandbox: bool | None = None) -> bool:
    """Validate that the environment is configured for authentication.

    Call this once at app startup to fail fast on a misconfigured deployment.
    Behaviour depends on the mode:

    - **Sandbox mode:** OAuth secret checks are skipped entirely (the sandbox
      simulates login and needs no client credentials). Instead the presence or
      absence of ``CSIAPPS_ACCESS_TOKEN`` is reported to stderr, and the function
      never raises.
    - **Production mode:** the OAuth URLs derived from the configured institute
      plus ``CSIAPPS_REDIRECT_URI`` are checked for a well-formed ``http(s)``
      scheme, and a :class:`ValueError` is raised listing any that are missing or
      malformed.

    Args:
        verbose: If ``True``, dump the resolved CSIAPPS environment (client id,
            auth/token/userinfo URLs, redirect URI, scope, and whether a client
            secret is set) to stderr. The client secret value itself is never
            printed — only whether one is present. Defaults to ``False``.
        sandbox: Force the mode for this check. ``True`` or ``False`` overrides
            detection; ``None`` (the default) resolves via
            [`is_sandbox_mode`][csiapps.config.is_sandbox_mode].

    Returns:
        bool: Always ``True`` when the environment is usable. In production the
        function raises rather than returning ``False``, so a ``True`` return is
        a positive assurance the required URLs are present.

    Raises:
        ValueError: In production mode only, if any of ``CSIAPPS_AUTH_URL``,
            ``CSIAPPS_TOKEN_URL``, or ``CSIAPPS_REDIRECT_URI`` is missing or does
            not start with ``http://`` or ``https://``.

    Example:
        ```python
        import csiapps

        # In sandbox mode: prints token status and returns True.
        csiapps.check_secrets(verbose=True)
        ```

    Note:
        The auth and token URLs are derived from the institute set via
        [`set_institute`][csiapps.config.set_institute], not read directly from
        the environment, so set the institute before calling this.
    """
    if sandbox is None:
        sandbox = config.is_sandbox_mode()

    if sandbox:
        if os.environ.get("CSIAPPS_ACCESS_TOKEN", ""):
            _message(
                "csiapps sandbox: CSIAPPS_ACCESS_TOKEN found - real registration reads enabled"
            )
        else:
            _message(
                "csiapps sandbox: no CSIAPPS_ACCESS_TOKEN set - running unauthenticated "
                "(set a token to emulate login and load /me)"
            )
        return True

    url_re = re.compile(r"^https?://")
    bad = []
    if not url_re.match(config.auth_url()):
        bad.append("CSIAPPS_AUTH_URL")
    if not url_re.match(config.token_url()):
        bad.append("CSIAPPS_TOKEN_URL")
    if not url_re.match(os.environ.get("CSIAPPS_REDIRECT_URI", "")):
        bad.append("CSIAPPS_REDIRECT_URI")
    if bad:
        raise ValueError("Invalid or missing URL env vars: " + ", ".join(bad))

    if verbose:
        _message(
            f"AUTH_URL: '{config.auth_url()}'  "
            f"REDIRECT_URI: '{os.environ.get('CSIAPPS_REDIRECT_URI', '')}'"
        )
        env_dump = {
            "CSIAPPS_CLIENT_ID": os.environ.get("CSIAPPS_CLIENT_ID", ""),
            "CSIAPPS_CLIENT_SECRET_SET": bool(os.environ.get("CSIAPPS_CLIENT_SECRET", "")),
            "CSIAPPS_AUTH_URL": config.auth_url(),
            "CSIAPPS_TOKEN_URL": config.token_url(),
            "CSIAPPS_REDIRECT_URI": os.environ.get("CSIAPPS_REDIRECT_URI", ""),
            "CSIAPPS_SCOPE": os.environ.get("CSIAPPS_SCOPE", "read write"),
            "CSIAPPS_USERINFO_URL": config.userinfo_url(),
        }
        _message("CSIAPPS environment on startup:")
        _message(json.dumps(env_dump, indent=2))

    return True

Client

make_request

make_request(endpoint: str, method: str = 'GET', body: dict | None = None, query: dict | None = None, headers: dict | None = None, token: str | None = None, timeout: float = 20, verbose: bool = False, paginate: bool = False, max_pages: int = 50, sandbox: bool | None = None) -> dict | list

Make an authenticated request to a CSIAPPS warehouse endpoint.

This is the low-level primitive the fetch_* helpers build on; reach for it directly when you need an endpoint those helpers do not cover. In sandbox mode the request is served by the local emulator with no network or auth; in production it is sent over HTTPS with a bearer token and a bounded retry on transient failures.

Parameters:

Name	Type	Description	Default
endpoint	str	Warehouse endpoint path, with or without leading/trailing slashes (e.g. "api/warehouse/data-records").	required
method	str	HTTP method, e.g. "GET" or "POST". Defaults to "GET".	'GET'
body	dict | None	JSON-serialisable request body for write methods. None sends no body.	None
query	dict | None	Query-string parameters as a dict. None is treated as {}.	None
headers	dict | None	Extra request headers merged over the Authorization header. Ignored in sandbox mode.	None
token	str | None	Bearer token to authenticate with. When omitted it is resolved per-session and then from the CSIAPPS_ACCESS_TOKEN environment variable (see the current_token resolver).	None
timeout	float	Per-request timeout in seconds. Defaults to 20.	20
verbose	bool	If True, print the method, endpoint, params, and raw response body for debugging. Defaults to False.	False
paginate	bool	If True, follow the response's next links and return a list of pages rather than a single response. Defaults to False.	False
max_pages	int	Upper bound on pages fetched when paginate is True, guarding against an unbounded loop. Defaults to 50.	50
sandbox	bool | None	Force sandbox (True) or production (False) routing. None (the default) resolves via is_sandbox_mode.	None

Returns:

Type	Description
dict | list	dict | list: The parsed JSON response. A single request returns the
dict | list	decoded body (typically a dict); with paginate=True it returns a
dict | list	list of page bodies. An empty response body yields [].

Raises:

Type	Description
RuntimeError	If no token is available in production mode, or if the API responds with a status of 400 or higher.

Example

Read ingested records back from the sandbox warehouse:

import csiapps

csiapps.set_sandbox_mode(True)
csiapps.make_request(
    "api/warehouse/data-records",
    query={"source_uuid": "my-source"},
)
# -> {'count': 0, 'next': None, 'previous': None, 'results': []}

Note

In sandbox mode only warehouse endpoints are emulated (see the csiapps.sandbox module); an unrecognised endpoint raises a RuntimeError with a 501-style message. Transient production failures (429/500/502/503/504) are retried up to three times with exponential backoff.

Source code in csiapps/client.py

def make_request(
    endpoint: str,
    method: str = "GET",
    body: dict | None = None,
    query: dict | None = None,
    headers: dict | None = None,
    token: str | None = None,
    timeout: float = 20,
    verbose: bool = False,
    paginate: bool = False,
    max_pages: int = 50,
    sandbox: bool | None = None,
) -> dict | list:
    """Make an authenticated request to a CSIAPPS warehouse endpoint.

    This is the low-level primitive the ``fetch_*`` helpers build on; reach for
    it directly when you need an endpoint those helpers do not cover. In sandbox
    mode the request is served by the local emulator with no network or auth; in
    production it is sent over HTTPS with a bearer token and a bounded retry on
    transient failures.

    Args:
        endpoint: Warehouse endpoint path, with or without leading/trailing
            slashes (e.g. ``"api/warehouse/data-records"``).
        method: HTTP method, e.g. ``"GET"`` or ``"POST"``. Defaults to ``"GET"``.
        body: JSON-serialisable request body for write methods. ``None`` sends no
            body.
        query: Query-string parameters as a dict. ``None`` is treated as ``{}``.
        headers: Extra request headers merged over the ``Authorization`` header.
            Ignored in sandbox mode.
        token: Bearer token to authenticate with. When omitted it is resolved
            per-session and then from the ``CSIAPPS_ACCESS_TOKEN`` environment
            variable (see the `current_token` resolver).
        timeout: Per-request timeout in seconds. Defaults to ``20``.
        verbose: If ``True``, print the method, endpoint, params, and raw
            response body for debugging. Defaults to ``False``.
        paginate: If ``True``, follow the response's ``next`` links and return a
            list of pages rather than a single response. Defaults to ``False``.
        max_pages: Upper bound on pages fetched when ``paginate`` is ``True``,
            guarding against an unbounded loop. Defaults to ``50``.
        sandbox: Force sandbox (``True``) or production (``False``) routing.
            ``None`` (the default) resolves via
            [`is_sandbox_mode`][csiapps.config.is_sandbox_mode].

    Returns:
        dict | list: The parsed JSON response. A single request returns the
        decoded body (typically a ``dict``); with ``paginate=True`` it returns a
        ``list`` of page bodies. An empty response body yields ``[]``.

    Raises:
        RuntimeError: If no token is available in production mode, or if the API
            responds with a status of 400 or higher.

    Example:
        Read ingested records back from the sandbox warehouse:

        ```python
        import csiapps

        csiapps.set_sandbox_mode(True)
        csiapps.make_request(
            "api/warehouse/data-records",
            query={"source_uuid": "my-source"},
        )
        # -> {'count': 0, 'next': None, 'previous': None, 'results': []}
        ```

    Note:
        In sandbox mode only warehouse endpoints are emulated (see the
        `csiapps.sandbox` module); an unrecognised endpoint raises a
        ``RuntimeError`` with a 501-style message. Transient production failures
        (429/500/502/503/504) are retried up to three times with exponential
        backoff.
    """
    if query is None:
        query = {}
    if headers is None:
        headers = {}
    if sandbox is None:
        sandbox = config.is_sandbox_mode()

    if sandbox:
        from . import sandbox as _sb

        return _sb._make_sandbox_request(
            endpoint=endpoint,
            method=method,
            body=body,
            query=query,
            verbose=verbose,
            paginate=paginate,
        )

    if not token:
        token = current_token()

    return _http_request(
        endpoint, method, body, query, headers, token, timeout, verbose, paginate, max_pages
    )

fetch_org_options

fetch_org_options(token: str | None = None, sandbox: bool | None = None) -> dict

Fetch sport-organisation options as a {value: label} dict.

The returned mapping plugs directly into a Shiny select input's choices= argument, which maps each value to its displayed label.

Parameters:

Name	Type	Description	Default
token	str | None	Bearer token to authenticate with. When omitted it is resolved per-session and then from CSIAPPS_ACCESS_TOKEN.	None
sandbox	bool | None	Force sandbox (True) or production (False) routing. None (the default) resolves via is_sandbox_mode.	None

Returns:

Name	Type	Description
dict	dict	A mapping of organisation id to organisation name, ready to pass as
	dict	ui.input_select(..., choices=...). Empty if no organisations are
	dict	available.

Raises:

Type	Description
RuntimeError	If no token is available in production mode, or the API responds with a status of 400 or higher.

Example

import csiapps

csiapps.set_sandbox_mode(True)
csiapps.create_sport_org("Rowing", id=7)
csiapps.fetch_org_options()   # -> {7: 'Rowing'}

Note

The {value: label} shape differs from the R package, which returned label/value pairs for R's selectInput — that shape raises TypeError: unhashable type: 'dict' in Shiny for Python. In sandbox mode the options come from the local registry populated by create_sport_org.

Source code in csiapps/client.py

def fetch_org_options(token: str | None = None, sandbox: bool | None = None) -> dict:
    """Fetch sport-organisation options as a ``{value: label}`` dict.

    The returned mapping plugs directly into a Shiny select input's ``choices=``
    argument, which maps each value to its displayed label.

    Args:
        token: Bearer token to authenticate with. When omitted it is resolved
            per-session and then from ``CSIAPPS_ACCESS_TOKEN``.
        sandbox: Force sandbox (``True``) or production (``False``) routing.
            ``None`` (the default) resolves via
            [`is_sandbox_mode`][csiapps.config.is_sandbox_mode].

    Returns:
        dict: A mapping of organisation id to organisation name, ready to pass as
        ``ui.input_select(..., choices=...)``. Empty if no organisations are
        available.

    Raises:
        RuntimeError: If no token is available in production mode, or the API
            responds with a status of 400 or higher.

    Example:
        ```python
        import csiapps

        csiapps.set_sandbox_mode(True)
        csiapps.create_sport_org("Rowing", id=7)
        csiapps.fetch_org_options()   # -> {7: 'Rowing'}
        ```

    Note:
        The ``{value: label}`` shape differs from the R package, which returned
        ``label``/``value`` pairs for R's ``selectInput`` — that shape raises
        ``TypeError: unhashable type: 'dict'`` in Shiny for Python. In sandbox
        mode the options come from the local registry populated by
        [`create_sport_org`][csiapps.sandbox.create_sport_org].
    """
    if sandbox is None:
        sandbox = config.is_sandbox_mode()
    if sandbox:
        from . import sandbox as _sb

        return _sb._sandbox_org_options()

    if not token:
        token = current_token()
    if not token:
        _auth_gate("fetch_org_options")

    url = config.site_url().rstrip("/") + config.SPORT_ORG_ENDPOINT
    resp = _perform(
        "GET",
        url,
        params={"limit": 1000},
        headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
    )
    if resp.status_code >= 400:
        raise RuntimeError(f"fetch_org_options failed ({resp.status_code}): {resp.text}")

    items = resp.json()
    if isinstance(items, dict) and items.get("results") is not None:
        return {it.get("id"): it.get("name") for it in items["results"]}
    if isinstance(items, list):
        return {v: v for v in items}
    return {}

fetch_profiles

fetch_profiles(token: str | None = None, filters: dict | None = None, sandbox: bool | None = None, max_pages: int = 50) -> list

Fetch all registration profiles accessible to the token, auto-paginating.

Follows the API's pagination automatically and returns every profile as a flat list. Pass the result through flatten_profile before rendering it in a table.

Parameters:

Name	Type	Description	Default
token	str | None	Bearer token to authenticate with. When omitted it is resolved per-session and then from CSIAPPS_ACCESS_TOKEN.	None
filters	dict | None	Query parameters narrowing the result, e.g. {"sport_org_id": 42}. In sandbox mode only sport_org_id is honoured. None fetches all accessible profiles.	None
sandbox	bool | None	Force sandbox (True) or production (False) routing. None (the default) resolves via is_sandbox_mode.	None
max_pages	int	Upper bound on pages fetched, matching make_request. Defaults to 50.	50

Returns:

Name	Type	Description
list	list	A list of profile dicts (nested, production-shaped). Empty if no
	list	profiles match.

Raises:

Type	Description
RuntimeError	If no token is available in production mode, or the API responds with a status of 400 or higher.

Warns:

Type	Description
UserWarning	If max_pages is reached while the server still advertises more pages; the result may be truncated. Pass a larger max_pages to fetch the rest.

Example

import csiapps

csiapps.set_sandbox_mode(True)
profiles = csiapps.fetch_profiles(filters={"sport_org_id": 7})
rows = [csiapps.flatten_profile(p) for p in profiles]

Note

Pagination terminates if the server ever repeats a next URL, so a misbehaving or hostile server cannot hang the app in an unbounded loop with unbounded memory growth.

Source code in csiapps/client.py

def fetch_profiles(
    token: str | None = None,
    filters: dict | None = None,
    sandbox: bool | None = None,
    max_pages: int = 50,
) -> list:
    """Fetch all registration profiles accessible to the token, auto-paginating.

    Follows the API's pagination automatically and returns every profile as a
    flat list. Pass the result through
    [`flatten_profile`][csiapps.client.flatten_profile] before rendering it in a
    table.

    Args:
        token: Bearer token to authenticate with. When omitted it is resolved
            per-session and then from ``CSIAPPS_ACCESS_TOKEN``.
        filters: Query parameters narrowing the result, e.g.
            ``{"sport_org_id": 42}``. In sandbox mode only ``sport_org_id`` is
            honoured. ``None`` fetches all accessible profiles.
        sandbox: Force sandbox (``True``) or production (``False``) routing.
            ``None`` (the default) resolves via
            [`is_sandbox_mode`][csiapps.config.is_sandbox_mode].
        max_pages: Upper bound on pages fetched, matching
            [`make_request`][csiapps.client.make_request]. Defaults to ``50``.

    Returns:
        list: A list of profile dicts (nested, production-shaped). Empty if no
        profiles match.

    Raises:
        RuntimeError: If no token is available in production mode, or the API
            responds with a status of 400 or higher.

    Warns:
        UserWarning: If ``max_pages`` is reached while the server still
            advertises more pages; the result may be truncated. Pass a larger
            ``max_pages`` to fetch the rest.

    Example:
        ```python
        import csiapps

        csiapps.set_sandbox_mode(True)
        profiles = csiapps.fetch_profiles(filters={"sport_org_id": 7})
        rows = [csiapps.flatten_profile(p) for p in profiles]
        ```

    Note:
        Pagination terminates if the server ever repeats a ``next`` URL, so a
        misbehaving or hostile server cannot hang the app in an unbounded loop
        with unbounded memory growth.
    """
    if filters is None:
        filters = {}
    if sandbox is None:
        sandbox = config.is_sandbox_mode()
    if sandbox:
        from . import sandbox as _sb

        return _sb._sandbox_profiles(filters.get("sport_org_id"))

    if not token:
        token = current_token()
    if not token:
        _auth_gate("fetch_profiles")

    url = config.site_url().rstrip("/") + config.PROFILE_ENDPOINT
    params = {**filters, "limit": 100, "offset": 0}
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}

    out = []
    next_url = url
    seen = set()
    for _ in range(max_pages):
        if next_url in seen:  # server cycled `next` back to a page we already fetched
            break
        seen.add(next_url)
        resp = _perform("GET", next_url, params=params, headers=headers)
        if resp.status_code >= 400:
            raise RuntimeError(f"fetch_profiles failed ({resp.status_code}): {resp.text}")
        payload = resp.json()
        out.extend(payload.get("results") or [])
        next_url = payload.get("next")
        params = None  # `next` already carries its query
        if not next_url:
            break
    else:
        # Loop hit max_pages with more pages still advertised. Warn loudly rather
        # than silently truncate, so a genuinely large result is never dropped
        # without notice (raise max_pages to fetch the rest).
        if next_url:
            import warnings

            warnings.warn(
                f"fetch_profiles: stopped after max_pages={max_pages}; results may be "
                f"truncated. Pass a larger max_pages to fetch all profiles.",
                stacklevel=2,
            )
    return out

fetch_profile

fetch_profile(profile_id: int | str, token: str | None = None, sandbox: bool | None = None) -> dict | None

Fetch a single registration profile by id.

Parameters:

Name	Type	Description	Default
profile_id	int | str	The profile's id. Coerced to a string and URL-encoded before being placed in the request path, so an unusual value cannot alter the URL.	required
token	str | None	Bearer token to authenticate with. When omitted it is resolved per-session and then from CSIAPPS_ACCESS_TOKEN.	None
sandbox	bool | None	Force sandbox (True) or production (False) routing. None (the default) resolves via is_sandbox_mode.	None

Returns:

Type	Description
dict | None	dict | None: The profile dict. In sandbox mode returns None when no
dict | None	profile with that id exists.

Raises:

Type	Description
RuntimeError	If no token is available in production mode, or the API responds with a status of 400 or higher.

Example

import csiapps

csiapps.set_sandbox_mode(True)
csiapps.fetch_profile(1)
# -> {'id': 1, 'person': {...}, 'sport': {...}, 'status': 'ACTIVE', ...}

Note

A production GET for a missing id raises RuntimeError (from the 4xx status) rather than returning None; only the sandbox reader distinguishes "not found" as None.

Source code in csiapps/client.py

def fetch_profile(
    profile_id: int | str,
    token: str | None = None,
    sandbox: bool | None = None,
) -> dict | None:
    """Fetch a single registration profile by id.

    Args:
        profile_id: The profile's id. Coerced to a string and URL-encoded before
            being placed in the request path, so an unusual value cannot alter
            the URL.
        token: Bearer token to authenticate with. When omitted it is resolved
            per-session and then from ``CSIAPPS_ACCESS_TOKEN``.
        sandbox: Force sandbox (``True``) or production (``False``) routing.
            ``None`` (the default) resolves via
            [`is_sandbox_mode`][csiapps.config.is_sandbox_mode].

    Returns:
        dict | None: The profile dict. In sandbox mode returns ``None`` when no
        profile with that id exists.

    Raises:
        RuntimeError: If no token is available in production mode, or the API
            responds with a status of 400 or higher.

    Example:
        ```python
        import csiapps

        csiapps.set_sandbox_mode(True)
        csiapps.fetch_profile(1)
        # -> {'id': 1, 'person': {...}, 'sport': {...}, 'status': 'ACTIVE', ...}
        ```

    Note:
        A production ``GET`` for a missing id raises ``RuntimeError`` (from the
        4xx status) rather than returning ``None``; only the sandbox reader
        distinguishes "not found" as ``None``.
    """
    if sandbox is None:
        sandbox = config.is_sandbox_mode()
    if sandbox:
        from . import sandbox as _sb

        return _sb._sandbox_profile(profile_id)

    if not token:
        token = current_token()
    if not token:
        _auth_gate("fetch_profile")

    # URL-encode the id so an unusual value can't alter the request path.
    enc_id = quote(str(profile_id), safe="")
    url = config.site_url().rstrip("/") + config.PROFILE_ENDPOINT + enc_id
    resp = _perform(
        "GET", url, headers={"Authorization": f"Bearer {token}", "Accept": "application/json"}
    )
    if resp.status_code >= 400:
        raise RuntimeError(f"fetch_profile failed ({resp.status_code}): {resp.text}")
    return resp.json()

flatten_profile

flatten_profile(p: dict) -> dict

Flatten a nested registration profile into a scalar row.

fetch_profiles returns deeply nested dicts; passing them straight to a Shiny data frame fails with "Unsupported dataframe type". This picks out the commonly displayed fields into a flat, one-level dict so a list of them builds a table (wrap in pandas/polars for render.data_frame).

Parameters:

Name	Type	Description	Default
p	dict	A single profile dict as returned by fetch_profiles or fetch_profile. Missing nested sections are tolerated (treated as empty).	required

Returns:

Name	Type	Description
dict	dict	A flat row with keys id, first_name, last_name,
	dict	email, dob, sport_id, sport, and status. Any absent
	dict	source field is None.

Example

import csiapps
import pandas as pd

profiles = csiapps.fetch_profiles()
df = pd.DataFrame(csiapps.flatten_profile(p) for p in profiles)

Source code in csiapps/client.py

def flatten_profile(p: dict) -> dict:
    """Flatten a nested registration profile into a scalar row.

    [`fetch_profiles`][csiapps.client.fetch_profiles] returns deeply nested
    dicts; passing them straight to a Shiny data frame fails with "Unsupported
    dataframe type". This picks out the commonly displayed fields into a flat,
    one-level dict so a list of them builds a table (wrap in pandas/polars for
    ``render.data_frame``).

    Args:
        p: A single profile dict as returned by
            [`fetch_profiles`][csiapps.client.fetch_profiles] or
            [`fetch_profile`][csiapps.client.fetch_profile]. Missing nested
            sections are tolerated (treated as empty).

    Returns:
        dict: A flat row with keys ``id``, ``first_name``, ``last_name``,
        ``email``, ``dob``, ``sport_id``, ``sport``, and ``status``. Any absent
        source field is ``None``.

    Example:
        ```python
        import csiapps
        import pandas as pd

        profiles = csiapps.fetch_profiles()
        df = pd.DataFrame(csiapps.flatten_profile(p) for p in profiles)
        ```
    """
    person = p.get("person") or {}
    sport = p.get("sport") or {}
    return {
        "id": p.get("id"),
        "first_name": person.get("first_name"),
        "last_name": person.get("last_name"),
        "email": person.get("email"),
        "dob": person.get("dob"),
        "sport_id": sport.get("id"),
        "sport": sport.get("name"),
        "status": p.get("status"),
    }

token_ready

token_ready() -> bool

Whether a CSIAPPS access token is available for the current context.

Returns True once a token is available: inside a Shiny app wrapped by :func:csiapps.server_wrapper, the per-session token stored at login; outside Shiny, the CSIAPPS_ACCESS_TOKEN environment variable.

The check is reactive-friendly: called from a reactive context it takes a dependency on the session's token, so a guard like req(token_ready()) re-fires automatically when login completes instead of sticking in the cancelled state. :func:make_request and the fetch_* helpers already gate themselves this way, so an explicit guard is only needed for work that should wait for login without making an API call (e.g. processing an uploaded file).

Source code in csiapps/client.py

def token_ready() -> bool:
    """Whether a CSIAPPS access token is available for the current context.

    Returns ``True`` once a token is available: inside a Shiny app wrapped by
    :func:`csiapps.server_wrapper`, the per-session token stored at login;
    outside Shiny, the ``CSIAPPS_ACCESS_TOKEN`` environment variable.

    The check is reactive-friendly: called from a reactive context it takes a
    dependency on the session's token, so a guard like ``req(token_ready())``
    re-fires automatically when login completes instead of sticking in the
    cancelled state. :func:`make_request` and the ``fetch_*`` helpers already
    gate themselves this way, so an explicit guard is only needed for work that
    should wait for login without making an API call (e.g. processing an
    uploaded file).
    """
    return bool(current_token())

Sandbox

register_sandbox_schema

register_sandbox_schema(source_uuid: str, schema: dict | str) -> dict

Register a JSON schema for a data source in the sandbox.

The schema is what the sandbox validates ingested records against, so register one before calling ingestion for that source. Registering again for the same source_uuid replaces the previous schema.

Parameters:

Name	Type	Description	Default
source_uuid	str	The data-source identifier to register the schema under. Must be a non-empty string.	required
schema	dict | str	The JSON Schema (Draft 7) to validate records against. May be a dict, a JSON string, or a path to a .json file.	required

Returns:

Name	Type	Description
dict	dict	The parsed schema that was stored (useful when a path or JSON
	dict	string was passed in).

Raises:

Type	Description
ValueError	If source_uuid is empty, or schema is not a dict, JSON string, or readable JSON file.

Example

import csiapps

csiapps.set_sandbox_mode(True)
csiapps.register_sandbox_schema(
    "hr-source",
    {"type": "object", "required": ["athlete_id", "hr"]},
)

Note

Validation uses jsonschema (Draft 7) where the R package used Ajv via jsonvalidate; validator wording can differ on edge cases. Registered schemas live for the process — see clear_sandbox to reset.

Source code in csiapps/sandbox.py

def register_sandbox_schema(source_uuid: str, schema: dict | str) -> dict:
    """Register a JSON schema for a data source in the sandbox.

    The schema is what the sandbox validates ingested records against, so
    register one before calling ingestion for that source. Registering again for
    the same ``source_uuid`` replaces the previous schema.

    Args:
        source_uuid: The data-source identifier to register the schema under.
            Must be a non-empty string.
        schema: The JSON Schema (Draft 7) to validate records against. May be a
            dict, a JSON string, or a path to a ``.json`` file.

    Returns:
        dict: The parsed schema that was stored (useful when a path or JSON
        string was passed in).

    Raises:
        ValueError: If ``source_uuid`` is empty, or ``schema`` is not a dict,
            JSON string, or readable JSON file.

    Example:
        ```python
        import csiapps

        csiapps.set_sandbox_mode(True)
        csiapps.register_sandbox_schema(
            "hr-source",
            {"type": "object", "required": ["athlete_id", "hr"]},
        )
        ```

    Note:
        Validation uses ``jsonschema`` (Draft 7) where the R package used Ajv via
        ``jsonvalidate``; validator wording can differ on edge cases. Registered
        schemas live for the process — see
        [`clear_sandbox`][csiapps.sandbox.clear_sandbox] to reset.
    """
    if not (isinstance(source_uuid, str) and source_uuid):
        raise ValueError("register_sandbox_schema: `source_uuid` must be a non-empty string.")

    if isinstance(schema, str):
        if os.path.isfile(schema):
            with open(schema) as f:
                schema = json.load(f)
        else:
            schema = json.loads(schema)
    if not isinstance(schema, dict):
        raise ValueError(
            "register_sandbox_schema: `schema` must be a list, a JSON string, "
            "or a path to a JSON file."
        )

    _state["schemas"][source_uuid] = schema
    _message(f"csiapps sandbox: schema registered for source '{source_uuid}'")
    return schema

create_sport_org

create_sport_org(name: str, id: int | None = None) -> dict

Create a dummy sport organisation in the sandbox registry.

Populates the local registry that the sandbox branches of fetch_org_options and fetch_profiles read from, so those helpers behave like production without a network call. Create an org before adding athletes to it with create_profile.

Parameters:

Name	Type	Description	Default
name	str	Display name for the organisation. Must be a non-empty string.	required
id	int | None	Organisation id, a positive integer in 1..999. If None (the default), an unused id in that range is chosen at random.	None

Returns:

Name	Type	Description
dict	dict	The created org with keys id, name, and
	dict	annual_cycle_start (today's date).

Raises:

Type	Description
ValueError	If name is empty, id is not a positive integer in 1..999, or an org with that id already exists.
RuntimeError	If all 999 ids are already in use.

Warns:

Type	Description
UserWarning	If called while not in sandbox mode — dummy orgs are only read by sandbox helpers and have no effect in production.

Example

import csiapps

csiapps.set_sandbox_mode(True)
csiapps.create_sport_org("Rowing", id=7)
# -> {'id': 7, 'name': 'Rowing', 'annual_cycle_start': '2026-07-16'}

Source code in csiapps/sandbox.py

def create_sport_org(name: str, id: int | None = None) -> dict:
    """Create a dummy sport organisation in the sandbox registry.

    Populates the local registry that the sandbox branches of
    [`fetch_org_options`][csiapps.client.fetch_org_options] and
    [`fetch_profiles`][csiapps.client.fetch_profiles] read from, so those
    helpers behave like production without a network call. Create an org before
    adding athletes to it with
    [`create_profile`][csiapps.sandbox.create_profile].

    Args:
        name: Display name for the organisation. Must be a non-empty string.
        id: Organisation id, a positive integer in ``1..999``. If ``None`` (the
            default), an unused id in that range is chosen at random.

    Returns:
        dict: The created org with keys ``id``, ``name``, and
        ``annual_cycle_start`` (today's date).

    Raises:
        ValueError: If ``name`` is empty, ``id`` is not a positive integer in
            ``1..999``, or an org with that ``id`` already exists.
        RuntimeError: If all 999 ids are already in use.

    Warns:
        UserWarning: If called while not in sandbox mode — dummy orgs are only
            read by sandbox helpers and have no effect in production.

    Example:
        ```python
        import csiapps

        csiapps.set_sandbox_mode(True)
        csiapps.create_sport_org("Rowing", id=7)
        # -> {'id': 7, 'name': 'Rowing', 'annual_cycle_start': '2026-07-16'}
        ```
    """
    if not config.is_sandbox_mode():
        warnings.warn(
            "create_sport_org: not in sandbox mode; dummy orgs are only read by "
            "sandbox helpers and have no effect in production.",
            stacklevel=2,
        )
    if not (isinstance(name, str) and name):
        raise ValueError("create_sport_org: `name` must be a non-empty string.")

    existing = _org_ids()
    if id is None:
        pool = [i for i in range(1, 1000) if i not in existing]
        if not pool:
            raise RuntimeError(
                "create_sport_org: too many sport orgs in the sandbox. Limit is 999."
            )
        id = random.choice(pool)
    else:
        ok = (
            isinstance(id, (int, float))
            and not isinstance(id, bool)
            and float(id).is_integer()
            and 0 < id <= 999
        )
        if not ok:
            raise ValueError("create_sport_org: `id` must be a positive integer in 1:999.")
        id = int(id)
        if id in existing:
            raise ValueError(f"create_sport_org: sport org id {id} already exists in the sandbox.")

    org = {"id": id, "name": name, "annual_cycle_start": date.today().isoformat()}
    _state["orgs"][str(id)] = org
    _message(f"csiapps sandbox: created sport org {id} ('{name}')")
    return org

create_profile

create_profile(n: int, sport_org_id: int, first_names: list[str] | None = None, last_names: list[str] | None = None) -> list

Create n dummy athlete profiles under an existing sandbox sport org.

The generated profiles are production-shaped and become readable through the sandbox branches of fetch_profiles and fetch_profile. Their ids also let ingested records resolve a subject when read back.

Parameters:

Name	Type	Description	Default
n	int	Number of profiles to create. Must be a non-negative integer.	required
sport_org_id	int	Id of an existing sandbox sport org (create one first with create_sport_org).	required
first_names	list[str] | None	Optional explicit first names, length n. If omitted, random distinct names are generated. Must be given together with last_names or not at all.	None
last_names	list[str] | None	Optional explicit last names, length n. Same rules as first_names.	None

Returns:

Name	Type	Description
list	list	The newly created profile dicts (production-shaped).

Raises:

Type	Description
ValueError	If n is not a non-negative integer, the sport org does not exist, only one of first_names/last_names is given, or the provided name lists are not each of length n.

Warns:

Type	Description
UserWarning	If called while not in sandbox mode — dummy profiles are only read by sandbox helpers and have no effect in production.

Example

import csiapps

csiapps.set_sandbox_mode(True)
csiapps.create_sport_org("Rowing", id=7)
athletes = csiapps.create_profile(3, sport_org_id=7)
len(athletes)   # -> 3

Note

Random names use faker where the R package used the babynames dataset; generated full names are guaranteed distinct within a call.

Source code in csiapps/sandbox.py

def create_profile(
    n: int,
    sport_org_id: int,
    first_names: list[str] | None = None,
    last_names: list[str] | None = None,
) -> list:
    """Create ``n`` dummy athlete profiles under an existing sandbox sport org.

    The generated profiles are production-shaped and become readable through the
    sandbox branches of [`fetch_profiles`][csiapps.client.fetch_profiles] and
    [`fetch_profile`][csiapps.client.fetch_profile]. Their ids also let ingested
    records resolve a ``subject`` when read back.

    Args:
        n: Number of profiles to create. Must be a non-negative integer.
        sport_org_id: Id of an existing sandbox sport org (create one first with
            [`create_sport_org`][csiapps.sandbox.create_sport_org]).
        first_names: Optional explicit first names, length ``n``. If omitted,
            random distinct names are generated. Must be given together with
            ``last_names`` or not at all.
        last_names: Optional explicit last names, length ``n``. Same rules as
            ``first_names``.

    Returns:
        list: The newly created profile dicts (production-shaped).

    Raises:
        ValueError: If ``n`` is not a non-negative integer, the sport org does
            not exist, only one of ``first_names``/``last_names`` is given, or
            the provided name lists are not each of length ``n``.

    Warns:
        UserWarning: If called while not in sandbox mode — dummy profiles are
            only read by sandbox helpers and have no effect in production.

    Example:
        ```python
        import csiapps

        csiapps.set_sandbox_mode(True)
        csiapps.create_sport_org("Rowing", id=7)
        athletes = csiapps.create_profile(3, sport_org_id=7)
        len(athletes)   # -> 3
        ```

    Note:
        Random names use ``faker`` where the R package used the ``babynames``
        dataset; generated full names are guaranteed distinct within a call.
    """
    if not config.is_sandbox_mode():
        warnings.warn(
            "create_profile: not in sandbox mode; dummy profiles are only read by "
            "sandbox helpers and have no effect in production.",
            stacklevel=2,
        )
    if not (isinstance(n, int) and not isinstance(n, bool) and n >= 0):
        raise ValueError("create_profile: `n` must be a non-negative integer.")
    sport_org_id = int(sport_org_id)
    if str(sport_org_id) not in _state["orgs"]:
        raise ValueError(
            f"create_profile: sport org '{sport_org_id}' does not exist; "
            f"create it first with create_sport_org({sport_org_id})."
        )

    if first_names is None and last_names is None:
        first_names, last_names = _random_names(n)
    elif first_names is None or last_names is None:
        raise ValueError("create_profile: provide both `first_names` and `last_names`, or neither.")
    else:
        if not (len(first_names) == n and len(last_names) == n):
            raise ValueError(
                "create_profile: `first_names` and `last_names` must each have length n."
            )

    start = len(_state["profiles"])
    new = [
        _make_profile(start + i + 1, sport_org_id, first_names[i], last_names[i])
        for i in range(n)
    ]
    _state["profiles"].extend(new)
    _message(
        f"csiapps sandbox: created {n} athlete(s) under sport org {sport_org_id} "
        f"({len(_state['profiles'])} total)"
    )
    return new

clear_sandbox

clear_sandbox(source_uuid: str | None = None) -> None

Reset sandbox state, entirely or for a single data source.

Useful in test teardown so registered schemas, ingested records, and on-disk payloads do not leak between runs.

Parameters:

Name	Type	Description	Default
source_uuid	str | None	If given, clear only that source's schema, records, and payload directory. If None (the default), clear everything: all schemas, records, dummy orgs, profiles, and payload directories.	None

Returns:

Type	Description
None	None

Example

import csiapps

csiapps.clear_sandbox("hr-source")   # one source
csiapps.clear_sandbox()              # everything

Source code in csiapps/sandbox.py

def clear_sandbox(source_uuid: str | None = None) -> None:
    """Reset sandbox state, entirely or for a single data source.

    Useful in test teardown so registered schemas, ingested records, and
    on-disk payloads do not leak between runs.

    Args:
        source_uuid: If given, clear only that source's schema, records, and
            payload directory. If ``None`` (the default), clear everything:
            all schemas, records, dummy orgs, profiles, and payload
            directories.

    Returns:
        None

    Example:
        ```python
        import csiapps

        csiapps.clear_sandbox("hr-source")   # one source
        csiapps.clear_sandbox()              # everything
        ```
    """
    if source_uuid is None:
        _state["schemas"] = {}
        _state["records"] = {}
        _state["orgs"] = {}
        _state["profiles"] = []
        d = _state["dir"]
        if d and os.path.isdir(d):
            for entry in os.listdir(d):
                path = os.path.join(d, entry)
                if os.path.isdir(path):
                    shutil.rmtree(path, ignore_errors=True)
        _message("csiapps sandbox: entire sandbox cleared")
    else:
        _state["schemas"].pop(source_uuid, None)
        _state["records"].pop(source_uuid, None)
        d = _state["dir"]
        if d:
            target = os.path.join(d, source_uuid)
            if os.path.isdir(target):
                shutil.rmtree(target, ignore_errors=True)
        _message(f"csiapps sandbox: cleared source '{source_uuid}'")
    return None

browse_sandbox

browse_sandbox(source_uuid: str | None = None) -> str

Open the sandbox payload directory in the system file explorer.

Each successful ingestion writes the submitted records to disk for inspection; this opens that directory so you can see the raw payloads.

Parameters:

Name	Type	Description	Default
source_uuid	str | None	If given, open that source's subdirectory. If None (the default), open the top-level sandbox payload directory.	None

Returns:

Name	Type	Description
str	str	The filesystem path that was opened.

Raises:

Type	Description
RuntimeError	If the target directory does not exist yet — typically because nothing has been ingested for that source.

Example

import csiapps

csiapps.browse_sandbox("hr-source")
# -> '/tmp/csiapps_sandbox_ab12cd/hr-source'

Source code in csiapps/sandbox.py

def browse_sandbox(source_uuid: str | None = None) -> str:
    """Open the sandbox payload directory in the system file explorer.

    Each successful ingestion writes the submitted records to disk for
    inspection; this opens that directory so you can see the raw payloads.

    Args:
        source_uuid: If given, open that source's subdirectory. If ``None`` (the
            default), open the top-level sandbox payload directory.

    Returns:
        str: The filesystem path that was opened.

    Raises:
        RuntimeError: If the target directory does not exist yet — typically
            because nothing has been ingested for that source.

    Example:
        ```python
        import csiapps

        csiapps.browse_sandbox("hr-source")
        # -> '/tmp/csiapps_sandbox_ab12cd/hr-source'
        ```
    """
    target = sandbox_dir()
    if source_uuid is not None:
        target = os.path.join(target, source_uuid)
    if not os.path.isdir(target):
        raise RuntimeError(
            f"csiapps sandbox: directory '{target}' does not exist. "
            "Have you ingested any data yet?"
        )
    webbrowser.open(target)
    return target

App wrappers

ui_wrapper

ui_wrapper(*args: TagChild, sandbox: bool | None = None) -> Tag

Wrap an app's UI in the standard CSI chrome.

Adds the CSI navbar, footer, an auth-status line, the favicon and redirect/reset message handlers, and — in sandbox mode — a banner making it obvious the app is not connected to the live warehouse. Use it in place of ui.page_fluid at the top of an app's UI definition; pair it with server_wrapper on the server side.

Parameters:

Name	Type	Description	Default
*args	TagChild	The app's own UI elements (Shiny tags / components), rendered below the auth-status line inside the chrome.	()
sandbox	bool | None	Force the sandbox banner on (True) or off (False). None (the default) resolves via is_sandbox_mode.	None

Returns:

Type	Description
Tag	A ui.page_fluid page containing the chrome and the supplied UI.

Example

from shiny import ui
import csiapps

app_ui = csiapps.ui_wrapper(
    ui.h2("My app"),
    ui.input_action_button("logout", "Log out"),
)

Note

The chrome styles are scoped by id and marked !important so a wrapped app's own theme cannot override the navbar and footer. Include an input_action_button("logout", ...) for the logout effect wired up by server_wrapper.

Source code in csiapps/app.py

def ui_wrapper(*args: TagChild, sandbox: bool | None = None) -> Tag:
    """Wrap an app's UI in the standard CSI chrome.

    Adds the CSI navbar, footer, an auth-status line, the favicon and
    redirect/reset message handlers, and — in sandbox mode — a banner making it
    obvious the app is not connected to the live warehouse. Use it in place of
    ``ui.page_fluid`` at the top of an app's UI definition; pair it with
    [`server_wrapper`][csiapps.app.server_wrapper] on the server side.

    Args:
        *args: The app's own UI elements (Shiny tags / components), rendered
            below the auth-status line inside the chrome.
        sandbox: Force the sandbox banner on (``True``) or off (``False``).
            ``None`` (the default) resolves via
            [`is_sandbox_mode`][csiapps.config.is_sandbox_mode].

    Returns:
        A ``ui.page_fluid`` page containing the chrome and the supplied UI.

    Example:
        ```python
        from shiny import ui
        import csiapps

        app_ui = csiapps.ui_wrapper(
            ui.h2("My app"),
            ui.input_action_button("logout", "Log out"),
        )
        ```

    Note:
        The chrome styles are scoped by id and marked ``!important`` so a wrapped
        app's own theme cannot override the navbar and footer. Include an
        ``input_action_button("logout", ...)`` for the logout effect wired up by
        [`server_wrapper`][csiapps.app.server_wrapper].
    """
    if sandbox is None:
        sandbox = config.is_sandbox_mode()

    children = [
        ui.head_content(
            ui.tags.script(ui.HTML(_HANDLERS_JS)),
            ui.tags.link(rel="shortcut icon", href=_FAVICON),
        ),
        _csi_chrome_styles(),
        _navbar_ui(),
    ]
    if sandbox:
        children.append(_sandbox_banner())
    # padding-bottom leaves room for the fixed-bottom footer so it never overlaps
    # app content on short pages (mirrors the R wrapper's fluidPage style).
    children.append(
        ui.div(
            ui.output_ui("auth_status"),
            *args,
            style="padding-bottom: 80px;",
        )
    )
    children.append(_footer_ui())
    return ui.page_fluid(*children)

server_wrapper

server_wrapper(app_specific_logic: Callable, sandbox: bool | None = None) -> Callable

Wrap an app's server function with CSIAPPS authentication.

Returns a Shiny server function that handles login before delegating to your own server logic. In production it runs the OAuth2 PKCE flow (redirect to CSIAPPS, exchange the returned code for a token, load /me for the header). In sandbox mode it simulates that login using CSIAPPS_ACCESS_TOKEN if set, or marks the session unauthenticated otherwise. Either way the per-session token is stored so make_request and the fetch_* helpers pick it up automatically.

Parameters:

Name	Type	Description	Default
app_specific_logic	Callable	Your app's server function with the usual Shiny (input, output, session) signature. It is called after auth is wired up, keeping its own lexical scope.	required
sandbox	bool | None	Force sandbox (True) or production (False) auth behaviour. None (the default) resolves via is_sandbox_mode.	None

Returns:

Name	Type	Description
Callable	Callable	A server function to hand to Shiny's App(app_ui, server).

Example

from shiny import App
import csiapps

def my_server(input, output, session):
    ...

app = App(app_ui, csiapps.server_wrapper(my_server))

Note

The wrapper registers a logout effect bound to an input.logout action button — include one in the UI (see ui_wrapper). Blocking token and /me calls run off the event loop so a slow endpoint cannot stall other sessions.

Source code in csiapps/app.py

def server_wrapper(
    app_specific_logic: Callable, sandbox: bool | None = None
) -> Callable:
    """Wrap an app's server function with CSIAPPS authentication.

    Returns a Shiny server function that handles login before delegating to your
    own server logic. In production it runs the OAuth2 PKCE flow (redirect to
    CSIAPPS, exchange the returned code for a token, load ``/me`` for the header).
    In sandbox mode it simulates that login using ``CSIAPPS_ACCESS_TOKEN`` if
    set, or marks the session unauthenticated otherwise. Either way the
    per-session token is stored so [`make_request`][csiapps.client.make_request]
    and the ``fetch_*`` helpers pick it up automatically.

    Args:
        app_specific_logic: Your app's server function with the usual Shiny
            ``(input, output, session)`` signature. It is called after auth is
            wired up, keeping its own lexical scope.
        sandbox: Force sandbox (``True``) or production (``False``) auth
            behaviour. ``None`` (the default) resolves via
            [`is_sandbox_mode`][csiapps.config.is_sandbox_mode].

    Returns:
        Callable: A server function to hand to Shiny's ``App(app_ui, server)``.

    Example:
        ```python
        from shiny import App
        import csiapps

        def my_server(input, output, session):
            ...

        app = App(app_ui, csiapps.server_wrapper(my_server))
        ```

    Note:
        The wrapper registers a logout effect bound to an ``input.logout``
        action button — include one in the UI (see
        [`ui_wrapper`][csiapps.app.ui_wrapper]). Blocking token and ``/me`` calls
        run off the event loop so a slow endpoint cannot stall other sessions.
    """
    if sandbox is None:
        sandbox = config.is_sandbox_mode()

    def server(input, output, session):
        user_token = reactive.value(None)
        userinfo = reactive.value(None)

        if sandbox:
            # Simulate the redirect: seed the token from the environment and hand
            # it to the same consumer a production login would.
            user_token.set(_seed_token_value())
        else:

            @reactive.effect
            async def _oauth():
                qs = parse_qs(session.clientdata.url_search().lstrip("?"))
                code = qs.get("code", [None])[0]
                state = qs.get("state", [None])[0]
                err = qs.get("error", [None])[0]

                if err:
                    user_token.set(
                        {"error": err, "error_description": qs.get("error_description", [None])[0]}
                    )
                    client.set_session_token(session, None)
                    await session.send_custom_message("csip_reset", {})

                # 1) no code + no token -> redirect to CSI
                if code is None and user_token() is None:
                    pk = auth.generate_pkce()
                    st = auth.pkce_state_encode(pk["verifier"])
                    params = {
                        "response_type": "code",
                        "client_id": os.environ.get("CSIAPPS_CLIENT_ID", ""),
                        "redirect_uri": os.environ.get("CSIAPPS_REDIRECT_URI", ""),
                        "scope": os.environ.get("CSIAPPS_SCOPE", "read write"),
                        "code_challenge": pk["challenge"],
                        "code_challenge_method": pk["method"],
                        "state": st,
                    }
                    await session.send_custom_message(
                        "csip_redirect", config.auth_url() + "?" + urlencode(params)
                    )
                    return

                # 2) have code but no token yet -> exchange
                if code is not None and user_token() is None:
                    verifier = auth.pkce_state_decode(state).get("v") if state else None
                    # exchange_code_for_token does a *blocking* httpx.post; run it
                    # off the event loop so a slow token endpoint can't stall every
                    # other session (a Python-only concern -- R has no shared loop).
                    tok = await asyncio.to_thread(
                        auth.exchange_code_for_token, code, verifier
                    )
                    user_token.set(tok)

        # Shared consumer (production + sandbox): store the token per-session and
        # load /me for the header.
        @reactive.effect
        @reactive.event(user_token)
        async def _consume():
            tok = user_token()
            client.set_session_token(session, None)

            if tok is None or tok.get("error"):
                await session.send_custom_message("csip_reset", {})
                return

            access_token = tok.get("access_token")
            if not access_token:
                return

            client.set_session_token(session, access_token)

            userinfo_url = config.userinfo_url()
            if userinfo_url:
                try:
                    # Blocking httpx.get -> run off the event loop so a slow /me
                    # endpoint can't stall other sessions (Python-only concern).
                    resp = await asyncio.to_thread(
                        httpx.get,
                        userinfo_url,
                        headers={"Authorization": f"Bearer {access_token}"},
                        follow_redirects=True,
                    )
                    resp.raise_for_status()
                    userinfo.set(resp.json())
                except Exception as e:  # a stale/expired token degrades gracefully
                    ui.notification_show(f"Error loading user info: {e}", type="error")

        @render.ui
        def auth_status():
            tok = user_token()
            if tok is None:
                return ui.tags.p("Redirecting to CSIAPPS for authentication...")
            if tok.get("error"):
                return ui.tags.p("Authentication error (see logs).")
            if tok.get("unauthenticated"):
                return ui.tags.p(
                    ui.HTML(
                        "Not authenticated — set CSIAPPS_ACCESS_TOKEN "
                        "to emulate login in sandbox mode."
                    )
                )
            return ui.TagList(ui.tags.br(), ui.tags.p(_signed_in_text(userinfo(), sandbox)))

        @reactive.effect
        @reactive.event(input.logout, ignore_none=True)
        async def _logout():
            userinfo.set(None)
            client.set_session_token(session, None)
            if sandbox:
                user_token.set(_seed_token_value())
            else:
                user_token.set(None)
                await session.send_custom_message("csip_reset", {})

        # Call the app's own server function so it keeps its lexical scope.
        app_specific_logic(input, output, session)

    return server