Build a JWT Authentication API

Build a production-grade JWT auth system with FastAPI and refresh token rotation.

Introduction

30 Second Summary

Every time you log in to an app, something behind the scenes verifies who you are and decides what you can access. That "something" is an authentication system, and building one correctly is one of the most critical skills in backend development.

In this project, you will build a production-grade JWT authentication API with FastAPI that issues access and refresh tokens, hashes passwords securely, and protects endpoints with layered dependencies.

What You'll Build

A FastAPI application that implements JWT-based authentication with short-lived access tokens, refresh token rotation with reuse detection, and a 3-layer dependency chain for protecting routes.

By the end of this project, you'll have:

  • A FastAPI application with user registration, login, and token issuance endpoints.
  • A refresh token rotation system with family-based reuse detection persisted in SQLite via SQLAlchemy.
  • Protected API endpoints using a layered dependency chain that separates token validation, user loading, and permission enforcement.
  • Secret Mission: Add role-based access control with an admin role that can list all users and revoke sessions.

Are there any prerequisites?

This project assumes familiarity with Python and FastAPI basics, including routing, Pydantic models, and dependency injection. You do not need prior experience with authentication or JWTs.

Before We Start

Authentication is the backbone of every production API. Without it, anyone could access your users' data, run costly model inferences, or impersonate other users.

In this step, you'll set up the project environment for building a JWT authentication system with FastAPI. You'll install the required packages and create the file structure for your API.

In this step, get ready to:
  • Create a project directory with a Python virtual environment.
  • Install FastAPI, PyJWT, pwdlib, and SQLAlchemy.
  • Generate a cryptographic secret key and create the project files.
Create the project directory and virtual environment

Every Python project needs its own isolated environment so package versions don't clash with other projects on your machine.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir jwt-auth-api
cd jwt-auth-api
  • Create a Python virtual environment by running:
python3 -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here won't affect your system Python or other projects.

  • Activate the virtual environment:

macOS

source venv/bin/activate

You should see (venv) appear at the start of your terminal prompt.

Windows

.\venv\Scripts\Activate.ps1

You should see (venv) appear at the start of your terminal prompt.

Don't see (venv) in your prompt?

On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy RemoteSigned -Scope CurrentUser in PowerShell, then try the activation command again.

On macOS, make sure you typed source before the path. Without it, the command won't activate the environment.

help me activate my Python virtual environment

Install dependencies and generate the secret key

Your authentication API relies on four packages. Each handles a specific layer of the security stack.

  • Install all project dependencies by running:
pip install "fastapi[standard]" pyjwt "pwdlib[argon2]" sqlalchemy

What are these packages?

  • fastapi[standard] is the web framework for building your API, bundled with a development server and validation tools.
  • pyjwt handles encoding and decoding JSON Web Tokens.
  • pwdlib[argon2] provides password hashing using the Argon2 algorithm, the current best practice for storing credentials.
  • sqlalchemy is the ORM for managing your database tables and queries.
  • Confirm all packages installed correctly by running:
python -c "import fastapi; import jwt; import pwdlib; import sqlalchemy; print('All imports OK')"

You should see All imports OK printed in your terminal.

Seeing an ImportError?

Make sure (venv) appears in your prompt. If it doesn't, your packages installed to the system Python instead of the virtual environment. Re-activate the venv and run the pip install command again.

help me fix an ImportError when verifying my JWT auth project packages

  • Generate a cryptographic secret key for signing your tokens by running:
openssl rand -hex 32
  • Copy the output string and save it somewhere safe (a notes app or sticky note). You'll paste this into your code in a later step.

What is this secret key for?

This random 256-bit string is what your server uses to sign and verify JWTs. Anyone who has this key can forge valid tokens, so in production you would store it in an environment variable rather than in source code.

Create the project file structure

Your authentication API is split across five files. Each one handles a single responsibility in the system.

  • Create all five project files by running:
touch main.py database.py models.py schemas.py auth.py
  • Confirm the files exist by running:
ls

You should see auth.py, database.py, main.py, models.py, and schemas.py listed alongside your venv folder.

What will each file do?

  • main.py defines your API endpoints (register, login, refresh, protected routes).
  • database.py sets up the SQLAlchemy engine and database session.
  • models.py describes your database tables (User, RefreshToken).
  • schemas.py defines the request and response shapes using Pydantic.
  • auth.py holds password hashing, token creation, and the dependency chain for protecting routes.
  • Open the jwt-auth-api folder in VS Code by running:
code .

code command not found?

If your terminal doesn't recognize code, open VS Code manually. Press Cmd+Space (macOS) or the Windows key (Windows) to open search, type Visual Studio Code, and press Enter. Then click File > Open Folder and select the jwt-auth-api folder on your Desktop.

help me open my project folder in VS Code

Your environment is ready and all five files are in place. Next up, you'll build the database layer and define your User and RefreshToken models.

Set Up the Database and Models

Your project files are in place and all dependencies are installed. The next piece you need is a data layer that stores user credentials and active sessions.

In this step, you'll configure SQLAlchemy to connect to a local SQLite database and define two ORM models: one for users and one for refresh tokens.

In this step, get ready to:
  • Configure the SQLAlchemy engine and session factory in database.py.
  • Define the User and RefreshToken ORM models in models.py.
  • Verify all project dependencies import correctly.
Configure the database connection

