Building an OAuth2 Flow in FastAPI

OAuth2 is the industry standard for secure API authentication and authorization. This guide covers the Authorization Code Grant flow—the most secure and commonly used type for web applications.

1. Understanding OAuth2 Basics

OAuth2 allows users to grant applications access to their data without sharing passwords. The flow involves three parties:

• Resource Owner: The user who owns the data
• Client Application: Your app (wants to access user data)
• Authorization Server: The service that manages permissions (e.g., Google, GitHub)

The Authorization Code Grant flow is best for:

• Web applications with a backend server
• Server-to-server communication
• High-security scenarios (tokens never exposed to browser)

Reference: OAuth 2.0 Official Specification

2. Authorization Code Grant Flow: Step by Step

Here's what happens when a user logs in with OAuth2:

User clicks "Login with Google"
↓
App redirects to Google's auth server
↓
Google shows login form
↓
User grants permissions
↓
Google redirects back with authorization code
↓
App backend exchanges code for access token
↓
App uses token to fetch user data
↓
User is logged in

The key advantage: the authorization code is short-lived and can only be used once. The access token is never exposed to the browser.

Reference: RFC 6749: Authorization Code Grant

3. Implementing OAuth2 in FastAPI

3.1 Installation

pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart httpx

3.2 Setting Up OAuth2 with Password Flow (Simple Example)

First, let's understand FastAPI's built-in OAuth2 support:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
app = FastAPI()
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class User(BaseModel):
username: str
email: str
full_name: str = None
disabled: bool = None
class UserInDB(User):
hashed_password: str
class Token(BaseModel):
access_token: str
token_type: str
# Fake database
fake_users_db = {
"john": {
"username": "john",
"full_name": "John Doe",
"email": "[email protected]",
"hashed_password": "$2b$12$...",  # hashed password
"disabled": False,
}
}
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def authenticate_user(username: str, password: str):
user = fake_users_db.get(username)
if not user or not verify_password(password, user["hashed_password"]):
return False
return user
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = fake_users_db.get(username)
if user is None:
raise credentials_exception
return user
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user["username"]}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user

4. Real OAuth2 with External Providers (Google Example)

To integrate with Google OAuth2, install:

pip install google-auth-httplib2 google-auth-oauthlib

4.1 Authorization Endpoint

from urllib.parse import urlencode
GOOGLE_CLIENT_ID = "your-google-client-id"
GOOGLE_CLIENT_SECRET = "your-google-client-secret"
REDIRECT_URI = "http://localhost:8000/auth/callback"
@app.get("/auth/login")
async def login_google():
params = {
"client_id": GOOGLE_CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"scope": "openid email profile",
"response_type": "code",
"access_type": "offline",
}
auth_url = "https://accounts.google.com/o/oauth2/v2/auth?" + urlencode(params)
return {"auth_url": auth_url}

4.2 Callback Endpoint (Exchange Code for Token)

import httpx
from pydantic import BaseModel
class TokenResponse(BaseModel):
access_token: str
token_type: str
expires_in: int
refresh_token: str = None
@app.get("/auth/callback")
async def callback(code: str):
# Exchange authorization code for access token
token_url = "https://oauth2.googleapis.com/token"
payload = {
"client_id": GOOGLE_CLIENT_ID,
"client_secret": GOOGLE_CLIENT_SECRET,
"code": code,
"redirect_uri": REDIRECT_URI,
"grant_type": "authorization_code",
}
async with httpx.AsyncClient() as client:
response = await client.post(token_url, json=payload)
tokens = response.json()
# Fetch user info using access token
user_info_url = "https://openidconnect.googleapis.com/v1/userinfo"
headers = {"Authorization": f"Bearer {tokens['access_token']}"}
async with httpx.AsyncClient() as client:
user_response = await client.get(user_info_url, headers=headers)
user_info = user_response.json()
# Save user to database and create session
return {
"user": user_info,
"access_token": tokens["access_token"],
"token_type": "Bearer",
}

