Automating Canvas API Token Refresh in Python

Institutional pipelines that synchronize Canvas LMS gradebooks, attendance records, and engagement metrics run unattended for weeks at a time, and Canvas OAuth 2.0 access tokens are short-lived by design. When a token expires in the middle of a sync, the next request returns 401 Unauthorized — and a naive script either crashes (leaving a partial grade write) or, worse, silently logs the failure and reports success. This guide builds a single, self-renewing CanvasTokenManager that keeps an encrypted token cache, refreshes proactively before expiry, and intercepts a mid-stream 401 to refresh-and-retry exactly once. It slots in underneath the transport layer described in Python Requests for LMS APIs and is the credential backbone for the wider API ingestion and sync workflows that move student data into a warehouse.

Prerequisites

Confirm each of these before running the procedure — the script assumes all of them are in place.

The expected upstream shape is a Canvas OAuth token document: {"access_token": "...", "refresh_token": "...", "expires_in": 3600, "token_type": "Bearer"}. The full lifecycle this manager implements — proactive refresh on expiry plus reactive renewal on a mid-stream 401 — looks like this:

Canvas token refresh sequence: proactive expiry refresh and reactive 401 retry A sequence diagram across five participants — Sync job, TokenManager, Encrypted cache, Canvas OAuth, and Canvas REST API. The sync job calls the TokenManager, which reads the cached token; if it is expired or missing the manager posts to the OAuth token endpoint, receives a fresh access token, and persists it encrypted. It then calls the REST API with the Bearer token. On 200 OK the payload returns; on 401 Unauthorized the manager refreshes once, persists the new token, retries the request, and returns the payload to the sync job. Accent-coloured arrows mark the refresh moments. Sync job TokenManager Encrypted cache Canvas OAuth Canvas REST API alt [ token expired or missing ] alt [ 200 OK ] else [ 401 Unauthorized ] 1 · make_request(endpoint) 2 · read token + expires_at 3 · POST /login/oauth2/token 4 · access_token, expires_in 5 · encrypt + persist 6 · request with Bearer token 7 · payload 8 · refresh token 9 · new access_token 10 · encrypt + persist 11 · retry with new token 12 · payload 13 · response call return OAuth refresh moment

Step-by-step implementation

1. Choose the credential model. Canvas offers long-lived developer access tokens (single-institution scripts, no programmatic refresh) and OAuth 2.0 refresh-token flows (multi-tenant or unattended, silently renewable). Pick OAuth 2.0 for any pipeline that runs longer than a token’s lifetime — it is the only model that can renew itself without a human re-consenting.

2. Persist an absolute expiry timestamp, not a duration. Canvas returns expires_in (seconds from now), which is useless after a restart. On every refresh, compute expires_at = time.time() + expires_in and store that — deterministic expiry tracking is what lets a cold process know whether its cached token is still good.

3. Encrypt the cache at rest. Tokens are bearer credentials to student PII, so serialize them through Fernet before they touch disk. Requiring TOKEN_ENCRYPTION_KEY to exist (rather than generating a throwaway key) prevents an insecure fallback that would write plaintext tokens.

4. Refresh proactively with a safety buffer. Treat the token as expired a few minutes before its real expires_at (a 300-second buffer here). This stops a token that is valid at the start of a long paginated pull from expiring mid-cursor — the same hazard covered in pagination strategies for bulk exports.

5. Intercept 401 as a reactive backstop. Clock skew, an admin revocation, or a key rotation can invalidate a token the buffer thinks is fine. Catch 401, force one refresh, and retry the request a single time — refreshing on every 401 in a loop would hammer the OAuth endpoint, so the retry is deliberately bounded.

6. Wrap transient failures in bounded backoff. Use tenacity to retry the whole request on transport errors with exponential backoff, capped at three attempts. This composes with — but is distinct from — the 401 refresh path, and dovetails with the dedicated error and retry logic for sync jobs and Canvas API rate-limit handling you layer on top.

7. Strip secrets and PII from every log line. Never log the token, the request body, or a student_id. Audit trails carry only timestamps, endpoint paths, and status codes — the minimum that satisfies the FERPA compliance boundary without leaking identifiers.

Complete runnable code block

python
import os
import time
import json
import logging
import requests
from pathlib import Path
from cryptography.fernet import Fernet
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# FERPA-safe logging: never log tokens, PII, or request payloads.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.StreamHandler()],
)
logger = logging.getLogger("canvas_token_manager")