SQLAlchemy needs three things to manage your database: an engine (the connection), a session factory (creates per-request sessions), and a base class (the parent for all your models).

  • Open database.py in VS Code and add the full database configuration by pasting this code:
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

# SQLite database file stored in the project directory
SQLALCHEMY_DATABASE_URL = "sqlite:///./auth.db"

# Create the database engine
# check_same_thread=False is required for SQLite with FastAPI
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# Session factory for creating database sessions
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


# Base class that all ORM models will inherit from
class Base(DeclarativeBase):
    pass


# Dependency that provides a database session per request
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

What does this code do?

  • SQLALCHEMY_DATABASE_URL points to a local SQLite file called auth.db that will be created in your project folder.
  • engine opens the connection. The check_same_thread setting is required because SQLite normally allows only one thread, but FastAPI handles requests across multiple threads.
  • SessionLocal is a factory. Each time you call it, you get a fresh database session scoped to a single request.
  • Base is the parent class all your ORM models will inherit from. SQLAlchemy uses it to track which tables exist.
  • get_db() is a FastAPI dependency. It opens a session, yields it to your endpoint, then closes it when the request finishes.
  • Save database.py.
  • Confirm the module loads by running this in your terminal:
python -c "from database import get_db, Base; print('database.py OK')"

You should see database.py OK printed with no errors.

Seeing a ModuleNotFoundError?

  • Make sure your virtual environment is active. You should see the environment name in parentheses at the start of your terminal prompt.
  • Run the command from inside your project directory (the folder containing database.py).
  • Check that you installed sqlalchemy in this virtual environment.

Help me debug my database.py import error

✔️ Awesome, I've got everything!

Great. Double check you've saved your file and let's move on to the models.

ⓧ I'd like to double check the full code

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./auth.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
Define the User and RefreshToken models

Your authentication system needs to persist two things: who the users are, and which refresh tokens are currently valid. Each of these becomes a SQLAlchemy model mapped to a database table.

  • Open models.py and add both models by pasting this code:
from datetime import datetime, timezone

from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    # Default role is "user", admin role unlocked in Secret Mission
    role = Column(String, default="user", nullable=False)
    is_active = Column(Boolean, default=True)

    # One user can have many refresh tokens (one per session)
    refresh_tokens = relationship("RefreshToken", back_populates="user")