5. Refresh Tokens

Access tokens expire. Use refresh tokens to get new access tokens without re-authenticating:

@app.post("/token/refresh")
async def refresh_token(refresh_token: str):
# Verify refresh token
try:
payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
except JWTError:
raise HTTPException(status_code=401, detail="Invalid refresh token")
# Create new access token
access_token = create_access_token(
data={"sub": username},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
)
return {"access_token": access_token, "token_type": "bearer"}

6. Scopes: Granular Permissions

OAuth2 scopes limit what an app can do:

from fastapi.security import OAuth2PasswordBearer, Scopes
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="token",
scopes={
"read:users": "Read user data",
"write:posts": "Create and edit posts",
"admin": "Admin access",
}
)
async def get_current_user(token: str = Depends(oauth2_scheme)):
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
scopes = payload.get("scopes", [])
return {"username": username, "scopes": scopes}
@app.get("/posts")
async def read_posts(current_user = Depends(get_current_user)):
if "read:posts" not in current_user["scopes"]:
raise HTTPException(status_code=403, detail="Not enough permissions")
return [{"id": 1, "title": "Post 1"}]

7. Using Authlib (Simpler OAuth2)

For easier OAuth2 integration, use Authlib:

pip install authlib

from authlib.integrations.httpx_client import AsyncOAuth2Client
oauth_client = AsyncOAuth2Client(
client_id=GOOGLE_CLIENT_ID,
client_secret=GOOGLE_CLIENT_SECRET,
redirect_uri=REDIRECT_URI,
)
@app.get("/auth/authorize")
async def authorize():
uri, state = oauth_client.create_authorization_url(
"https://accounts.google.com/o/oauth2/v2/auth",
scope="openid email profile"
)
return {"authorization_url": uri, "state": state}
@app.get("/auth/callback")
async def handle_callback(code: str, state: str):
token = await oauth_client.fetch_token(
"https://oauth2.googleapis.com/token",
code=code,
state=state
)
return {"token": token}

Reference: Authlib Documentation

8. Security Best Practices

• Use HTTPS only: OAuth2 tokens must travel over encrypted connections
• Store secrets securely: Never hardcode API keys; use environment variables
• Validate state parameter: Prevent CSRF attacks by checking the state value matches
• Short-lived access tokens: 15-30 minutes is typical
• Secure refresh tokens: Store in httponly, secure cookies or secure storage
• Validate JWT signatures: Always verify the token signature before trusting it
• Use scopes properly: Request only the permissions you need

9. Testing OAuth2 Endpoints

from fastapi.testclient import TestClient
client = TestClient(app)
def test_login():
response = client.post(
"/token",
data={"username": "john", "password": "secret"}
)
assert response.status_code == 200
assert "access_token" in response.json()
def test_protected_endpoint():
# Get token
token_response = client.post(
"/token",
data={"username": "john", "password": "secret"}
)
token = token_response.json()["access_token"]
# Use token to access protected endpoint
headers = {"Authorization": f"Bearer {token}"}
response = client.get("/users/me", headers=headers)
assert response.status_code == 200
assert response.json()["username"] == "john"

10. Common OAuth2 Providers

• Google: Google OAuth2
• GitHub: GitHub OAuth Apps
• Microsoft: Microsoft OAuth2
• Facebook: Facebook Login

11. Summary Checklist

• Use Authorization Code Grant for web apps
• Never expose access tokens to the browser
• Validate authorization codes immediately
• Use short-lived access tokens (15-30 minutes)
• Store refresh tokens securely (httponly cookies)
• Validate JWT signatures on every request
• Use state parameter to prevent CSRF
• Request only necessary scopes
• Always use HTTPS in production
• Log failed authentication attempts

Further Reading

• FastAPI OAuth2 with Scopes
• RFC 6749: OAuth 2.0 Authorization Framework
• Authlib OAuth 2.0 Guide
• python-jose Documentation
• OWASP: CSRF Attacks
• Auth0: Authentication & Authorization