get_token.py

"""Gets an access token from Subcontext and keeps it in memory until shortly
before it runs out. Copy get_token() into your own server code.

An access token lets your code use the Subcontext API for two hours. You get
one by sending your API application's client id and client secret to
Subcontext (an OAuth "client credentials" request). Run this on your server,
never in a web page, because it holds the secret.

    SUBCONTEXT_CLIENT_ID=... SUBCONTEXT_CLIENT_SECRET=... python3 get_token.py

SUBCONTEXT_CLIENT_ID and SUBCONTEXT_CLIENT_SECRET: from the Subcontext
  dashboard, Integrations, API applications. The client secret is shown only
  once, when you create the application (or rotate its secret).
SUBCONTEXT_SCOPE (optional): what the token may do, separated by spaces.
  Default "read". It can only include scopes ticked on the API application.
SUBCONTEXT_BASE_URL (optional): the Subcontext address. Default
  https://subcontext.com.

Needs Python 3.8 or later. Uses the standard library only.
"""

import json
import os
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request

BASE_URL = os.environ.get("SUBCONTEXT_BASE_URL", "https://subcontext.com").rstrip("/")
CLIENT_ID = os.environ.get("SUBCONTEXT_CLIENT_ID")
CLIENT_SECRET = os.environ.get("SUBCONTEXT_CLIENT_SECRET")
SCOPE = os.environ.get("SUBCONTEXT_SCOPE", "read")

# Ask for a new token this long before the current one runs out, so a token
# never expires in the middle of a request.
RENEW_EARLY_SECONDS = 300

_cached = None  # {"access_token": ..., "expires_at": seconds since the epoch}
_lock = threading.Lock()


def get_token():
    """Returns a token that is good for at least a few more minutes."""
    global _cached
    with _lock:
        if _cached and time.time() < _cached["expires_at"] - RENEW_EARLY_SECONDS:
            return _cached["access_token"]
        _cached = _request_token()
        return _cached["access_token"]


def forget_token():
    """Forget the cached token, for example after the API answers 401."""
    global _cached
    with _lock:
        _cached = None


def _request_token():
    if not CLIENT_ID or not CLIENT_SECRET:
        raise RuntimeError("Set SUBCONTEXT_CLIENT_ID and SUBCONTEXT_CLIENT_SECRET "
                           "(see the comment at the top of this file).")
    form = urllib.parse.urlencode({
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": SCOPE,
        # What the token is for: the Subcontext API. Required.
        "resource": BASE_URL + "/api/v1",
    }).encode()
    request = urllib.request.Request(
        BASE_URL + "/oauth/token", data=form, method="POST",
        headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"})
    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            body = json.load(response)
    except urllib.error.HTTPError as error:
        try:
            details = json.load(error)
        except ValueError:
            details = {}
        message = details.get("error_description") or details.get("error") or "no details"
        raise RuntimeError("Token request failed (%s): %s" % (error.code, message)) from None
    return {"access_token": body["access_token"], "expires_at": time.time() + body["expires_in"]}


if __name__ == "__main__":
    # Get a token twice and show that the second one came from memory.
    try:
        first = get_token()
    except (RuntimeError, urllib.error.URLError) as error:
        print(error, file=sys.stderr)
        sys.exit(1)
    print("Got a token:", first[:8] + "...")
    print("It runs out at", time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(_cached["expires_at"])))
    second = get_token()
    print("Asked again: the same token came from memory, no new request." if second == first
          else "Asked again: got a new token.")