FastAPI JWT Authentication (Beginner JWT Flow)

fastapi==0.116.1
uvicorn[standard]==0.30.6
PyJWT==2.10.1
pydantic-settings==2.6.1

JWT_SECRET_KEY=replace-this-with-a-long-random-secret-key-32chars-min
JWT_ACCESS_TOKEN_EXPIRES=900
JWT_REFRESH_TOKEN_EXPIRES=604800

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    JWT_SECRET_KEY: str = "dev-secret-change-me"
    JWT_ACCESS_TOKEN_EXPIRES: int = 900
    JWT_REFRESH_TOKEN_EXPIRES: int = 604800

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

settings = Settings()

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env
uvicorn app.main:app --reload

from datetime import UTC, datetime, timedelta
from typing import Any

import jwt
from fastapi import HTTPException

from .settings import settings

ALGORITHM = "HS256"


def create_jwt_token(
    user_data: dict[str, Any],
    secret: str = settings.JWT_SECRET_KEY,
    expires_in: int = settings.JWT_ACCESS_TOKEN_EXPIRES,
    token_type: str = "access",
) -> str:
    payload = user_data.copy()
    payload["token_type"] = token_type
    payload["iat"] = datetime.now(UTC)
    payload["exp"] = datetime.now(UTC) + timedelta(seconds=expires_in)
    token = jwt.encode(payload, secret, algorithm=ALGORITHM)
    return token


def create_access_token(user_data: dict[str, Any]) -> str:
    return create_jwt_token(
        user_data=user_data,
        expires_in=settings.JWT_ACCESS_TOKEN_EXPIRES,
        token_type="access",
    )


def create_refresh_token(user_data: dict[str, Any]) -> str:
    return create_jwt_token(
        user_data=user_data,
        expires_in=settings.JWT_REFRESH_TOKEN_EXPIRES,
        token_type="refresh",
    )


def verify_jwt_token(token: str, secret: str = settings.JWT_SECRET_KEY) -> dict[str, Any]:
    try:
        payload = jwt.decode(token, secret, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

from typing import Any

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel

from .jwt_utils import create_access_token, create_refresh_token, verify_jwt_token

app = FastAPI(title="FastAPI JWT", version="0.1.0")
security = HTTPBearer(auto_error=False)

DUMMY_USERS: dict[str, dict[str, Any]] = {
    "thirdy": {"id": "user-1", "username": "thirdy", "password": "demo123", "role": "admin"},
    "maria": {"id": "user-2", "username": "maria", "password": "demo123", "role": "user"},
}

class LoginRequest(BaseModel):
    username: str
    password: str

class RefreshRequest(BaseModel):
    refresh_token: str

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

@app.post("/login", response_model=TokenResponse)
def login(payload: LoginRequest):
    user = DUMMY_USERS.get(payload.username)
    if not user or user["password"] != payload.password:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    token_payload = {"sub": user["id"], "username": user["username"], "role": user["role"]}
    return {
        "access_token": create_access_token(token_payload),
        "refresh_token": create_refresh_token(token_payload),
        "token_type": "bearer",
    }


def get_access_payload(credentials: HTTPAuthorizationCredentials | None = Depends(security)):
    if credentials is None:
        raise HTTPException(status_code=401, detail="Missing Bearer token")

    payload = verify_jwt_token(credentials.credentials)
    if payload.get("token_type") != "access":
        raise HTTPException(status_code=401, detail="Access token required")
    return payload

@app.get("/me")
def me(payload = Depends(get_access_payload)):
    return {
        "user_id": payload["sub"],
        "username": payload["username"],
        "role": payload["role"],
        "message": "You used a valid access token",
    }

@app.post("/refresh", response_model=TokenResponse)
def refresh_tokens(payload: RefreshRequest):
    decoded = verify_jwt_token(payload.refresh_token)
    if decoded.get("token_type") != "refresh":
        raise HTTPException(status_code=401, detail="Refresh token required")

    next_payload = {"sub": decoded["sub"], "username": decoded["username"], "role": decoded["role"]}
    return {
        "access_token": create_access_token(next_payload),
        "refresh_token": create_refresh_token(next_payload),
        "token_type": "bearer",
    }