API Automation: OAuth & Tokens

By the end of this lesson you can:

• Authenticate with an API key (header or query) and a bearer token.
• Explain the OAuth client-credentials flow and exchange credentials for a token.
• Refresh an expired access token automatically.
• Keep every secret in the environment — never hard-coded, never logged, never committed.

API key      a single long secret string you send with each request
bearer token a token (often time-limited) sent in the Authorization header
OAuth        you exchange client credentials for a short-lived access token,
             then send THAT token — refreshing it when it expires

API key — load from the environment

import os, requests
from dotenv import load_dotenv

load_dotenv()
API_KEY = os.environ["API_KEY"]      # crashes loudly if missing — good

# some APIs want it as a header:
headers = {"X-API-Key": API_KEY}
requests.get("https://api.example.com/data", headers=headers, timeout=10)

# others as a query parameter:
requests.get("https://api.example.com/data",
             params={"api_key": API_KEY}, timeout=10)

The key comes from os.environ (set via .env in dev, real env vars in prod) — exactly the pattern from Lesson 8. It is never written in the code that you commit.

Bearer token

token = os.environ["ACCESS_TOKEN"]
headers = {"Authorization": f"Bearer {token}"}
requests.get("https://api.example.com/me", headers=headers, timeout=10)

The Authorization: Bearer <token> header is the most common scheme. The word "Bearer" is literal; the token follows.

OAuth client-credentials flow

For machine-to-machine automation (no human logging in), OAuth's client-credentials flow is standard: you trade a client_id + client_secret for a short-lived access_token.

def get_token() -> dict:
    resp = requests.post(
        "https://auth.example.com/oauth/token",
        data={
            "grant_type": "client_credentials",
            "client_id": os.environ["CLIENT_ID"],
            "client_secret": os.environ["CLIENT_SECRET"],
            "scope": "read:data",
        },
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()      # {"access_token": "...", "expires_in": 3600, ...}

You get back a token and how many seconds it lasts (expires_in). Use the token in the Authorization header until it expires, then request a fresh one.

Auto-refresh: a token manager

import time

class TokenManager:
    def __init__(self):
        self._token = None
        self._expires_at = 0.0

    def token(self) -> str:
        # refresh slightly early (60s buffer) to avoid edge-of-expiry failures
        if not self._token or time.time() > self._expires_at - 60:
            data = get_token()
            self._token = data["access_token"]
            self._expires_at = time.time() + data["expires_in"]
        return self._token

    def headers(self) -> dict:
        return {"Authorization": f"Bearer {self.token()}"}

auth = TokenManager()
requests.get("https://api.example.com/data",
             headers=auth.headers(), timeout=10)

The manager caches the token and transparently refreshes it before expiry — so the rest of your code just calls auth.headers() and never worries about token lifecycle.

Goal: a small, reusable client that combines auth, auto-refresh, the resilient session from Lesson 24, and a 401-triggered re-auth — the way real API clients are built.

import os, time, logging
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from dotenv import load_dotenv

load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("client")

class ApiClient:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip("/")
        self._token = None
        self._expires_at = 0.0
        self.session = self._make_session()

    def _make_session(self) -> requests.Session:
        retry = Retry(total=5, backoff_factor=1,
                      status_forcelist=[429, 500, 502, 503, 504],
                      respect_retry_after_header=True)
        s = requests.Session()
        s.mount("https://", HTTPAdapter(max_retries=retry))
        return s

    def _refresh(self) -> None:
        resp = self.session.post(
            os.environ["TOKEN_URL"],
            data={"grant_type": "client_credentials",
                  "client_id": os.environ["CLIENT_ID"],
                  "client_secret": os.environ["CLIENT_SECRET"]},
            timeout=10)
        resp.raise_for_status()
        data = resp.json()
        self._token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"]
        log.info("token refreshed (…%s), valid %ds",
                 self._token[-4:], data["expires_in"])

    def _auth_headers(self) -> dict:
        if not self._token or time.time() > self._expires_at - 60:
            self._refresh()
        return {"Authorization": f"Bearer {self._token}"}

    def get(self, path: str, **kwargs) -> dict:
        url = f"{self.base_url}/{path.lstrip('/')}"
        resp = self.session.get(url, headers=self._auth_headers(),
                                timeout=10, **kwargs)
        if resp.status_code == 401:          # token rejected — force refresh once
            log.warning("401 — refreshing token and retrying")
            self._token = None
            resp = self.session.get(url, headers=self._auth_headers(),
                                    timeout=10, **kwargs)
        resp.raise_for_status()
        return resp.json()

client = ApiClient("https://api.example.com/v1")
me = client.get("/me")
print("authenticated as", me.get("name"))

INFO token refreshed (…a1b2), valid 3600s
authenticated as Acme Service Account

Read the code

This client layers everything: secrets pulled from the environment, OAuth token fetched and cached with a 60-second refresh buffer, the resilient retry session underneath, and a 401 handler that refreshes once and retries (covering the case where a token is revoked early). Notice the log shows only the last four characters of the token — never the whole thing. client.get("/anything") now just works, authenticated and resilient. This is the foundation you'd build any real integration on.

Use a real free API that needs a key (e.g. OpenWeather, GitHub's token, or any you can sign up for) — store the key in .env, never in the code.

import os

def auth_headers() -> dict:
    token = os.getenv("ACCESS_TOKEN")
    if not token:
        raise RuntimeError("Set ACCESS_TOKEN in your .env")
    return {"Authorization": f"Bearer {token}"}

def masked(secret: str) -> str:
    if len(secret) <= 8:
        return "…"                       # too short to reveal anything
    return f"{secret[:7]}…{secret[-4:]}"

print(masked("sk-live-abcd1234567890"))  # sk-live…7890

APIs verify you with an API key (header or query), a bearer token (Authorization: Bearer …), or OAuth — where you exchange client_id/client_secret for a short-lived access token and refresh it before it expires. A small token manager caches and auto-refreshes so the rest of your code stays simple, and a 401 handler re-authenticates if a token is revoked early. The non-negotiable rule across all of it: secrets come from the environment, never the source code; never print or log them; git-ignore .env; and rotate any key the moment it's exposed.