class RefreshToken(Base):
    __tablename__ = "refresh_tokens"

    id = Column(Integer, primary_key=True, index=True)
    # Store a SHA-256 hash of the token, never the raw token
    token_hash = Column(String, unique=True, index=True, nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    # Revoked tokens are rejected on use
    revoked = Column(Boolean, default=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    expires_at = Column(DateTime, nullable=False)

    user = relationship("User", back_populates="refresh_tokens")

What does this code do?

The User model maps to a users table. It stores the username, email, a hashed password (never plaintext), a role for authorization, and an active flag.

The RefreshToken model maps to a refresh_tokens table. It stores a SHA-256 hash of the token (not the raw value), links to a user via user_id, and tracks whether the token has been revoked or expired.

The relationship() calls create a bidirectional link. From a User you can access user.refresh_tokens to see all their sessions. From a RefreshToken you can access token.user to find who owns it.

  • Save models.py.
  • Confirm both models load by running this in your terminal:
python -c "from models import User, RefreshToken; print('Models OK')"

You should see Models OK printed with no errors.

Seeing an ImportError or SyntaxError?

  • A SyntaxError means a typo in models.py. Check for missing commas between Column arguments or unmatched parentheses.
  • An ImportError on from database import Base means database.py has an issue. Go back and confirm the previous verification passed.
  • Make sure both files are saved in the same folder. Python resolves imports relative to the directory you run the command from.

Help me debug my models.py import error

✔️ Awesome, I've got everything!

Great. Double check you've saved your file and let's verify the full dependency stack.

ⓧ I'd like to double check the full code

from datetime import datetime, timezone

from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    role = Column(String, default="user", nullable=False)
    is_active = Column(Boolean, default=True)

    refresh_tokens = relationship("RefreshToken", back_populates="user")


class RefreshToken(Base):
    __tablename__ = "refresh_tokens"

    id = Column(Integer, primary_key=True, index=True)
    token_hash = Column(String, unique=True, index=True, nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    revoked = Column(Boolean, default=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    expires_at = Column(DateTime, nullable=False)

    user = relationship("User", back_populates="refresh_tokens")
Verify your project dependencies

One final check to confirm every package your project depends on is importable in this environment.

  • Run the full dependency verification by running this command:
python -c "import fastapi; import jwt; import pwdlib; import sqlalchemy; print('All imports OK')"

✔️ I see All imports OK

All four packages are installed and importable. Your environment is fully configured.

ⓧ I see a ModuleNotFoundError

The error message names the missing package. Install it with the correct command:

  • For fastapi: run pip install "fastapi[standard]".
  • For jwt: run pip install pyjwt. The package name on PyPI is pyjwt but you import it as jwt.
  • For pwdlib: run pip install 'pwdlib[argon2]'.
  • For sqlalchemy: run pip install sqlalchemy.

After installing, re-run the verification command above.

Your data layer is ready. Next up, you'll build the registration endpoint that hashes passwords and saves new users to this database.

User Registration with Password Hashing

Your database models are ready, but they're just empty tables. Before anyone can authenticate, you need a way for users to register with a securely stored password.

In this step, you'll define the data shapes for requests and responses, build password hashing and token utilities, and wire up the first endpoint that creates users.

In this step, get ready to:
  • Define Pydantic schemas for request and response validation.
  • Implement password hashing and token creation utilities.
  • Create the POST /register endpoint and test it via Swagger UI.
Define the request and response schemas

Every endpoint needs clearly defined input and output shapes. Pydantic models validate incoming data automatically and serialize outgoing responses.

  • Open schemas.py in VS Code and paste the following code:
from pydantic import BaseModel, EmailStr


# Schema for incoming registration requests
class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str


# Schema for outgoing user responses (no password)
class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    role: str
    is_active: bool

    model_config = {"from_attributes": True}


# Schema for token pair responses
class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str


# Schema for refresh token requests
class TokenRefreshRequest(BaseModel):
    refresh_token: str

What does this code do?

  • UserCreate defines the three fields a user submits when registering: username, email, and password.
  • UserResponse controls what the API sends back. Notice it excludes the password entirely.
  • model_config = {"from_attributes": True} lets Pydantic read data directly from SQLAlchemy model attributes.
  • Token and TokenRefreshRequest define the shapes for authentication responses and refresh requests you'll use in later steps.
  • Save schemas.py.

Why does EmailStr need a separate package?

EmailStr uses the email-validator library under the hood to validate email format. You already have it because it's bundled with fastapi[standard] which you installed in Step 1.

Build the password hashing and token utilities

Storing plain-text passwords is the most common security mistake in web applications. You need a one-way hashing function that makes it computationally expensive to reverse.

  • Open auth.py in VS Code and paste the following imports and configuration:
import hashlib
import secrets
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy.orm import Session

from database import get_db
from models import RefreshToken, User

# Replace this with the output of: openssl rand -hex 32
SECRET_KEY = "replace-with-your-openssl-rand-hex-32-output"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 7

# Argon2 password hasher (recommended by FastAPI)
password_hash = PasswordHash.recommended()
# Pre-computed hash to prevent timing attacks on user lookup
DUMMY_HASH = password_hash.hash("dummypassword")

# OAuth2 scheme tells FastAPI where to find the token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

What does this code do?

  • SECRET_KEY is used to sign JWTs. You generated this value with openssl rand -hex 32 in Step 1.
  • PasswordHash.recommended() creates an Argon2 hasher. Argon2 is more resistant to GPU cracking than Bcrypt.
  • DUMMY_HASH is a pre-computed hash used to ensure constant-time responses even when a user doesn't exist. This prevents attackers from detecting valid usernames by timing the response.
  • oauth2_scheme tells FastAPI to look for a Bearer token in the Authorization header and point the interactive docs at the /token endpoint.
  • Below the configuration you just pasted, add the utility functions:

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


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)


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


def create_refresh_token() -> str:
    return secrets.token_urlsafe(32)


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()

What do these functions do?

  • hash_password() takes a plain password and returns an Argon2 hash. This is what gets stored in the database.
  • verify_password() checks a plain password against a stored hash. The argument order is password first, hash second.
  • create_access_token() builds a JWT with a sub claim, an expiry time, and a type claim that prevents refresh tokens from being used as access tokens.
  • create_refresh_token() generates a cryptographically random string. hash_token() produces a SHA-256 hash of it for safe database storage.
  • Save auth.py.
  • Replace the placeholder value in SECRET_KEY with the actual hex string you generated in Step 1 using openssl rand -hex 32.

Forgot your secret key?

Run openssl rand -hex 32 again in your terminal to generate a new one. Paste the output between the quotes on the SECRET_KEY line.

Help me generate and set my SECRET_KEY

Create the registration endpoint

With the schemas and utilities ready, you can now build the endpoint that accepts registration data, hashes the password, and stores the new user.

  • Open main.py in VS Code and paste the following code:
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy.orm import Session

from auth import hash_password
from database import Base, engine, get_db
from models import User
from schemas import UserCreate, UserResponse

# Create all database tables on startup
Base.metadata.create_all(bind=engine)

app = FastAPI(title="JWT Auth API")


@app.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate, db: Session = Depends(get_db)):
    # Check if the username or email is already taken
    existing_user = db.query(User).filter(
        (User.username == user_data.username) | (User.email == user_data.email)
    ).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username or email already registered",
        )
    # Create the user with a hashed password
    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

What does this code do?

  • Base.metadata.create_all(bind=engine) creates the users and refresh_tokens tables in auth.db the first time the app starts.
  • The /register endpoint first checks for duplicate usernames or emails. If none exist, it hashes the password with Argon2 and saves the new user.
  • response_model=UserResponse ensures the password hash is never exposed in the API response.
  • Save main.py.
  • Start the development server by running this command in your terminal:
fastapi dev main.py
  • You should see output showing the server started at http://127.0.0.1:8000 with auto-reload enabled.
  • Open your browser and navigate to http://127.0.0.1:8000/docs to open the Swagger UI.
  • Click the POST /register endpoint to expand it.
  • Click Try it out.
  • Replace the request body with this JSON:
{
  "username": "testuser",
  "email": "test@example.com",
  "password": "securepassword123"
}
  • Click Execute.
  • You should see a 201 response with the user details. Notice the response shows role and is_active fields, but no password.

Server won't start?

  • Make sure your virtual environment is active. You should see the environment name in your terminal prompt.
  • Check for typos in your import statements. The error message will tell you which module or name failed to import.
  • Confirm you saved all three files (schemas.py, auth.py, main.py) before running the server.

Help me debug my FastAPI server startup error

✔️ Awesome, I've got everything!

Great. Double check you've saved all your files.

ⓧ I'd like to double check the full code

Here is the full code for each file at this point:

from pydantic import BaseModel, EmailStr


class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str


class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    role: str
    is_active: bool

    model_config = {"from_attributes": True}


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


class TokenRefreshRequest(BaseModel):
    refresh_token: str
import hashlib
import secrets
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy.orm import Session

from database import get_db
from models import RefreshToken, User

SECRET_KEY = "replace-with-your-openssl-rand-hex-32-output"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 7

password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


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


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)


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


