FastAPI Authentication Guide 2026 | JWT, OAuth2, API Keys & Google Login

pip install python-jose[cryptography] passlib[bcrypt] pydantic-settings

# app/auth/config.py
from pydantic_settings import BaseSettings
 
class AuthSettings(BaseSettings):
    JWT_SECRET: str
    JWT_ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
 
auth_settings = AuthSettings()

# app/auth/jwt.py
from datetime import datetime, timedelta
from jose import jwt, JWTError
from .config import auth_settings
 
def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=auth_settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire, "type": "access"})
    return jwt.encode(to_encode, auth_settings.JWT_SECRET, algorithm=auth_settings.JWT_ALGORITHM)
 
def create_refresh_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=auth_settings.REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, auth_settings.JWT_SECRET, algorithm=auth_settings.JWT_ALGORITHM)
 
def decode_token(token: str) -> dict:
    return jwt.decode(token, auth_settings.JWT_SECRET, algorithms=[auth_settings.JWT_ALGORITHM])

# app/auth/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
 
from .jwt import decode_token
from .service import get_user_by_id
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
 
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_token(token)
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = await get_user_by_id(user_id)
    if user is None:
        raise credentials_exception
    return user
 
async def get_current_active_user(current_user = Depends(get_current_user)):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# app/auth/router.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
 
from .jwt import create_access_token, create_refresh_token
from .service import authenticate_user
from .schemas import Token, UserCreate, UserResponse
 
router = APIRouter(prefix="/auth", tags=["Authentication"])
 
@router.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = await authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect credentials")
    
    return {
        "access_token": create_access_token({"sub": str(user.id)}),
        "refresh_token": create_refresh_token({"sub": str(user.id)}),
        "token_type": "bearer"
    }
 
@router.post("/register", response_model=UserResponse)
async def register(user_data: UserCreate):
    return await create_user(user_data)

# app/auth/security.py
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)

from fastapi import Security, HTTPException
from fastapi.security import APIKeyHeader
 
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
 
async def get_api_key(api_key: str = Security(api_key_header)):
    if not api_key:
        raise HTTPException(status_code=401, detail="API Key required")
    
    key_data = await verify_api_key(api_key)
    if not key_data:
        raise HTTPException(status_code=403, detail="Invalid API Key")
    return key_data
 
@app.get("/api/protected")
async def protected_endpoint(api_key = Depends(get_api_key)):
    return {"message": "Authenticated via API Key"}

from authlib.integrations.starlette_client import OAuth
 
oauth = OAuth()
oauth.register(
    name='google',
    client_id=settings.GOOGLE_CLIENT_ID,
    client_secret=settings.GOOGLE_CLIENT_SECRET,
    authorize_url='https://accounts.google.com/o/oauth2/auth',
    token_url='https://oauth2.googleapis.com/token',
    userinfo_endpoint='https://openidconnect.googleapis.com/v1/userinfo',
    client_kwargs={'scope': 'openid email profile'},
)
 
@router.get("/login/google")
async def login_google(request: Request):
    redirect_uri = request.url_for('auth_callback')
    return await oauth.google.authorize_redirect(request, redirect_uri)
 
@router.get("/callback")
async def auth_callback(request: Request):
    token = await oauth.google.authorize_access_token(request)
    user_info = token.get('userinfo')
    # Create or update user, generate JWT
    return {"access_token": create_access_token({"sub": user_info["email"]})}

from enum import Enum
from functools import wraps
 
class Role(str, Enum):
    USER = "user"
    ADMIN = "admin"
    SUPERADMIN = "superadmin"
 
def require_role(allowed_roles: list[Role]):
    async def role_checker(current_user = Depends(get_current_user)):
        if current_user.role not in allowed_roles:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return current_user
    return role_checker
 
@app.get("/admin/users")
async def list_users(user = Depends(require_role([Role.ADMIN, Role.SUPERADMIN]))):
    return await get_all_users()

from starlette.middleware.sessions import SessionMiddleware
 
app.add_middleware(SessionMiddleware, secret_key=settings.SECRET_KEY)
 
@router.post("/login")
async def login(request: Request, credentials: LoginRequest):
    user = await authenticate_user(credentials.email, credentials.password)
    if user:
        request.session["user_id"] = str(user.id)
        return {"message": "Logged in"}
    raise HTTPException(status_code=401)
 
async def get_session_user(request: Request):
    user_id = request.session.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401)
    return await get_user_by_id(user_id)