OAuth2 Password Flow and Social Login Integration

FastAPI ships with built-in OAuth2 support that integrates directly with Swagger UI — when you use OAuth2PasswordBearer and OAuth2PasswordRequestForm, the /docs interface shows an “Authorize” button where developers can log in and test authenticated endpoints in the browser. This standard OAuth2 Password Flow is the most widely used grant type for first-party applications (where the application itself collects credentials). For social login, OAuth2’s Authorization Code Flow delegates authentication to providers like Google or GitHub and exchanges an authorization code for user information.

OAuth2 Password Flow with Swagger UI #

# app/auth/oauth2.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

app = FastAPI()

# OAuth2PasswordBearer:
# - Adds the "Authorize" button to Swagger UI
# - Reads Authorization: Bearer {token} header from requests
# - tokenUrl points to the login endpoint
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

@app.post("/auth/token")
def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db:        Session                   = Depends(get_db),
):
    """
    OAuth2 Password Flow login endpoint.
    Swagger UI calls this when the user clicks Authorize.
    Accepts application/x-www-form-urlencoded (not JSON).
    """
    user = authenticate_user(db, 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 = create_access_token(user.id, user.role)

    # OAuth2 spec requires this exact response format:
    return {
        "access_token": access_token,
        "token_type":   "bearer",
    }

# Now protected endpoints use the oauth2_scheme:
def get_current_user(
    token: str     = Depends(oauth2_scheme),
    db:    Session = Depends(get_db),
) -> User:
    payload = decode_access_token(token)
    user_id = int(payload["sub"])
    user    = db.get(User, user_id)
    if not user or not user.is_active:
        raise HTTPException(401, "Invalid credentials")
    return user

Social Login — Google OAuth2 Integration #

import httpx
import secrets
from fastapi import Request
from fastapi.responses import RedirectResponse

GOOGLE_CLIENT_ID     = settings.google_client_id
GOOGLE_CLIENT_SECRET = settings.google_client_secret
GOOGLE_REDIRECT_URI  = "https://yourapp.com/auth/google/callback"

# Step 1: Redirect user to Google's authorisation endpoint
@router.get("/auth/google")
def google_login(request: Request):
    # state prevents CSRF: client sends it, Google returns it in callback
    state = secrets.token_urlsafe(16)
    request.session["oauth_state"] = state   # requires SessionMiddleware

    google_auth_url = (
        "https://accounts.google.com/o/oauth2/v2/auth"
        f"?client_id={GOOGLE_CLIENT_ID}"
        f"&redirect_uri={GOOGLE_REDIRECT_URI}"
        f"&response_type=code"
        f"&scope=openid%20email%20profile"
        f"&state={state}"
    )
    return RedirectResponse(google_auth_url)

# Step 2: Google redirects back with ?code=... &state=...
@router.get("/auth/google/callback")
async def google_callback(
    code:    str,
    state:   str,
    request: Request,
    db:      Session = Depends(get_db),
):
    # Verify state to prevent CSRF
    if state != request.session.get("oauth_state"):
        raise HTTPException(400, "Invalid OAuth state")

    # Exchange code for tokens
    async with httpx.AsyncClient() as client:
        token_resp = await client.post("https://oauth2.googleapis.com/token", data={
            "client_id":     GOOGLE_CLIENT_ID,
            "client_secret": GOOGLE_CLIENT_SECRET,
            "code":          code,
            "redirect_uri":  GOOGLE_REDIRECT_URI,
            "grant_type":    "authorization_code",
        })
        token_data = token_resp.json()

        # Get user info using the Google access token
        user_info = await client.get(
            "https://www.googleapis.com/oauth2/v3/userinfo",
            headers={"Authorization": f"Bearer {token_data['access_token']}"},
        )
        google_user = user_info.json()

    # Find or create local user
    email = google_user["email"]
    user  = db.scalars(select(User).where(User.email == email)).first()
    if not user:
        user = User(
            email         = email,
            name          = google_user.get("name", email.split("@")[0]),
            password_hash = "",   # no password for OAuth users
            role          = "user",
        )
        db.add(user)
        db.flush()

    access_token = create_access_token(user.id, user.role)
    # Redirect to frontend with token (or set HttpOnly cookie)
    return RedirectResponse(f"/login-success?token={access_token}")

Common Mistakes #

Mistake 1 — Forgetting OAuth2PasswordRequestForm Depends() #

❌ Wrong — form data not parsed:

@app.post("/auth/token")
def login(form_data: OAuth2PasswordRequestForm):   # missing = Depends()!
    ...

✅ Correct — class dependencies always need Depends():

def login(form_data: OAuth2PasswordRequestForm = Depends()):   # ✓

Mistake 2 — Not validating OAuth state parameter #

❌ Wrong — CSRF attack possible:

@router.get("/auth/google/callback")
async def callback(code: str, db: Session = Depends(get_db)):
    # No state check — attacker can trick user into authenticating

✅ Correct — always verify the state parameter against the session value.

Mistake 3 — Returning the Google access_token to the client #

❌ Wrong — client gets Google token, not your app’s token:

return {"access_token": token_data["access_token"]}   # this is Google's token!

✅ Correct — create your own JWT for the user and return that.

Quick Reference #