Python Fastapi Jwt Oauth2 Authentication

Written by

in

Building a secure, scalable API is a top priority for modern Python developers, and FastAPI has quickly become the go‑to framework for this task. When you combine FastAPI’s async‑first design with JSON Web Tokens (JWT) and the OAuth2 standard, you get a powerful authentication system that’s both easy to implement and highly robust. In this guide, we’ll walk through every step required to set up JWT‑based OAuth2 authentication in a FastAPI project—covering everything from project scaffolding to token generation, route protection, and best‑practice security tips. Whether you’re building a microservice, a mobile backend, or a full‑stack web app, the concepts and code snippets below will give you a production‑ready foundation.

Why Choose FastAPI for Authentication?

FastAPI shines for authentication for three main reasons:

  • Declarative security dependencies: You can inject authentication logic directly into route handlers using Depends, keeping your code clean and testable.
  • Automatic OpenAPI documentation: FastAPI generates interactive Swagger UI and ReDoc docs that include your OAuth2 flow out of the box.
  • Performance: Built on Starlette and Pydantic, FastAPI handles high‑throughput, async I/O with minimal overhead—crucial for token validation at scale.

Understanding JWT and OAuth2

What is a JWT?

A JSON Web Token (JWT) is a compact, URL‑safe string that contains three parts: a header, a payload (claims), and a signature. The payload can store user identifiers, roles, and expiration timestamps, while the signature ensures the token’s integrity using a secret key or RSA private key.

How OAuth2 Fits In

OAuth2 is an authorization framework that defines several “grant types” for obtaining access tokens. In FastAPI, the most common is the Password Grant (also called Resource Owner Password Credentials). This flow lets a client exchange a username and password for a JWT access token, which can then be sent in the Authorization: Bearer header for protected endpoints.

Project Setup: Installing Dependencies

First, create a fresh virtual environment and install the required packages:

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install fastapi uvicorn python‑jose[cryptography] passlib[bcrypt] pydantic

We’ll use python-jose for JWT encoding/decoding and passlib for secure password hashing.

Defining the Data Models

FastAPI leverages Pydantic models for request validation and response serialization. Below are the core models for users, token payloads, and OAuth2 responses.

from pydantic import BaseModel, EmailStr
from typing import Optional

class User(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

Setting Up the Security Scheme

FastAPI provides a convenient OAuth2PasswordBearer class that extracts the token from the Authorization header.

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

The tokenUrl points to the endpoint that issues JWTs (we’ll create it later).

Implementing Password Hashing

Never store plain‑text passwords. Use passlib with the bcrypt algorithm:

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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)

Creating the JWT Utility Functions

These helpers handle token creation and validation. Adjust the SECRET_KEY and ALGORITHM to suit your security policies.

from datetime import datetime, timedelta
from jose import JWTError, jwt

SECRET_KEY = "YOUR_SUPER_SECRET_KEY_CHANGE_ME"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise JWTError()
        return TokenData(username=username)
    except JWTError:
        return None

Mock Database and User Retrieval

For demonstration, we’ll use an in‑memory dictionary. In production, replace this with a real database query.

fake_users_db = {
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderland",
        "email": "alice@example.com",
        "hashed_password": get_password_hash("secret123"),
        "disabled": False,
    }
}

def get_user(db, username: str) -> Optional[UserInDB]:
    user_dict = db.get(username)
    if user_dict:
        return UserInDB(**user_dict)
    return None

def authenticate_user(db, username: str, password: str):
    user = get_user(db, username)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

Building the Token Endpoint (OAuth2 Password Flow)

This endpoint validates credentials and returns an access token (and optionally a refresh token).

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

app = FastAPI()

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_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(data={"sub": user.username})
    refresh_token = create_refresh_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

Protecting Routes with Dependency Injection

FastAPI’s Depends makes it straightforward to enforce authentication on any endpoint.

from fastapi import Security

async def get_current_user(token: str = Depends(oauth2_scheme)):
    token_data = decode_token(token)
    if not token_data:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    user = get_user(fake_users_db, token_data.username)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

async def get_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_active_user)):
    return current_user

Implementing Refresh Token Logic

Refresh tokens allow clients to obtain a new access token without re‑entering credentials. Below is a simple endpoint that validates a refresh token and issues a fresh access token.

class RefreshTokenRequest(BaseModel):
    refresh_token: str

@app.post("/token/refresh", response_model=Token)
async def refresh_access_token(request: RefreshTokenRequest):
    token_data = decode_token(request.refresh_token)
    if not token_data:
        raise HTTPException(status_code=401, detail="Invalid refresh token")
    new_access_token = create_access_token(data={"sub": token_data.username})
    return {"access_token": new_access_token, "token_type": "bearer"}

Testing the Authentication Flow

Use curl or an API client like Postman to verify each step:

  1. Obtain a token:
    curl -X POST "http://localhost:8000/token" -d "grant_type=password

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *