Implementing the OAuth2 Authorization Code Flow in FastAPI

Almost every commercial API product eventually has to act as an OAuth2 client: your service connects to your customer's Google Workspace, HubSpot, Shopify or GitHub account and works on their behalf. That is a different job from issuing credentials for your own API, which is covered in the parent Handling API Authentication in Python guide. This page resolves one specific decision — which OAuth2 flow to implement and how to hold the resulting tokens — and gives you the FastAPI routes, the storage schema, the refresh logic and the background sweep that keeps revoked grants from silently rotting your product.

The short answer: use the authorization code flow with PKCE, always, even when the provider says PKCE is optional and even though you have a confidential server-side client. It costs you fifteen lines and it removes an entire class of interception bug from your incident log. Everything below assumes an async-native FastAPI setup and httpx as your async HTTP client.

The flow, end to end

Four parties move through eight steps. Your API never sees the customer's password; the provider never sees your client secret in a browser. The two pieces of state you generate before the redirect — a random state value and a PKCE code verifier — are what make the round trip safe. The state proves the callback belongs to a session you started, defeating cross-site request forgery on the callback route. The verifier proves the process redeeming the code is the same process that requested it, defeating code interception through a malicious app or a leaky referrer.

Authorization code flow with PKCE, step by step Sequence diagram across four lanes — user browser, your FastAPI service, the provider authorization server, and your token store — showing the redirect with a code challenge and state, the consent step, the callback carrying a code, the token exchange with the verifier, and the encrypted write of access and refresh tokens. User browser Your FastAPI Provider auth Token store GET /oauth/connect 302 + challenge + state login + consent screen 302 back with code + state GET /oauth/callback POST /token + verifier access + refresh tokens encrypt + store grant

Here are the two routes. They depend on Starlette's SessionMiddleware for the short-lived verifier and state — a signed cookie is the cheapest correct place to hold them, because the values live for about ninety seconds and never need to survive a deploy.

Python
# app/oauth_client.py
import base64
import hashlib
import os
import secrets
from urllib.parse import urlencode

import httpx
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import RedirectResponse

router = APIRouter(prefix="/oauth", tags=["oauth"])

AUTHORIZE_URL = os.getenv("PROVIDER_AUTHORIZE_URL", "")
TOKEN_URL = os.getenv("PROVIDER_TOKEN_URL", "")
CLIENT_ID = os.getenv("PROVIDER_CLIENT_ID", "")
CLIENT_SECRET = os.getenv("PROVIDER_CLIENT_SECRET", "")
REDIRECT_URI = os.getenv("PROVIDER_REDIRECT_URI", "")
SCOPES = os.getenv("PROVIDER_SCOPES", "")
POST_CONNECT_URL = os.getenv("APP_POST_CONNECT_URL", "/")
HTTP_TIMEOUT = float(os.getenv("PROVIDER_HTTP_TIMEOUT", "10"))