class CanvasTokenManager:
    def __init__(self, base_url: str, client_id: str, client_secret: str,
                 token_path: str = "/etc/edtech/canvas_tokens.enc") -> None:
        self.base_url = base_url.rstrip("/")
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_path = Path(token_path)

        # Require an explicit key so we never silently write plaintext tokens.
        enc_key = os.environ.get("TOKEN_ENCRYPTION_KEY")
        if not enc_key:
            raise EnvironmentError("TOKEN_ENCRYPTION_KEY environment variable is required.")
        self._fernet = Fernet(enc_key.encode("utf-8"))
        self._token_data: dict = {}
        self._load_cached_token()

    def _load_cached_token(self) -> None:
        if self.token_path.exists():
            try:
                decrypted = self._fernet.decrypt(self.token_path.read_bytes())
                self._token_data = json.loads(decrypted.decode("utf-8"))
                logger.info("Token cache loaded.")
            except Exception as exc:  # corrupt/rotated key -> force a refresh
                logger.error("Failed to decrypt token cache: %s", exc)
                self._token_data = {}

    def _save_token(self, token_data: dict) -> None:
        self._token_data = token_data
        encrypted = self._fernet.encrypt(json.dumps(token_data).encode("utf-8"))
        self.token_path.parent.mkdir(parents=True, exist_ok=True)
        self.token_path.write_bytes(encrypted)
        logger.info("Token cache updated.")

    def _is_expired(self, buffer_seconds: int = 300) -> bool:
        if not self._token_data:
            return True
        return time.time() >= (self._token_data.get("expires_at", 0) - buffer_seconds)

    def refresh_token(self) -> str:
        logger.info("Refreshing token via OAuth 2.0.")
        payload = {
            "grant_type": "refresh_token",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "refresh_token": self._token_data.get("refresh_token", ""),
        }
        resp = requests.post(f"{self.base_url}/login/oauth2/token", data=payload, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        # Store an absolute expiry so a cold restart can reason about validity.
        data["expires_at"] = time.time() + data.get("expires_in", 3600)
        # Canvas returns a fresh refresh_token only on some flows; keep the old one otherwise.
        if "refresh_token" not in data and "refresh_token" in self._token_data:
            data["refresh_token"] = self._token_data["refresh_token"]
        self._save_token(data)
        return data["access_token"]

    def get_valid_token(self) -> str:
        return self.refresh_token() if self._is_expired() else self._token_data["access_token"]

    @retry(
        retry=retry_if_exception_type(requests.exceptions.HTTPError),
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        reraise=True,
    )
    def make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        token = self.get_valid_token()
        headers = kwargs.pop("headers", {})
        headers["Authorization"] = f"Bearer {token}"
        url = f"{self.base_url}{endpoint}"

        resp = requests.request(method, url, headers=headers, **kwargs)

        # Reactive backstop: refresh once on a 401 the expiry buffer missed.
        if resp.status_code == 401:
            logger.warning("401 Unauthorized on %s — refreshing and retrying once.", endpoint)
            headers["Authorization"] = f"Bearer {self.refresh_token()}"
            resp = requests.request(method, url, headers=headers, **kwargs)

        resp.raise_for_status()
        return resp


if __name__ == "__main__":
    mgr = CanvasTokenManager(
        base_url="https://canvas.instructure.com",
        client_id=os.environ["CANVAS_CLIENT_ID"],
        client_secret=os.environ["CANVAS_CLIENT_SECRET"],
    )
    r = mgr.make_request("GET", "/api/v1/users/self")
    logger.info("Authenticated as user id=%s (status %s)", r.json().get("id"), r.status_code)

Verification and output validation

Confirm the manager works end-to-end without exposing any secret:

  • First-run write. After the initial refresh_token(), assert the cache exists and is opaque: Path("/etc/edtech/canvas_tokens.enc").read_bytes()[:10] should start with gAAAAA (the Fernet version byte), never readable JSON.
  • Expiry math. assert mgr._token_data["expires_at"] > time.time() immediately after a refresh, and assert mgr._is_expired(buffer_seconds=0) is False.
  • Identity check. GET /api/v1/users/self returns HTTP 200 and a JSON body with an integer id — proof the bearer header is accepted.
  • Cold-start reuse. Construct a second CanvasTokenManager in a fresh process; it should load the cache and serve the same access_token without hitting the OAuth endpoint (no “Refreshing token” log line).
  • Forced renewal. Set mgr._token_data["expires_at"] = 0, call get_valid_token(), and confirm exactly one refresh occurs and a new access_token string is returned.

Troubleshooting

  • 401 Unauthorized on every request, even after refresh. The refresh_token itself is invalid or revoked (Canvas refresh tokens are single-tenant and can be cleared by an admin). Re-run the one-time authorization code flow to mint a new refresh token; the manager cannot bootstrap consent on its own.
  • invalid_grant from /login/oauth2/token. Either the refresh_token rotated and you persisted the wrong one, or client_secret is stale after a key regeneration. Verify the stored refresh_token matches the latest grant and that the secret in your secrets manager is current.
  • cryptography.fernet.InvalidToken on load. TOKEN_ENCRYPTION_KEY changed between writes, or the cache file is corrupt. The code already degrades to an empty cache and forces a refresh — rotate deliberately by deleting the file and re-authorizing, not by swapping keys in place.
  • KeyError: 'access_token' after a refresh. Canvas returned an error body with HTTP 200 is rare, but a 400/401 that slipped past raise_for_status is not — confirm grant_type is exactly refresh_token and that the developer key is enabled for this account.
  • 429 Too Many Requests during a refresh storm. Multiple workers each constructing their own manager will each refresh. Instantiate one manager per process and share it across threads; for rate pacing on the data calls, layer in Canvas API rate-limit handling.
  • Token expires mid-pagination on a long export. The 300-second buffer is too small for a multi-minute cursor walk. Widen buffer_seconds, or call get_valid_token() at the top of each page rather than once per job — see pagination strategies for bulk exports.

Part of: Python Requests for LMS APIs