def create_refresh_token() -> str:
    return secrets.token_urlsafe(32)


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy.orm import Session

from auth import hash_password
from database import Base, engine, get_db
from models import User
from schemas import UserCreate, UserResponse

Base.metadata.create_all(bind=engine)

app = FastAPI(title="JWT Auth API")


@app.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate, db: Session = Depends(get_db)):
    existing_user = db.query(User).filter(
        (User.username == user_data.username) | (User.email == user_data.email)
    ).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username or email already registered",
        )
    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

Your API now accepts registrations and safely stores hashed passwords. Next up, you'll add the login endpoint that verifies credentials and issues JWT tokens.

Login and Token Issuance

Your API can register users, but it has no way to verify who they are on subsequent requests. Without a login system, any client could claim to be any user.

In this step, you will build the login endpoint that authenticates users and issues JWT tokens. The endpoint will return a short-lived access token (15 minutes) and a long-lived refresh token (7 days), following the same pattern used by Auth0 and Okta in production.

In this step, get ready to:
  • Implement a timing-attack-safe authentication function.
  • Create a login endpoint that issues access and refresh tokens.
  • Test the full login flow and verify the token contents.
Implement the authenticate_user function

Login systems need to verify a username and password against the database. A naive implementation would return immediately if the username does not exist, which leaks information about which usernames are registered.

An attacker can measure response times to determine whether a username exists. This is called a timing attack. To prevent it, your function will always run the password hash verification, even when the user is not found.

  • In auth.py, scroll to the bottom of the file (below the hash_token function) and add the following code:
def authenticate_user(db: Session, username: str, password: str) -> User | None:
    # Look up the user by username
    user = db.query(User).filter(User.username == username).first()
    if not user:
        # Still run verification against a dummy hash to prevent timing attacks
        verify_password(password, DUMMY_HASH)
        return None
    # Verify the provided password against the stored hash
    if not verify_password(password, user.hashed_password):
        return None
    return user

What does this code do?

This function looks up a user in the database and checks their password. The key security detail is the DUMMY_HASH line.

  • If the username does not exist, the function still runs verify_password against DUMMY_HASH. This ensures the response takes the same amount of time regardless of whether the username is valid.
  • If the user exists but the password is wrong, it returns None. If both username and password are correct, it returns the user object.
  • Save auth.py.

You won't see a change yet. The next chunk uses this function in the login endpoint.

Seeing an import error in your terminal?

If the dev server shows an error about authenticate_user, that is expected. Your main.py already imports this function, but it did not exist until now. The error will resolve once you save auth.py with the new function.

Help me debug an import error in my FastAPI project.

Create the POST /token endpoint

The OAuth2 password flow defines a standard way for clients to exchange a username and password for tokens. FastAPI provides OAuth2PasswordRequestForm to handle this form data automatically.

  • In main.py, find the closing line of the register function (return new_user) and add the following endpoint below it:
@app.post("/token", response_model=Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Session = Depends(get_db),
):
    # Authenticate the user with timing-safe verification
    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"},
        )
    # Issue a short-lived access token with username and role claims
    access_token = create_access_token(
        data={"sub": user.username, "role": user.role},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    # Generate a refresh token and store its hash in the database
    refresh_token = create_refresh_token()
    refresh_token_entry = RefreshToken(
        token_hash=hash_token(refresh_token),
        user_id=user.id,
        expires_at=datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
    )
    db.add(refresh_token_entry)
    db.commit()
    # Return both tokens to the client
    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer",
    )

What does this endpoint do?

  • OAuth2PasswordRequestForm parses the username and password fields from the request body.
  • The access token embeds the sub (subject) claim with the username and a role claim. It expires in 15 minutes.
  • The refresh token is a random string. Only its SHA-256 hash gets stored in the RefreshToken table. If the database leaks, attackers cannot reconstruct the original tokens.
  • The WWW-Authenticate: Bearer header tells clients this endpoint expects Bearer token authentication.
  • Save main.py.
  • Check your terminal. The dev server should reload without errors.

Seeing a NameError or ImportError?

  • Make sure you saved both auth.py and main.py. The server imports authenticate_user from auth.py, so both files must be saved.
  • Confirm the function name is exactly authenticate_user with no typos. The import in main.py must match exactly.

Help me fix this import or name error in my FastAPI auth project.

✔️ Awesome, I've got everything!

Great. Double check you have saved both files before moving on.

ⓧ I'd like to double check the full code

Here is the complete auth.py at this point:

import hashlib
import secrets
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy.orm import Session

from database import get_db
from models import RefreshToken, User

SECRET_KEY = "replace-with-your-openssl-rand-hex-32-output"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 7

password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


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


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)


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


def create_refresh_token() -> str:
    return secrets.token_urlsafe(32)


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


def authenticate_user(db: Session, username: str, password: str) -> User | None:
    user = db.query(User).filter(User.username == username).first()
    if not user:
        verify_password(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

Here is the complete main.py at this point:

from datetime import datetime, timedelta, timezone
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from auth import (
    ACCESS_TOKEN_EXPIRE_MINUTES,
    REFRESH_TOKEN_EXPIRE_DAYS,
    authenticate_user,
    create_access_token,
    create_refresh_token,
    get_active_user,
    hash_password,
    hash_token,
)
from database import Base, engine, get_db
from models import RefreshToken, User
from schemas import Token, TokenRefreshRequest, UserCreate, UserResponse

Base.metadata.create_all(bind=engine)

app = FastAPI(title="JWT Auth API")


@app.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate, db: Session = Depends(get_db)):
    existing_user = db.query(User).filter(
        (User.username == user_data.username) | (User.email == user_data.email)
    ).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username or email already registered",
        )
    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


@app.post("/token", response_model=Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Session = Depends(get_db),
):
    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(
        data={"sub": user.username, "role": user.role},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    refresh_token = create_refresh_token()
    refresh_token_entry = RefreshToken(
        token_hash=hash_token(refresh_token),
        user_id=user.id,
        expires_at=datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
    )
    db.add(refresh_token_entry)
    db.commit()
    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer",
    )
Test the login flow via Swagger UI

FastAPI automatically generates interactive API docs. You can use the Swagger UI to register a user and then log in to receive your tokens.

  • Open your browser and navigate to http://127.0.0.1:8000/docs.
  • If you have not already registered a user, expand the POST /register section, click Try it out, and submit a request with a username, email, and password.
  • Expand the POST /token section and click Try it out.
  • Enter the same username and password you registered with in the username and password fields.
  • Click Execute.

You should see a 200 response with an access_token, refresh_token, and token_type in the response body.

  • Copy the access_token value from the response.
  • Open jwt.io in a new browser tab.
  • Paste the access token into the Encoded field on the left side.
  • On the right side under Payload, confirm you see a sub claim matching your username, a role claim set to user, and an exp claim set to 15 minutes from now.

Why verify at jwt.io?

jwt.io decodes the token without needing your secret key (the payload is base64-encoded, not encrypted). This lets you visually confirm the claims are correct before you build the verification logic in the next step.

Getting a 401 Unauthorized response?

  • Make sure you are using the exact username (not email) you registered with. The login form requires the username field, not the email.
  • Double check you have not restarted the server with a fresh database since registering. Restarting may recreate auth.db if the file was deleted.
  • Check that the password is entered exactly as you registered it. Passwords are case-sensitive.

Help me debug why my /token endpoint returns 401.

Your API now authenticates users and issues tokens. Next up, you will build the dependency chain that validates these tokens on every protected request.

Protected Routes with Layered Dependencies

Your API can now register users and issue tokens. But tokens are only useful if endpoints actually check them.

In this step, you will build a 3-layer dependency chain that extracts and validates tokens, checks user status, and enforces roles. Then you will wire two endpoints through that chain to see it in action.

In this step, get ready to:
  • Build a layered dependency chain in auth.py that validates JWTs, checks user status, and checks roles.
  • Create protected /users/me and /inference endpoints.
  • Test that unauthenticated requests are rejected and authenticated requests succeed.
Build the token validation layer

The first layer of the dependency chain is the most critical. It intercepts every protected request, extracts the Bearer token from the header, and decodes it with PyJWT. If anything is wrong, the request stops here.

  • In auth.py, scroll to the bottom of the file (after the authenticate_user function).
  • Add the get_current_user dependency by pasting this code at the end of the file:
async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Session = Depends(get_db),
) -> User:
    # Define the exception once so every failure path returns the same response
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # Decode the JWT and verify its signature and expiry
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # Reject refresh tokens used as access tokens
        if payload.get("type") != "access":
            raise credentials_exception
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except InvalidTokenError:
        raise credentials_exception
    # Look up the user in the database
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception
    return user

What does this code do?

This function is Layer 1 of your dependency chain. FastAPI calls it automatically whenever a route declares it as a dependency.

  • oauth2_scheme extracts the Bearer token from the Authorization header.
  • jwt.decode verifies the signature, checks expiry, and returns the payload.
  • The type claim check prevents someone from using a refresh token as an access token.
  • The WWW-Authenticate: Bearer header in the 401 response tells clients this endpoint requires a Bearer token.

Why one exception for every failure?

Every failure case returns the same generic error message. If you returned specific messages like "user not found" or "token expired", an attacker could probe your system to learn which usernames are valid.

Add the active-user check and role factory

Layer 2 adds a simple guard for disabled accounts. Layer 3 is a factory function that produces role-specific dependencies you can reuse on any endpoint.

  • Still in auth.py, paste the following two functions below get_current_user:
async def get_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    # Block disabled accounts from accessing any protected route
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


def require_role(role: str):
    # Factory that returns a dependency tailored to the specified role
    async def role_checker(
        current_user: Annotated[User, Depends(get_active_user)],
    ) -> User:
        if current_user.role != role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions",
            )
        return current_user
    return role_checker

What does this code do?

  • get_active_user depends on get_current_user. FastAPI resolves the chain automatically: extract token, then check active status.
  • require_role is a factory. Calling require_role("admin") returns a new dependency that first runs the active-user check, then verifies the role claim.
  • This pattern means you can protect any endpoint by simply adding Depends(get_active_user) or Depends(require_role("admin")) to its signature.
  • Save auth.py.
  • Check your terminal. The FastAPI dev server should reload without any errors.

Seeing an import error in the terminal?

  • Make sure get_current_user is defined before get_active_user. The Depends(get_current_user) reference must point to an existing function.
  • Confirm your function signatures match exactly. A missing Annotated wrapper or a typo in Depends will cause a runtime error.

Ask for help debugging your dependency chain.

✔️ Awesome, I've got everything!

Great. Make sure you have saved auth.py before moving on.

ⓧ I'd like to double check the full code

import hashlib
import secrets
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy.orm import Session

from database import get_db
from models import RefreshToken, User

SECRET_KEY = "replace-with-your-openssl-rand-hex-32-output"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 7

password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


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


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)


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


def create_refresh_token() -> str:
    return secrets.token_urlsafe(32)


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


def authenticate_user(db: Session, username: str, password: str) -> User | None:
    user = db.query(User).filter(User.username == username).first()
    if not user:
        verify_password(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Session = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "access":
            raise credentials_exception
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except InvalidTokenError:
        raise credentials_exception
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception
    return user


async def get_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


def require_role(role: str):
    async def role_checker(
        current_user: Annotated[User, Depends(get_active_user)],
    ) -> User:
        if current_user.role != role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions",
            )
        return current_user
    return role_checker
Create the protected endpoints

Now that the dependency chain exists, wiring a protected endpoint takes a single parameter. Any route that declares Depends(get_active_user) in its signature automatically runs all three layers before executing.

  • Switch to main.py.
  • Scroll to the bottom of the file (after the /token endpoint).
  • Add the two protected endpoints by pasting this code at the end:
@app.get("/users/me", response_model=UserResponse)
def read_current_user(current_user: Annotated[User, Depends(get_active_user)]):
    # Return the authenticated user's profile
    return current_user


@app.post("/inference")
def run_inference(current_user: Annotated[User, Depends(get_active_user)]):
    # Simulated AI model endpoint - only accessible with a valid token
    return {
        "user": current_user.username,
        "result": "Model prediction: positive sentiment (confidence: 0.94)",
        "model": "sentiment-v2",
    }

What does this code do?

  • /users/me returns the profile of whoever is making the request. The user object is resolved by the dependency chain before the function body runs.
  • /inference simulates a protected AI model endpoint. In a real application, this is where you would call your ML model. The response proves the caller's identity by including their username.
  • Save main.py.
  • Check your terminal. The server should reload and show no errors.

Server showing an error after saving?

  • Make sure get_active_user is imported from auth at the top of main.py. Your import block should include it.
  • Check that auth.py is saved with all three new functions. If get_active_user is missing from that file, the import will fail.

Help me fix my import error between main.py and auth.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved main.py before moving on to testing.

ⓧ I'd like to double check the full code

from datetime import datetime, timedelta, timezone
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from auth import (
    ACCESS_TOKEN_EXPIRE_MINUTES,
    REFRESH_TOKEN_EXPIRE_DAYS,
    authenticate_user,
    create_access_token,
    create_refresh_token,
    get_active_user,
    hash_password,
    hash_token,
)
from database import Base, engine, get_db
from models import RefreshToken, User
from schemas import Token, TokenRefreshRequest, UserCreate, UserResponse

Base.metadata.create_all(bind=engine)

app = FastAPI(title="JWT Auth API")


@app.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate, db: Session = Depends(get_db)):
    existing_user = db.query(User).filter(
        (User.username == user_data.username) | (User.email == user_data.email)
    ).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username or email already registered",
        )
    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


@app.post("/token", response_model=Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Session = Depends(get_db),
):
    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(
        data={"sub": user.username, "role": user.role},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    refresh_token = create_refresh_token()
    refresh_token_entry = RefreshToken(
        token_hash=hash_token(refresh_token),
        user_id=user.id,
        expires_at=datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
    )
    db.add(refresh_token_entry)
    db.commit()
    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer",
    )


@app.get("/users/me", response_model=UserResponse)
def read_current_user(current_user: Annotated[User, Depends(get_active_user)]):
    return current_user


@app.post("/inference")
def run_inference(current_user: Annotated[User, Depends(get_active_user)]):
    return {
        "user": current_user.username,
        "result": "Model prediction: positive sentiment (confidence: 0.94)",
        "model": "sentiment-v2",
    }
Test the protected routes

The real proof is in the request. You will now confirm that your dependency chain rejects unauthenticated callers and allows authenticated ones through.

  • Open the Swagger UI in your browser at http://127.0.0.1:8000/docs.
  • Expand the POST /inference endpoint.
  • Click Try it out, then Execute without providing any token.

You should see a 401 response with the message Not authenticated. The dependency chain rejected the request because no Bearer token was provided.

  • Now click the Authorize button at the top right of the Swagger page.
  • Enter the username and password you registered in the previous step.
  • Click Authorize in the dialog, then Close.
  • Go back to POST /inference and click Execute again.

This time you should see a 200 response with your username and the simulated model prediction.

How does the Authorize button work?

The Authorize button calls your /token endpoint behind the scenes with the credentials you entered. It stores the returned access token in the browser session and attaches it as a Bearer header on all subsequent requests.

This is why you set tokenUrl="token" when creating OAuth2PasswordBearer. Swagger uses that URL to know where to send login credentials.

Still getting 401 after authorizing?

  • Make sure you registered a user first via POST /register. The Authorize dialog needs valid credentials.
  • Check that your username and password match exactly what you registered. Passwords are case-sensitive.
  • If you restarted the server and lost your database, register a new user and try again.

Help me figure out why my /inference request still returns 401 after I authorize.

Your API now enforces authentication on every protected route through a clean dependency chain. Next up, you will build refresh token rotation so users can get new access tokens without logging in again.

Refresh Token Rotation

Your API now protects routes with a layered dependency chain. An authenticated user gets 15 minutes of access before their token expires.

Forcing users to log in again every 15 minutes would be a terrible experience. Refresh tokens solve this by letting the client silently request a new access token. In this step, you will build a /refresh endpoint that implements token rotation with reuse detection.

In this step, get ready to:
  • Create a /refresh endpoint that issues new token pairs.
  • Implement reuse detection that revokes all tokens when a stolen token is replayed.
  • Test the full rotation lifecycle in Swagger UI.
Create the POST /refresh endpoint

The refresh endpoint accepts a raw refresh token, hashes it to find the matching database record, performs security checks, and then issues a completely new token pair.

  • In main.py, find the login function's closing return statement.
  • Add the following /refresh endpoint below the login function and above the read_current_user function:
@app.post("/refresh", response_model=Token)
def refresh(body: TokenRefreshRequest, db: Session = Depends(get_db)):
    # Hash the incoming token to compare against stored hashes
    token_hash = hash_token(body.refresh_token)
    stored_token = db.query(RefreshToken).filter(
        RefreshToken.token_hash == token_hash
    ).first()
    if not stored_token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid refresh token",
        )
    # REUSE DETECTION: if this token was already revoked, someone stole it
    if stored_token.revoked:
        db.query(RefreshToken).filter(
            RefreshToken.user_id == stored_token.user_id
        ).update({"revoked": True})
        db.commit()
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token reuse detected. All sessions revoked.",
        )
    # Check if the token has expired
    if stored_token.expires_at < datetime.now(timezone.utc):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Refresh token expired",
        )
    # Revoke the old token (rotation)
    stored_token.revoked = True
    user = db.query(User).filter(User.id == stored_token.user_id).first()
    # Issue a fresh access token
    access_token = create_access_token(
        data={"sub": user.username, "role": user.role},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    # Issue a fresh refresh token and store its hash
    new_refresh_token = create_refresh_token()
    new_refresh_entry = RefreshToken(
        token_hash=hash_token(new_refresh_token),
        user_id=user.id,
        expires_at=datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
    )
    db.add(new_refresh_entry)
    db.commit()
    return Token(
        access_token=access_token,
        refresh_token=new_refresh_token,
        token_type="bearer",
    )

What does this code do?

The endpoint performs four security checks in sequence before issuing new tokens:

  • Token lookup: hashes the incoming token and searches the database. If no match exists, the token was never issued by your server.
  • Reuse detection: if the token was already revoked (used once before), an attacker is replaying it. Revoke ALL of that user's tokens as a safety measure.
  • Expiry check: even valid tokens must not be past their 7-day lifetime.
  • Rotation: mark the old token as revoked, generate a brand-new access/refresh pair, and store the new refresh token's hash.
  • Save main.py.
  • Your dev server auto-reloads. Open http://127.0.0.1:8000/docs in your browser. You should see a new POST /refresh endpoint listed in the Swagger UI.

Don't see the /refresh endpoint?

Make sure the function is placed between the login and read_current_user functions. Check that the decorator reads @app.post("/refresh", response_model=Token) exactly.

If the terminal shows an import error, confirm that TokenRefreshRequest is imported at the top of main.py (it was added in a previous step).

Help me debug my /refresh endpoint not appearing in Swagger UI.

Understand how reuse detection protects against stolen tokens

The reuse detection block is the most important security feature in this endpoint. It catches a scenario where an attacker steals a refresh token after the legitimate user has already used it.

Why revoke the entire token family?

Imagine this sequence: a user refreshes their token (the old one gets revoked), but an attacker previously copied that old token. When the attacker tries to use it, the server sees it was already revoked. That means someone else used it first.

The server cannot tell which party (user or attacker) holds the new token. The safest response is to revoke ALL of that user's refresh tokens, forcing everyone to log in again. This is the same pattern Auth0 and Okta use in production.

The key line is the query that filters by user_id and sets revoked to True on every matching row. This wipes out the entire token family in one database call.

Test the full rotation flow

You will test three scenarios: a successful refresh, rejection of the old token, and family revocation on replay.

  • In the Swagger UI at http://127.0.0.1:8000/docs, expand the POST /token endpoint.
  • Click Try it out.
  • Enter the username and password you registered earlier, then click Execute.
  • Copy the refresh_token value from the response body. Keep this somewhere handy.

Now use that refresh token to get a new pair.

  • Expand the POST /refresh endpoint and click Try it out.
  • Paste the refresh token you copied into the refresh_token field of the request body.
  • Click Execute. You should see a 200 response with a brand-new access_token and refresh_token.

Now test that the old token is rejected.

  • Paste the same old refresh token into the POST /refresh endpoint again.
  • Click Execute. You should see a 401 response with the detail Token reuse detected. All sessions revoked..

What just happened?

The first refresh succeeded and revoked the old token. When you replayed the revoked token, the server detected reuse and revoked every refresh token for that user. Even the new token from the first refresh is now invalid.

To continue testing other endpoints, log in again via POST /token to get a fresh set of tokens.

  • Verify the family revocation by trying the new refresh token from your first successful refresh. Paste it into POST /refresh and click Execute. It should also return 401, confirming that the entire token family was wiped.

Getting a 422 instead of 401?

A 422 means the request body format is wrong. Make sure the JSON body has a refresh_token key (not token or refreshToken). The request body should look like: {"refresh_token": "your-token-here"}.

Help me debug a 422 error on my /refresh endpoint.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and confirmed the rotation flow works.

ⓧ I'd like to double check the full code

from datetime import datetime, timedelta, timezone
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from auth import (
    ACCESS_TOKEN_EXPIRE_MINUTES,
    REFRESH_TOKEN_EXPIRE_DAYS,
    authenticate_user,
    create_access_token,
    create_refresh_token,
    get_active_user,
    hash_password,
    hash_token,
)
from database import Base, engine, get_db
from models import RefreshToken, User
from schemas import Token, TokenRefreshRequest, UserCreate, UserResponse

Base.metadata.create_all(bind=engine)

app = FastAPI(title="JWT Auth API")


@app.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate, db: Session = Depends(get_db)):
    existing_user = db.query(User).filter(
        (User.username == user_data.username) | (User.email == user_data.email)
    ).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username or email already registered",
        )
    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