def _b64url(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()


@router.get("/connect")
async def connect(request: Request) -> RedirectResponse:
    verifier = _b64url(secrets.token_bytes(64))
    challenge = _b64url(hashlib.sha256(verifier.encode()).digest())
    state = secrets.token_urlsafe(32)
    request.session["pkce_verifier"] = verifier
    request.session["oauth_state"] = state

    params = {
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "scope": SCOPES,
        "state": state,
        "code_challenge": challenge,
        "code_challenge_method": "S256",
    }
    return RedirectResponse(f"{AUTHORIZE_URL}?{urlencode(params)}", status_code=302)

The callback route does three things in order: reject provider-side errors, verify state with a constant-time comparison, then redeem the code. Never skip the constant-time compare — == on a secret leaks timing, and this is a two-word fix.

Python
@router.get("/callback")
async def callback(
    request: Request,
    code: str | None = None,
    state: str | None = None,
    error: str | None = None,
) -> RedirectResponse:
    match error:
        case None:
            pass
        case "access_denied":
            raise HTTPException(status_code=400, detail="Connection declined by user")
        case _:
            raise HTTPException(status_code=400, detail=f"Provider error: {error}")

    expected = request.session.pop("oauth_state", None)
    verifier = request.session.pop("pkce_verifier", None)
    if not (code and state and expected and verifier):
        raise HTTPException(status_code=400, detail="Incomplete OAuth callback")
    if not secrets.compare_digest(state, expected):
        raise HTTPException(status_code=400, detail="State mismatch")

    payload = {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": REDIRECT_URI,
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "code_verifier": verifier,
    }
    async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
        response = await client.post(
            TOKEN_URL, data=payload, headers={"Accept": "application/json"}
        )
    if response.status_code != 200:
        raise HTTPException(status_code=502, detail="Token exchange failed")

    await save_grant(user_id=request.session["user_id"], data=response.json())
    return RedirectResponse(POST_CONNECT_URL, status_code=303)

Two details that bite people in production. First, redirect_uri must match the registered value byte for byte, including the trailing slash and the scheme — a staging environment on http://localhost and production on https:// need two separate registered URIs, not one wildcard. Second, authorization codes are single-use and usually expire in sixty seconds, so a user who double-clicks the consent button generates a second callback that fails; return a friendly "already connected" page rather than a 400 when the state is already consumed.

Storing tokens without leaking them

A refresh token is a long-lived bearer credential for someone else's account. Treat it like a password, not like a cache entry. One row per user per provider, both tokens encrypted at rest with a key that lives in your secret manager rather than your database, and a unique constraint that makes reconnecting an upsert instead of a duplicate.

Shape of the stored OAuth grant row An annotated database row for the oauth_grants table listing user and provider under a unique constraint, encrypted access and refresh token columns, an expiry timestamp and a status column, with callouts explaining encryption at rest and refreshing 300 seconds early. oauth_grants user_id, provider unique access_token_enc bytea refresh_token_enc bytea expires_at timestamptz status active | revoked One row per user and provider — upsert it Encrypt both tokens before they hit disk Refresh 300 s before this, never after it

The model below uses SQLAlchemy 2.0 typed mappings, which pair well with async database access under an async engine. Fernet gives you authenticated symmetric encryption in one line; if you already run a KMS, swap the two helper calls and keep the rest.

Python
# app/models.py
import os
from datetime import datetime, timedelta, timezone

from cryptography.fernet import Fernet
from sqlalchemy import String, UniqueConstraint
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

FERNET = Fernet(os.getenv("TOKEN_ENC_KEY", Fernet.generate_key().decode()).encode())


class Base(DeclarativeBase):
    pass


class OAuthGrant(Base):
    __tablename__ = "oauth_grants"
    __table_args__ = (UniqueConstraint("user_id", "provider", name="uq_user_provider"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[str] = mapped_column(String(64), index=True)
    provider: Mapped[str] = mapped_column(String(32))
    access_token_enc: Mapped[bytes]
    refresh_token_enc: Mapped[bytes | None]
    scope: Mapped[str] = mapped_column(String(512), default="")
    expires_at: Mapped[datetime]
    status: Mapped[str] = mapped_column(String(16), default="active")

    def set_tokens(self, data: dict) -> None:
        self.access_token_enc = FERNET.encrypt(data["access_token"].encode())
        if refresh := data.get("refresh_token"):
            self.refresh_token_enc = FERNET.encrypt(refresh.encode())
        ttl = int(data.get("expires_in", os.getenv("PROVIDER_DEFAULT_TTL", "3600")))
        self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=ttl)
        self.scope = data.get("scope", self.scope)

Note the if refresh := data.get("refresh_token") guard. Many providers return a refresh token only on the first authorization and omit it from every subsequent refresh response. Overwriting the column with None there is the single most common way builders lock themselves out of a customer account — the access token expires an hour later and there is nothing left to refresh with.

Refreshing early, under a lock

Refresh proactively inside a skew window, not reactively on a 401. If you wait for the provider to reject you, every in-flight request for that customer fails at once, and you will spend an afternoon debugging 401 errors that were entirely predictable. Three hundred seconds is the right window for a one-hour token: long enough to absorb clock drift and a retry, short enough that you refresh once per hour per customer instead of constantly.

Access token lifetime and the refresh skew window A timeline of a 3600 second access token showing requests served from the cached token for the first 3300 seconds, then a 300 second skew window in which the background job refreshes before expiry. Served from the cached access token — zero cost Sweep job refreshes in the 300 s skew window access token valid — 3600 s 0 s 1800 s 3300 s 3600 s Refresh after 3600 s and every in-flight call 401s at once.

Concurrency is the real hazard here. Two workers that notice the same expiring token at the same moment will both call the token endpoint, and providers that use refresh token rotation will invalidate the first response when the second arrives — you end up with a stored token the provider already retired. A SELECT ... FOR UPDATE on the grant row serialises it for free, which is why this function takes a session and runs inside a transaction.

Python
# app/tokens.py
import os
from datetime import datetime, timedelta, timezone

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import FERNET, OAuthGrant
from app.oauth_client import CLIENT_ID, CLIENT_SECRET, HTTP_TIMEOUT, TOKEN_URL

SKEW = int(os.getenv("OAUTH_REFRESH_SKEW_SECONDS", "300"))


class GrantRevoked(Exception):
    """The customer or provider withdrew this authorization."""


async def valid_access_token(
    session: AsyncSession, user_id: str, provider: str
) -> str:
    stmt = (
        select(OAuthGrant)
        .where(OAuthGrant.user_id == user_id, OAuthGrant.provider == provider)
        .with_for_update()
    )
    grant = await session.scalar(stmt)
    if grant is None or grant.status != "active":
        raise GrantRevoked(f"{provider} not connected for {user_id}")

    deadline = grant.expires_at - timedelta(seconds=SKEW)
    if deadline > datetime.now(timezone.utc):
        return FERNET.decrypt(grant.access_token_enc).decode()

    if grant.refresh_token_enc is None:
        grant.status = "revoked"
        await session.commit()
        raise GrantRevoked("no refresh token stored")

    payload = {
        "grant_type": "refresh_token",
        "refresh_token": FERNET.decrypt(grant.refresh_token_enc).decode(),
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    }
    async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
        response = await client.post(TOKEN_URL, data=payload)

    match response.status_code:
        case 200:
            grant.set_tokens(response.json())
            await session.commit()
            return FERNET.decrypt(grant.access_token_enc).decode()
        case 400 | 401:
            grant.status = "revoked"
            await session.commit()
            raise GrantRevoked(response.json().get("error", "invalid_grant"))
        case _:
            response.raise_for_status()
            raise GrantRevoked("unreachable")

A 5xx from the token endpoint should be retried rather than treated as revocation — wrap the call with tenacity-based retries and an exponential backoff. Only invalid_grant on a 400 is a permanent verdict.

Handling revoked grants in a background job

Grants die quietly. A customer removes your app from their provider's security page, an admin rotates the workspace, a password change invalidates every session — and your product finds out hours later when a sync fails. Fix that with a scheduled sweep that touches every grant inside its skew window and downgrades anything the provider rejects. Run it on a worker, not in the request path; the APScheduler versus Celery beat trade-off and the broader Celery, RQ and arq comparison both apply. For a job this small, arq on the Redis you already pay for is the cheapest option.

Grant lifecycle state machine State machine showing an active grant moving into refreshing when under 300 seconds remain, returning to active on success, moving to revoked on an invalid_grant response, then prompting the customer to reconnect which restores the active state. under 300 s left invalid_grant active tokens fresh refreshing row lock held revoked syncs paused reconnect prompt email plus in-app banner 200 OK, tokens rotated notify customer re-approves the scopes
Python
# app/jobs.py
import os

from arq import cron
from arq.connections import RedisSettings
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.models import OAuthGrant
from app.tokens import GrantRevoked, valid_access_token

engine = create_async_engine(os.getenv("DATABASE_URL", ""), pool_pre_ping=True)
Session = async_sessionmaker(engine, expire_on_commit=False)
BATCH = int(os.getenv("OAUTH_SWEEP_BATCH", "200"))


async def sweep_grants(ctx) -> dict[str, int]:
    refreshed = revoked = 0
    async with Session() as session:
        ids = (
            await session.scalars(
                select(OAuthGrant.id)
                .where(OAuthGrant.status == "active")
                .order_by(OAuthGrant.expires_at)
                .limit(BATCH)
            )
        ).all()

    for grant_id in ids:
        async with Session.begin() as session:
            grant = await session.get(OAuthGrant, grant_id)
            try:
                await valid_access_token(session, grant.user_id, grant.provider)
                refreshed += 1
            except GrantRevoked:
                revoked += 1
                await ctx["redis"].enqueue_job(
                    "notify_reconnect", grant.user_id, grant.provider
                )
    return {"refreshed": refreshed, "revoked": revoked}


class WorkerSettings:
    redis_settings = RedisSettings.from_dsn(os.getenv("REDIS_URL", "redis://localhost"))
    cron_jobs = [cron(sweep_grants, minute={0, 15, 30, 45})]
    functions = [sweep_grants]

Emit the returned counters through your logger so a spike in revocations shows up on a dashboard — structured logging with structlog makes that a one-line change. A sudden climb in the revoked count is usually a provider policy change, not customer churn, and you want to know within an hour.

When to choose this flow, and when not to

The authorization code flow is the right default whenever a human must approve access to their own account. It is the wrong tool when no human is involved.

FlowChoose it whenSkip it when
Code + PKCEA customer approves access to their accountNo end user is present
Client credentialsYour server owns the dataPer-user scopes matter
Device codeCLI or TV with no browserYou control the browser
ImplicitNeverAlways — it is deprecated

Scope selection is the other half of this decision, and it is commercial rather than technical. Ask for the narrowest scope set that makes your feature work, because an over-broad consent screen kills conversion at exactly the moment a trial user is deciding whether to trust you. Requesting write access to a customer's entire mailbox when you only need to read message headers doubles the drop-off on that screen in my experience, and it also drags you into a security review at any company larger than about fifty people. Start narrow, then trigger a second authorization round only when the customer opts into a feature that genuinely needs more.

Cost matters here too. Each connected customer costs you one refresh call per hour, or roughly 720 calls a month — negligible against any provider quota, but not free if you are on a metered plan. Sweeping every fifteen minutes across ten thousand grants means four database scans an hour and about 10,000 outbound refreshes a day, which one small worker handles comfortably. If you are still at a few hundred customers, skip the sweep entirely and refresh lazily inside valid_access_token; add the job when a support ticket about a silently broken integration costs more than the worker.

Migrating from stored credentials or long-lived tokens

If your v1 shipped with customers pasting a personal access token into a settings field, move to OAuth2 in four steps and do not force a hard cutover.

  1. Add the oauth_grants table and both routes behind a feature flag. Existing token rows keep working untouched.
  2. Change your provider client to resolve a credential through one function that tries an OAuth grant first, then falls back to the legacy stored token.
  3. Show a "Reconnect with one click" banner to legacy customers. Most convert within two weeks because the flow is genuinely easier than copying a token.
  4. Set a deadline, email the stragglers twice, then delete the legacy column. The same staged approach works for your own credentials — see rotating API keys without downtime.

Builder verdict

Implement the authorization code flow with PKCE and skip every library that promises to do it for you. The whole client is about 150 lines of code you fully understand, and the alternatives — authlib integrations, provider-specific SDKs, hosted connector platforms at forty dollars per connected account per month — all hide the two things that actually break in production: refresh concurrency and revocation detection. You will debug those anyway, so own them. The winner on shipping velocity is the code above plus a fifteen-minute sweep job: one afternoon to build, near-zero marginal cost per customer, and it turns a silent integration failure into an email your customer can act on. Handwritten beats hosted here, and it is not close.

FAQ

What does running the OAuth2 client cost per connected customer? About 720 refresh calls and 720 small database transactions a month, which rounds to nothing on any managed Postgres and Redis pair. The real cost is the worker: one 512 MB instance at roughly seven dollars a month handles tens of thousands of grants. Hosted connector platforms charge per connected account, so break-even against building it yourself arrives at about fifty customers.

How do I rotate the encryption key protecting stored refresh tokens? Use Fernet's MultiFernet with the new key first and the old key second, deploy, then run a background pass that decrypts and re-encrypts every row. Drop the old key from the list on the next deploy. Because tokens live behind one column, this never requires customers to reconnect and takes minutes even at scale.

Will migrating from stored personal access tokens break existing customers? Not if you keep the fallback path alive during the transition. Resolve credentials through a single function that prefers an OAuth grant and falls back to the legacy column, then delete the column only after the last customer converts. The risk lives entirely in hard cutovers.

How many revoked grants should I expect in normal operation? Between one and three percent of connected accounts per month for mature providers, mostly from staff changes and password resets rather than churn. Track that rate as its own metric: if it doubles in a week, the provider changed a policy or shortened a refresh token lifetime, and your reconnect emails need to go out before support tickets do.

Do I still need PKCE if my client secret never leaves the server? Yes. PKCE costs fifteen lines and defends against authorization code interception during the browser redirect, which a server-held secret does nothing about. Several providers now reject authorization requests without a code challenge, so implementing it also removes a future migration.