Authentication — JWT and OAuth2

FastAPI combines OAuth2PasswordBearer with JWT to implement standard token-based authentication.

---

Installation

pip install python-jose[cryptography]  # JWT
pip install passlib[bcrypt]            # Password hashing

---

Password Hashing

from passlib.context import CryptContext

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


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


hashed = hash_password("my-secret-password")
print(verify_password("my-secret-password", hashed))  # True
print(verify_password("wrong-password", hashed))      # False

---

Creating and Verifying JWT Tokens

from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from pydantic import BaseModel

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


class TokenData(BaseModel):
    username: str | None = None


def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    payload = data.copy()
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    payload.update({"exp": expire})
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def decode_token(token: str) -> TokenData:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise ValueError("No user info in token")
        return TokenData(username=username)
    except JWTError:
        raise ValueError("Invalid token")

---

Full Authentication Flow

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from datetime import timedelta
from typing import Annotated

app = FastAPI()

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

# OAuth2: extract token from Authorization: Bearer <token> header
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

fake_users_db = {
    "alice": {
        "username": "alice",
        "hashed_password": pwd_context.hash("password123"),
        "email": "alice@example.com",
        "role": "admin",
        "disabled": False,
    },
    "bob": {
        "username": "bob",
        "hashed_password": pwd_context.hash("bobpass"),
        "email": "bob@example.com",
        "role": "user",
        "disabled": False,
    },
}


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)]
) -> dict:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        token_data = decode_token(token)
    except ValueError:
        raise credentials_exception

    user = fake_users_db.get(token_data.username)
    if user is None or user["disabled"]:
        raise credentials_exception
    return user


async def get_active_user(
    current_user: Annotated[dict, Depends(get_current_user)]
) -> dict:
    if current_user["disabled"]:
        raise HTTPException(status_code=400, detail="Inactive account")
    return current_user


async def require_admin(
    user: Annotated[dict, Depends(get_active_user)]
) -> dict:
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return user


CurrentUser = Annotated[dict, Depends(get_active_user)]
AdminUser = Annotated[dict, Depends(require_admin)]


@app.post("/auth/login")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = fake_users_db.get(form_data.username)
    if not user or not pwd_context.verify(form_data.password, user["hashed_password"]):
        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"]},
        expires_delta=timedelta(minutes=30),
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me")
async def read_me(current_user: CurrentUser):
    return {
        "username": current_user["username"],
        "email": current_user["email"],
        "role": current_user["role"],
    }


@app.get("/admin/users")
async def list_all_users(admin: AdminUser):
    return list(fake_users_db.values())

---

Refresh Token Pattern

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from datetime import timedelta

app = FastAPI()


class TokenPair(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"


def create_token_pair(username: str) -> TokenPair:
    access_token = create_access_token(
        {"sub": username, "type": "access"},
        expires_delta=timedelta(minutes=30),
    )
    refresh_token = create_access_token(
        {"sub": username, "type": "refresh"},
        expires_delta=timedelta(days=7),
    )
    return TokenPair(access_token=access_token, refresh_token=refresh_token)


@app.post("/auth/refresh")
async def refresh_token(refresh_token: str) -> TokenPair:
    try:
        token_data = decode_token(refresh_token)
    except ValueError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    return create_token_pair(token_data.username)

---

Summary

Component	Role
OAuth2PasswordBearer	Extract Bearer token
OAuth2PasswordRequestForm	Parse login form
passlib + bcrypt	Password hashing
python-jose + JWT	Token creation/verification
Depends(get_current_user)	Auth dependency injection
Refresh Token	Renew access token

The standard practice is to keep Access Tokens short-lived (30 min) and Refresh Tokens long-lived (7 days).