@app.post("/token", response_model=Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Session = Depends(get_db),
):
    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(
        data={"sub": user.username, "role": user.role},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    refresh_token = create_refresh_token()
    refresh_token_entry = RefreshToken(
        token_hash=hash_token(refresh_token),
        user_id=user.id,
        expires_at=datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
    )
    db.add(refresh_token_entry)
    db.commit()
    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer",
    )


@app.post("/refresh", response_model=Token)
def refresh(body: TokenRefreshRequest, db: Session = Depends(get_db)):
    token_hash = hash_token(body.refresh_token)
    stored_token = db.query(RefreshToken).filter(
        RefreshToken.token_hash == token_hash
    ).first()
    if not stored_token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid refresh token",
        )
    if stored_token.revoked:
        db.query(RefreshToken).filter(
            RefreshToken.user_id == stored_token.user_id
        ).update({"revoked": True})
        db.commit()
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token reuse detected. All sessions revoked.",
        )
    if stored_token.expires_at < datetime.now(timezone.utc):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Refresh token expired",
        )
    stored_token.revoked = True
    user = db.query(User).filter(User.id == stored_token.user_id).first()
    access_token = create_access_token(
        data={"sub": user.username, "role": user.role},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    new_refresh_token = create_refresh_token()
    new_refresh_entry = RefreshToken(
        token_hash=hash_token(new_refresh_token),
        user_id=user.id,
        expires_at=datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
    )
    db.add(new_refresh_entry)
    db.commit()
    return Token(
        access_token=access_token,
        refresh_token=new_refresh_token,
        token_type="bearer",
    )


@app.get("/users/me", response_model=UserResponse)
def read_current_user(current_user: Annotated[User, Depends(get_active_user)]):
    return current_user


@app.post("/inference")
def run_inference(current_user: Annotated[User, Depends(get_active_user)]):
    return {
        "user": current_user.username,
        "result": "Model prediction: positive sentiment (confidence: 0.94)",
        "model": "sentiment-v2",
    }

Your authentication system now handles the full token lifecycle. Next up, you will tackle the Secret Mission and add role-based access control with admin-only endpoints.

Secret mission

Role-Based Access Control

Your API authenticates users, but every authenticated user has the same power. Can you add admin-only endpoints that let privileged users manage sessions and view all registered accounts?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Project files (main.py, auth.py, database.py, models.py, schemas.py).
  • SQLite database file (auth.db).
  • Python virtual environment.
  • Installed pip packages (fastapi, pyjwt, pwdlib, sqlalchemy).

Keep everything running

No action needed. Choose this if you are still actively building or want to keep experimenting with your auth API.

  • Your project folder stays on your Desktop.
  • The auth.db file retains all registered users and refresh tokens.
  • You can restart the server anytime by activating your virtual environment and running fastapi dev main.py.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files stay on disk so you can pick up where you left off.

  • Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
  • Deactivate your virtual environment by running:
deactivate

When you come back, reactivate the virtual environment and start the server again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your code, database, and virtual environment permanently.

  • Stop the FastAPI dev server by pressing Ctrl+C in the terminal.
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the project folder by running:

macOS

cd ~/Desktop
rm -rf jwt-auth-api

Windows

cd ~\Desktop
rmdir /s /q jwt-auth-api

Replace jwt-auth-api with whatever you named your project folder in Step 1.

  • Confirm deletion by running ls (macOS) or dir (Windows). Your project folder should no longer appear in the listing.

Nice Work!

Nice Work!

Nice work! You just built a production-grade JWT authentication system with FastAPI from scratch.

Your API now handles user registration, login, token issuance, refresh token rotation, and protected endpoints. These are the same security patterns used by Auth0 and Okta in production.

You've learned how to:

  • Build a complete JWT authentication system with FastAPI, including secure password hashing with Argon2 and timing attack prevention.
  • Implement refresh token rotation with family-based reuse detection that revokes all sessions when a compromised token is replayed.
  • Create a layered dependency injection chain that cleanly separates token validation, user loading, and permission enforcement.
  • Secret Mission: Added admin-only endpoints with a reusable require_role dependency factory and session revocation.

Ready to quiz yourself?