Secure a Bookstore REST API with FastAPI

Build a secured bookstore API with JWT auth, RBAC, and input validation.

Introduction

30 Second Summary

Every time you log into an app or website, a security system works behind the scenes to verify your identity and control what you can do. Without these protections, anyone could access, modify, or delete data they should never touch.

In this project, you will build a bookstore API with FastAPI and progressively lock it down with production-grade security. Starting with an open API that anyone can access, you will add authentication, authorization, and input validation layer by layer.

What You'll Build

A secured bookstore REST API with JWT token authentication, admin-only write permissions, and Pydantic-enforced input validation.

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

  • A working login endpoint that issues JWT tokens after verifying hashed passwords.
  • Role-based access control that restricts book creation and deletion to admin users only.
  • Strict input validation with Pydantic Field constraints that automatically rejects malformed requests.
  • Secret Mission: Add rate limiting and CORS configuration to defend against brute-force attacks and cross-origin threats.

Are there any prerequisites?

This project assumes basic Python knowledge and familiarity with REST APIs and HTTP methods. You should already have Python and VS Code installed on your computer.

Set Up the Project and Build the Unsecured API

Before you can lock anything down, you need something to lock down. In this step, you will build a simple bookstore API that lets anyone list, add, and delete books with zero restrictions.

By the end, you will see exactly why this is a problem. Any anonymous visitor can wipe your entire catalog with a single request.

In this step, get ready to:
  • Create a project folder with a Python virtual environment and install dependencies.
  • Build an unsecured bookstore API with three endpoints in main.py.
  • Test the API and spot the security gap.
Create a project folder and install dependencies

You need a dedicated folder for this project and a Python virtual environment to keep your packages isolated from the rest of your system.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir secure-bookstore-api
cd secure-bookstore-api
  • Create a Python virtual environment by running:
python -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay local to this project and won't interfere with other Python projects on your machine.

  • Activate the virtual environment by running the command for your operating system:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

On Windows, if you get a "running scripts is disabled" error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell, then try the activate command again.

On macOS, make sure you used source before the path. Running the script directly without source won't activate the environment.

Why can't I activate my virtual environment?

  • Install the three packages your project needs by running:
pip install "fastapi[standard]" pyjwt "pwdlib[argon2]"

What are these packages?

  • fastapi[standard] installs FastAPI along with a development server and form-data support.
  • pyjwt handles creating and verifying JWT tokens for authentication (you will use this in a later step).
  • pwdlib[argon2] provides secure password hashing with the Argon2 algorithm.

You should see a list of successfully installed packages in the output, including fastapi, pyjwt, and pwdlib.

Seeing an install error?

Make sure (venv) appears in your terminal prompt before running pip install. Installing without an active virtual environment can cause permission errors.

If you see a "No matching distribution" error, confirm you are running Python 3.10 or higher by running python --version.

Help me fix my pip install error.

Build the unsecured bookstore API

Now you will create the API itself. It stores books in a Python list (no database needed) and exposes endpoints for listing, adding, and removing books.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the secure-bookstore-api folder, and click Open.
  • In the Explorer sidebar on the left, click the New File icon (a page with a plus sign).
  • Name the file main.py and press Enter.
  • Set up the foundation of your API by copying this code into main.py:
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

# Create the FastAPI app
app = FastAPI(title="Secure Bookstore API")

# Define the shape of a book using Pydantic
class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int

# In-memory list of books (acts as our database)
books_db: list[dict] = [
    {"id": 1, "title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "year": 1925, "isbn": None},
    {"id": 2, "title": "To Kill a Mockingbird", "author": "Harper Lee", "year": 1960, "isbn": None},
]
# Track the next available book ID
next_book_id = 3

What does this code do?

  • Book is a Pydantic model that defines the shape of a book. FastAPI uses it to automatically validate incoming data and generate documentation.
  • books_db is a plain Python list that stores book records in memory. It starts with two sample books so you have data to test with.
  • next_book_id keeps track of the next ID to assign when a new book is added.
  • Save main.py.

You won't see a visible change yet. The next chunk adds the first endpoint that makes this data accessible.

  • Add the first endpoint below your existing code in main.py:
# Return all books in the database
@app.get("/books", response_model=list[Book])
async def list_books():
    return books_db

What does this endpoint do?

The @app.get("/books") decorator tells FastAPI to respond to GET requests at the /books path. The response_model parameter tells FastAPI to validate and format the response as a list of Book objects.

  • Save main.py.
  • Open a terminal inside VS Code by pressing Ctrl+` (backtick key, usually above Tab).

Does the virtual environment carry over?

VS Code's integrated terminal usually auto-activates your virtual environment when it detects a venv folder in the project. Look for (venv) in the prompt. If you don't see it, re-run the activation command from the previous section.

  • Start the FastAPI development server by running:
fastapi dev main.py

You should see output that includes Uvicorn running on http://127.0.0.1:8000. The server is now running and watching for file changes.

  • Open your browser and navigate to http://127.0.0.1:8000/docs.

You should see the Swagger UI with a GET /books endpoint listed. FastAPI generates this interactive documentation automatically from your code.

  • Click the GET /books row to expand it.
  • Click Try it out.
  • Click Execute.

You should see a JSON response containing the two sample books: "The Great Gatsby" and "To Kill a Mockingbird".

Don't see the Swagger docs page?

Make sure the terminal still shows the server running. If you see an error, check that (venv) is in your prompt. The fastapi command is only available inside the virtual environment.

Double-check that the URL is exactly http://127.0.0.1:8000/docs with no trailing spaces or extra characters.

Help me get the FastAPI dev server running.

  • Add the second endpoint below the list_books function in main.py:
# Add a new book to the database
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(book: Book):
    global next_book_id
    new_book = {"id": next_book_id, **book.model_dump(exclude={"id"})}
    books_db.append(new_book)
    next_book_id += 1
    return new_book

What does this endpoint do?

  • The @app.post decorator handles POST requests, which are used to create new resources.
  • FastAPI automatically parses the JSON request body into a Book object. If the data doesn't match the model, it returns a validation error.
  • model_dump(exclude={"id"}) converts the book to a dictionary but leaves out the client-provided ID. The server assigns its own ID using next_book_id.
  • Save main.py. The dev server automatically reloads when you save.
  • Refresh your browser at http://127.0.0.1:8000/docs. You should see a new POST /books endpoint listed.
  • Expand POST /books, click Try it out.
  • Replace the example JSON in the request body with this:
{
  "id": 0,
  "title": "1984",
  "author": "George Orwell",
  "year": 1949
}
  • Click Execute.

You should see a 201 Created response with the book assigned "id": 3.

Seeing a 422 Validation Error?

Make sure the JSON you pasted is valid. Every field name must be in double quotes, and there should be no trailing commas after the last field.

Help me fix my POST /books request.

  • Add the final endpoint below the create_book function in main.py:
# Remove a book by its ID
@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(book_id: int):
    for i, book in enumerate(books_db):
        if book["id"] == book_id:
            books_db.pop(i)
            return
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Book not found",
    )

What does this endpoint do?

  • {book_id} in the path is a path parameter. FastAPI extracts the value from the URL and passes it as the book_id argument.
  • The function loops through books_db to find a matching ID. If found, it removes the book and returns a 204 No Content response.
  • If the book doesn't exist, it raises an HTTPException with a 404 Not Found status.
  • Save main.py and refresh the Swagger docs in your browser.
  • Expand DELETE /books/{book_id}, click Try it out.
  • Enter 1 in the book_id field.
  • Click Execute.

You should see a 204 No Content response. "The Great Gatsby" has been removed from the list.

Getting a 404 Not Found instead?

The in-memory database resets every time the server reloads. If you saved the file between adding the book and deleting it, the server restarted and the list went back to the original two books. Try entering 1 or 2 (the IDs of the sample books) instead.

Help me fix my DELETE endpoint.

✔️ Awesome, I've got everything!

Make sure you have saved main.py and all three endpoints are visible in the Swagger docs.

ⓧ I'd like to double check the full code

Here is the complete main.py file. Make sure your code matches this exactly.

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI(title="Secure Bookstore API")


class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int


books_db: list[dict] = [
    {"id": 1, "title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "year": 1925, "isbn": None},
    {"id": 2, "title": "To Kill a Mockingbird", "author": "Harper Lee", "year": 1960, "isbn": None},
]
next_book_id = 3


@app.get("/books", response_model=list[Book])
async def list_books():
    return books_db


@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(book: Book):
    global next_book_id
    new_book = {"id": next_book_id, **book.model_dump(exclude={"id"})}
    books_db.append(new_book)
    next_book_id += 1
    return new_book


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(book_id: int):
    for i, book in enumerate(books_db):
        if book["id"] == book_id:
            books_db.pop(i)
            return
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Book not found",
    )
Spot the security gap

Take a step back and look at what you just did. You listed books, created a new one, and deleted another. At no point did the API ask who you are or whether you are allowed to do those things.

In a production environment, this means any person on the internet could wipe your entire book catalog or flood it with junk data. This is the exact problem you will fix over the next steps.

Why build the insecure version first?

Starting with an unsecured API lets you see the "before" state clearly. Each security layer you add in the following steps directly addresses a vulnerability you can already observe.

You will add password hashing, token authentication, role-based access control, and input validation. Each one closes a specific gap.

Your unsecured bookstore API is up and running. Next up, you will add password hashing and user registration so the API knows who is making requests.

Add Password Hashing and User Registration

Your bookstore API is up and running. In the last step, you proved that anyone can create or delete books without any credentials at all.

The first layer of security to add is password hashing. Storing passwords as plain text is one of the most dangerous mistakes an API can make. In this step, you will set up hashed passwords, build a user database, and create a registration endpoint so new users can sign up securely.

In this step, get ready to:
  • Add user models and a fake user database with hashed passwords.
  • Create helper functions for password verification and user authentication.
  • Build a registration endpoint that hashes passwords before storing them.
Add user models and a fake user database

Before you can authenticate users, you need data models that describe what a user looks like and a database to store them in. You also need a way to hash passwords so they are never stored in plain text.

You will use pwdlib with Argon2 to hash passwords. Argon2 won the Password Hashing Competition and is recommended by OWASP for its resistance to GPU-based attacks.

First, your file needs some restructuring. The app = FastAPI(...) line currently sits near the top of the file. You need to move it lower so that helper functions and models can be defined above the endpoints.

  • In main.py, find this line near the top of the file:
app = FastAPI(title="Secure Bookstore API")
  • Cut this line using Cmd+X (macOS) or Ctrl+X (Windows).
  • Paste it on a new blank line directly above @app.get("/books", response_model=list[Book]).
  • Now find the line from pydantic import BaseModel. Add the following code below it, above the existing class Book(BaseModel)::
from pwdlib import PasswordHash

# --- Security Configuration ---

# Create a password hasher using Argon2 (the recommended algorithm)
password_hash = PasswordHash.recommended()
# A dummy hash used to prevent timing attacks during authentication
DUMMY_HASH = password_hash.hash("dummypassword")

# --- Models ---


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


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


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None
    role: str = "user"


class UserInDB(User):
    hashed_password: str

What does this code do?

PasswordHash.recommended() creates a hasher pre-configured with Argon2. Every password you hash will be turned into a long, random-looking string that cannot be reversed back to the original.

DUMMY_HASH is a pre-computed hash of a throwaway password. You will use it later in the authentication function to prevent timing attacks. You will learn more about this in the next substep.

The four new models define the shapes of your security data. Token and TokenData will be used in the next step for JWT tokens. User holds public user info, while UserInDB extends it with a hashed_password field for storage.

  • Save main.py.
  • Check your terminal where the dev server is running. It should restart with no errors.

Seeing an ImportError for pwdlib?

Make sure your virtual environment is activated. The terminal prompt should show (venv) at the beginning. If it does not, activate it and restart the dev server.

Also confirm that you installed pwdlib with the Argon2 extra. The install command is pip install "pwdlib[argon2]", not just pip install pwdlib.

Help me fix an ImportError for pwdlib in my FastAPI project

Now you need a user database. In a production app, this would be a real database like PostgreSQL. For this project, you will use an in-memory dictionary that stores two pre-built users with different roles.

  • In main.py, find the books_db list. Add the following code above it:
# --- Fake Databases ---

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        # password_hash.hash() converts "secret" into an Argon2 hash
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "admin",
    },
    "janedoe": {
        "username": "janedoe",
        "full_name": "Jane Doe",
        "email": "janedoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "user",
    },
}

What does this code do?

This dictionary acts as your user database. Each user has a role field. johndoe is an "admin" and janedoe is a "user". You will use these roles later to control who can create and delete books.

Notice that password_hash.hash("secret") runs at startup. Both users share the plaintext password "secret", but the hashes stored in the dictionary are completely different random-looking strings. Even identical passwords produce unique hashes.

  • Save main.py.
  • Check your terminal. The dev server should restart with no errors.

Seeing a NameError for password_hash?

Make sure the password_hash = PasswordHash.recommended() line appears above fake_users_db in your file. Python reads top to bottom, so the hasher must be defined before you call .hash() on it.

Help me fix a NameError for password_hash in my FastAPI project

Create helper functions

With models and a user database in place, you need functions that use them. These helpers will verify passwords, look up users, and authenticate login attempts.

  • In main.py, find the line next_book_id = 3. Add the following code below it:
# --- Helper Functions ---


def verify_password(plain_password: str, hashed_password: str) -> bool:
    # Compare a plaintext password against its Argon2 hash
    return password_hash.verify(plain_password, hashed_password)


def get_password_hash(password: str) -> str:
    # Turn a plaintext password into an Argon2 hash
    return password_hash.hash(password)


def get_user(db: dict, username: str) -> UserInDB | None:
    # Look up a user by username and return a UserInDB object
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
    return None

What does this code do?

verify_password() takes a plaintext password and a hash, then returns True or False. It wraps pwdlib's .verify() method, which handles all the Argon2 comparison logic internally.

get_user() looks up a username in the database dictionary and returns a UserInDB object. If the username does not exist, it returns None.

  • Save main.py. You will not see a visible change yet. The next function uses these helpers.

Now you need the main authentication function. This is where the DUMMY_HASH you created earlier comes into play.

  • In main.py, add the following function directly below get_user():
def authenticate_user(db: dict, username: str, password: str) -> UserInDB | bool:
    user = get_user(db, username)
    if not user:
        # Verify against dummy hash to prevent timing attacks
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

Why verify against a dummy hash when the user does not exist?

Password verification takes a measurable amount of time (Argon2 is deliberately slow). If you skip verification when a user is not found, an attacker can measure response times to figure out which usernames exist in your database. This is called a timing attack.

By always running verify_password() against DUMMY_HASH, the response takes the same amount of time whether the username exists or not. The attacker learns nothing.

  • Finally, add a # --- App --- comment on the blank line directly above app = FastAPI(title="Secure Bookstore API").
  • Save main.py.
  • Check your terminal. The dev server should restart with no errors.

Seeing a SyntaxError or IndentationError?

Check that each function definition starts with def at column 0 (no leading spaces). Also make sure there is a blank line between each function.

If the error points to the @app.get line, confirm that app = FastAPI(title="Secure Bookstore API") appears above the book endpoints but below your helper functions.

Help me fix a SyntaxError or IndentationError in my FastAPI main.py

Add a registration endpoint and test it

Your helper functions are ready but nothing in the API uses them yet. Time to build the first endpoint that puts password hashing to work: a user registration route.

  • In main.py, find @app.get("/books", response_model=list[Book]). Add the following code above it:
# --- Auth Endpoints ---


@app.post("/register")
async def register_user(
    username: str,
    password: str,
    role: str = "user",
):
    # Prevent duplicate usernames
    if username in fake_users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    # Hash the password before storing
    hashed = get_password_hash(password)
    fake_users_db[username] = {
        "username": username,
        "full_name": None,
        "email": None,
        "hashed_password": hashed,
        "disabled": False,
        "role": role,
    }
    return {"username": username, "hashed_password": hashed, "role": role}


# --- Book Endpoints ---

What does this endpoint do?

POST /register accepts a username, password, and optional role as query parameters. It first checks if the username is already taken.

If the username is available, it hashes the password using get_password_hash() and stores the new user in fake_users_db. The response includes the hashed password so you can see that the original plaintext is never stored.

  • Save main.py.
  • Open your browser and go to http://127.0.0.1:8000/docs.
  • Find the POST /register endpoint and click it to expand.
  • Click Try it out.
  • Enter testuser for the username.
  • Enter mypassword123 for the password.
  • Leave the role as user.
  • Click Execute.

In the response body, you should see a JSON object with the username testuser and a hashed_password that starts with $argon2id$. That long string is the Argon2 hash. The plaintext password mypassword123 is nowhere in the response.

Not seeing the /register endpoint in Swagger?

Make sure you saved main.py and that the dev server restarted in your terminal. Try a hard refresh in the browser with Cmd+Shift+R (macOS) or Ctrl+Shift+R (Windows).

If the endpoint appears but returns a 500 error, check that get_password_hash is defined above the register endpoint in your file. Python needs to see the function definition before the endpoint calls it.

Help me fix my /register endpoint in FastAPI Swagger docs

✔️ Awesome, I've got everything!

Double check you have saved main.py and your dev server is running without errors.

ⓧ I'd like to double check the full code

Here is the complete main.py file as it should look at the end of this step.

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from pwdlib import PasswordHash

# --- Security Configuration ---

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

# --- Models ---


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


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


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None
    role: str = "user"


class UserInDB(User):
    hashed_password: str


class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int


# --- Fake Databases ---

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "admin",
    },
    "janedoe": {
        "username": "janedoe",
        "full_name": "Jane Doe",
        "email": "janedoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "user",
    },
}

books_db: list[dict] = [
    {"id": 1, "title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "year": 1925, "isbn": None},
    {"id": 2, "title": "To Kill a Mockingbird", "author": "Harper Lee", "year": 1960, "isbn": None},
]
next_book_id = 3

# --- Helper Functions ---


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


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


def get_user(db: dict, username: str) -> UserInDB | None:
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
    return None


def authenticate_user(db: dict, username: str, password: str) -> UserInDB | bool:
    user = get_user(db, username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


# --- App ---

app = FastAPI(title="Secure Bookstore API")

# --- Auth Endpoints ---


@app.post("/register")
async def register_user(
    username: str,
    password: str,
    role: str = "user",
):
    if username in fake_users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    hashed = get_password_hash(password)
    fake_users_db[username] = {
        "username": username,
        "full_name": None,
        "email": None,
        "hashed_password": hashed,
        "disabled": False,
        "role": role,
    }
    return {"username": username, "hashed_password": hashed, "role": role}


# --- Book Endpoints ---


@app.get("/books", response_model=list[Book])
async def list_books():
    return books_db


@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(book: Book):
    global next_book_id
    new_book = {"id": next_book_id, **book.model_dump(exclude={"id"})}
    books_db.append(new_book)
    next_book_id += 1
    return new_book


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(book_id: int):
    for i, book in enumerate(books_db):
        if book["id"] == book_id:
            books_db.pop(i)
            return
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Book not found",
    )

Your API now hashes every password before storing it and has a working registration endpoint. Next up, you will use these credentials to issue JWT tokens so users can prove their identity on every request.

Implement JWT Authentication

Your bookstore API now has password hashing and user registration. Users can sign up and their passwords are safely stored as Argon2 hashes. But there is still no way to prove who you are when making a request.

This is where JWT authentication comes in. A JWT (JSON Web Token) is a small, signed piece of data that the server hands to a user after they log in. The user sends it back with every request to prove their identity.

In this step, get ready to:
  • Set up JWT configuration and a token creation function.
  • Build a login endpoint that issues JWT access tokens.
  • Protect the books endpoint so only authenticated users can access it.
Set up JWT configuration and the token creation function

Before you can issue tokens, you need three pieces of configuration: a secret key for signing tokens, the signing algorithm, and an expiration time.

  • In main.py, replace the entire import section at the top of the file (the first three lines) and update the # --- Security Configuration --- section by copying this code:
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pydantic import BaseModel
from pwdlib import PasswordHash

# --- Security Configuration ---

# To generate your own secret key, run:
# python -c "import secrets; print(secrets.token_hex(32))"
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

What does this code do?

  • The new imports bring in jwt for creating and reading tokens, OAuth2PasswordBearer for extracting tokens from requests, and OAuth2PasswordRequestForm for handling login form data.
  • SECRET_KEY is the password your server uses to sign every token. In production, you would generate a unique key and store it as an environment variable.
  • ALGORITHM tells PyJWT to use HS256 (HMAC with SHA-256) for signing.
  • oauth2_scheme tells FastAPI to look for a Bearer token in the Authorization header of incoming requests. It also powers the Authorize button in Swagger docs.
  • Save main.py.
  • Check your terminal. The dev server should auto-reload without errors. You won't see a visible change yet because no endpoint uses these constants.

Seeing an import error?

If the server shows ModuleNotFoundError: No module named 'jwt', make sure your virtual environment is active and that you installed pyjwt (not jwt). The package name on PyPI is pyjwt but the import is import jwt.

Help me fix my import error in main.py

Now you need a function that actually builds the JWT token string. This function takes user data, adds an expiration timestamp, and signs everything with your secret key.

  • In main.py, find the authenticate_user function. Add the create_access_token function directly below it by copying this code:
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    # Copy the data so we don't modify the original
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        # Default to 15 minutes if no expiration provided
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    # Add the expiration claim to the token payload
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

What does this code do?

  • data.copy() makes a safe copy of the token payload so the original dictionary is not mutated.
  • The exp claim is a standard JWT field that tells the server when the token expires. PyJWT automatically rejects expired tokens during decoding.
  • jwt.encode() takes the payload, signs it with your SECRET_KEY using the HS256 algorithm, and returns a compact string like eyJhbGci....
  • Save main.py. You won't see a change yet. The next chunk creates the endpoint that calls this function.
Create the login endpoint and token dependencies

You have a function that creates tokens, but nothing that validates them on incoming requests. You need two dependency functions that FastAPI calls automatically every time a protected endpoint is hit.

  • In main.py, find the line that says # --- App ---. Add this new Dependencies section directly above it by copying this code:
# --- Dependencies ---


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> UserInDB:
    # Prepare a reusable 401 error for any auth failure
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # Decode the JWT and extract the username from the 'sub' claim
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    # Look up the user in the database
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> UserInDB:
    # Block disabled users even if their token is valid
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

What does this code do?

These two functions form a dependency chain that FastAPI runs before any protected endpoint.

  • get_current_user extracts the Bearer token from the request header, decodes it with jwt.decode(), and looks up the user in the database. If anything goes wrong (expired token, tampered payload, missing user), it raises a 401 Unauthorized error.
  • get_current_active_user adds an extra check on top: even if the token is valid, a disabled user gets blocked with a 400 error.
  • The sub claim (short for "subject") is a standard JWT field that identifies who the token belongs to. Your login endpoint will set this to the username.
  • Save main.py. You won't see a change yet. The next chunk creates the login endpoint that issues the tokens these dependencies validate.

Now you need the endpoint where users actually log in. This endpoint accepts a username and password, verifies them, and returns a signed JWT token.

  • In main.py, find the # --- Auth Endpoints --- comment. Add the /token endpoint directly below the comment and above the existing /register endpoint by copying this code:
@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    # Verify the username and password against the fake database
    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"},
        )
    # Create a token that expires after ACCESS_TOKEN_EXPIRE_MINUTES
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")

What does this code do?

  • OAuth2PasswordRequestForm is a FastAPI utility that automatically parses a username and password from the request body. This follows the OAuth2 Password flow standard.
  • The endpoint calls your authenticate_user function from the previous step. If the credentials are wrong, it returns a 401 error.
  • On success, it creates a JWT with the username stored in the sub claim and returns it as access_token in the response. The Swagger docs will use this endpoint automatically when you click the Authorize button.
  • Save main.py.
  • Open your Swagger docs at http://127.0.0.1:8000/docs. You should see a new POST /token endpoint in the list. You should also see an Authorize button with a lock icon in the top-right area of the page.

Don't see the /token endpoint?

Make sure the /token endpoint is placed after app = FastAPI(title="Secure Bookstore API"). If the endpoint is defined before the app variable exists, Python will raise a NameError.

Help me fix my /token endpoint in FastAPI

Protect the books endpoint and test authentication

The login endpoint works, but your book endpoints are still wide open. Anyone can list, create, or delete books without a token. Time to lock down the GET /books endpoint by requiring authentication.

  • In main.py, scroll down to the # --- Book Endpoints --- section.
  • Find this code:
@app.get("/books", response_model=list[Book])
async def list_books():
    return books_db
  • Replace the list_books function signature to require the get_current_active_user dependency, so the function now looks like this:
@app.get("/books", response_model=list[Book])
async def list_books(
    current_user: Annotated[UserInDB, Depends(get_current_active_user)],
):
    return books_db

What changed?

Adding current_user: Annotated[UserInDB, Depends(get_current_active_user)] as a parameter tells FastAPI to run the entire dependency chain (extract token, decode it, look up user, check if active) before the endpoint body executes. If any check fails, FastAPI returns an error before your function ever runs.

  • Save main.py.

Now test the full authentication flow in Swagger docs to see your security in action.

  • Refresh your Swagger docs at http://127.0.0.1:8000/docs.
  • Click the GET /books endpoint to expand it.
  • Click Try it out, then click Execute.

You should see a 401 Unauthorized response with the message "Not authenticated". The lock is working. Now log in and try again.

  • Click the Authorize button (lock icon) at the top of the Swagger docs page.
  • Enter johndoe in the username field.
  • Enter secret in the password field.
  • Click Authorize.

Swagger calls your /token endpoint behind the scenes, gets a JWT back, and attaches it to every future request as a Bearer token.

  • Click Close to dismiss the Authorize dialog.
  • Go back to the GET /books endpoint and click Execute again.

This time you should see a 200 OK response with the list of books. Your endpoint is now protected by JWT authentication.

Still getting 401 after authorizing?

Make sure you typed the password as secret (all lowercase). If you see Incorrect username or password in the Authorize dialog, the credentials did not match the fake_users_db entries.

If you restarted the dev server between logging in and executing the request, the token may have been lost. Click Authorize again and re-enter the credentials.

Help me debug my JWT authentication in FastAPI

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py.

ⓧ I'd like to double check the full code

Here is the complete main.py file at this point in the project. Compare yours against it to spot any differences.

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

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pydantic import BaseModel
from pwdlib import PasswordHash

# --- Security Configuration ---

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# --- Models ---


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


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


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None
    role: str = "user"


class UserInDB(User):
    hashed_password: str


class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int


# --- Fake Databases ---

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "admin",
    },
    "janedoe": {
        "username": "janedoe",
        "full_name": "Jane Doe",
        "email": "janedoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "user",
    },
}

books_db: list[dict] = [
    {"id": 1, "title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "year": 1925, "isbn": None},
    {"id": 2, "title": "To Kill a Mockingbird", "author": "Harper Lee", "year": 1960, "isbn": None},
]
next_book_id = 3

# --- Helper Functions ---


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


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


def get_user(db: dict, username: str) -> UserInDB | None:
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
    return None


def authenticate_user(db: dict, username: str, password: str) -> UserInDB | bool:
    user = get_user(db, username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


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=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


# --- Dependencies ---


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> UserInDB:
    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])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


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


# --- App ---

app = FastAPI(title="Secure Bookstore API")

# --- Auth Endpoints ---


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.post("/register")
async def register_user(
    username: str,
    password: str,
    role: str = "user",
):
    if username in fake_users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    hashed = get_password_hash(password)
    fake_users_db[username] = {
        "username": username,
        "full_name": None,
        "email": None,
        "hashed_password": hashed,
        "disabled": False,
        "role": role,
    }
    return {"username": username, "hashed_password": hashed, "role": role}


# --- Book Endpoints ---


@app.get("/books", response_model=list[Book])
async def list_books(
    current_user: Annotated[UserInDB, Depends(get_current_active_user)],
):
    return books_db


@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(book: Book):
    global next_book_id
    new_book = {"id": next_book_id, **book.model_dump(exclude={"id"})}
    books_db.append(new_book)
    next_book_id += 1
    return new_book


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(book_id: int):
    for i, book in enumerate(books_db):
        if book["id"] == book_id:
            books_db.pop(i)
            return
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Book not found",
    )

Your API now requires users to prove who they are before accessing data. Next up, you will add role-based access control so only admin users can create or delete books.

Add Role-Based Access Control

Your API now requires a valid JWT token before anyone can access the book endpoints. That's a strong first layer of security.

However, every authenticated user has the same power right now. A regular user like janedoe can delete books just as easily as an admin like johndoe. Role-based access control (RBAC) fixes this by restricting specific actions to specific roles.

In this step, get ready to:
  • Build a require_role dependency factory that checks user permissions.
  • Protect the write endpoints so only admin users can create and delete books.
  • Test access control by logging in as different users in the Swagger docs.
Review your existing role infrastructure

Your project already has the building blocks for role-based access. The User model includes a role field, and your fake database assigns "admin" to johndoe and "user" to janedoe.

How do roles flow through the JWT system?

When a user logs in, the /token endpoint encodes their username into the JWT. When a protected endpoint is called, get_current_user decodes the token, looks up the full user from the database, and returns the complete UserInDB object. That object already includes the user's role.

The missing piece is enforcement. Nothing currently checks whether a user's role permits the action they're trying to perform.

Create the require_role dependency factory

FastAPI's dependency injection system lets you create reusable checks that run before your endpoint logic. You'll build a factory function that generates a role-checking dependency on the fly.

  • In main.py, scroll down to the # --- Dependencies --- section.
  • Find the end of the get_current_active_user function, then add the following code below it:
def require_role(*allowed_roles: str):
    # Inner function that FastAPI calls as a dependency
    async def role_checker(
        current_user: Annotated[UserInDB, Depends(get_current_active_user)],
    ) -> UserInDB:
        # Block the request if the user's role isn't allowed
        if current_user.role not in allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="You do not have permission to perform this action",
            )
        return current_user
    return role_checker

What does this code do?

require_role is a factory function. You call it with one or more role names (like "admin"), and it returns a brand-new dependency function tailored to those roles.

  • *allowed_roles: str accepts any number of role strings, so you can write require_role("admin") or require_role("admin", "editor").
  • The inner role_checker calls get_current_active_user first, which means it automatically handles token validation and disabled-user checks.
  • If the user's role is not in the allowed list, the function raises a 403 Forbidden error. This is distinct from 401 Unauthorized. A 401 means "I don't know who you are." A 403 means "I know who you are, but you're not allowed."
  • Save main.py.

You won't see a change yet. The next substep uses this dependency on your book endpoints.

Seeing a syntax error after saving?

  • Make sure require_role is at the same indentation level as get_current_active_user. It's a top-level function, not nested inside another function.
  • Check that the inner role_checker function is indented one level inside require_role, and the return role_checker line is back at the require_role level.

Why is my require_role function causing errors in FastAPI?

Protect the book endpoints with role checks

Now you'll wire require_role("admin") into the endpoints that should be admin-only. The GET /books endpoint stays accessible to any authenticated user.

  • In main.py, scroll down to the # --- Book Endpoints --- section.
  • Find the create_book function signature. It currently looks like this:
async def create_book(book: Book):
  • Replace that line with the following, which adds a current_user parameter that requires the "admin" role:
async def create_book(
    book: Book,
    current_user: Annotated[UserInDB, Depends(require_role("admin"))],
):
  • Now find the delete_book function signature. It currently looks like this:
async def delete_book(book_id: int):
  • Replace that line with the following to apply the same admin check:
async def delete_book(
    book_id: int,
    current_user: Annotated[UserInDB, Depends(require_role("admin"))],
):

Why does GET /books stay the same?

Reading the book catalog is a low-risk operation. Any authenticated user should be able to browse books. The GET /books endpoint already uses get_current_active_user, which verifies the token without checking roles.

Creating and deleting books are write operations that change data. Restricting those to admins follows the principle of least privilege.

  • Save main.py. The dev server will auto-reload.

Seeing an error in the terminal after saving?

  • Check that the require_role function is defined above the # --- App --- section. Python needs the function to exist before it's referenced in your endpoint definitions.
  • Make sure the rest of the function body (the lines after def create_book(...): and def delete_book(...):) is still indented correctly. Only the function signature changed.

Help me debug my FastAPI role-based access control setup.

Test RBAC by logging in as different users

The real test is whether a regular user gets blocked from write operations while an admin can proceed normally.

  • Open the Swagger docs at http://127.0.0.1:8000/docs.
  • Click the green Authorize button at the top of the page.
  • Enter janedoe as the username and secret as the password.
  • Click Authorize, then click Close.
  • Expand the POST /books endpoint and click Try it out.
  • Click Execute to send the request.

You should see a 403 Forbidden response with the message "You do not have permission to perform this action". The role check is working.

Now test with the admin account to confirm that admins can still create books.

  • Click the Authorize button again.
  • Click Logout to clear the current session.
  • Enter johndoe as the username and secret as the password.
  • Click Authorize, then click Close.
  • Expand the POST /books endpoint and click Try it out again.
  • Click Execute.

This time you should see a 201 Created response with the new book returned in the response body. Admins have full access.

Still getting 403 as johndoe?

  • Make sure you clicked Logout before re-authorizing. Swagger caches the previous token, so you must log out first.
  • Check that johndoe in your fake_users_db has "role": "admin". If the role is missing or misspelled, the check will fail.
  • Verify that require_role("admin") uses a lowercase "admin" string. It must match the role value in the database exactly.

Why am I still getting a 403 error when logged in as johndoe?

✔️ Awesome, I've got everything!

Double check you've saved main.py and your dev server reloaded without errors.

ⓧ I'd like to double check the full code

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

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pydantic import BaseModel
from pwdlib import PasswordHash

# --- Security Configuration ---

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# --- Models ---


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


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


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None
    role: str = "user"


class UserInDB(User):
    hashed_password: str


class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int


# --- Fake Databases ---

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "admin",
    },
    "janedoe": {
        "username": "janedoe",
        "full_name": "Jane Doe",
        "email": "janedoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "user",
    },
}

books_db: list[dict] = [
    {"id": 1, "title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "year": 1925, "isbn": None},
    {"id": 2, "title": "To Kill a Mockingbird", "author": "Harper Lee", "year": 1960, "isbn": None},
]
next_book_id = 3

# --- Helper Functions ---


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


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


def get_user(db: dict, username: str) -> UserInDB | None:
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
    return None


def authenticate_user(db: dict, username: str, password: str) -> UserInDB | bool:
    user = get_user(db, username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


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=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


# --- Dependencies ---


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> UserInDB:
    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])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


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


def require_role(*allowed_roles: str):
    async def role_checker(
        current_user: Annotated[UserInDB, Depends(get_current_active_user)],
    ) -> UserInDB:
        if current_user.role not in allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="You do not have permission to perform this action",
            )
        return current_user
    return role_checker


# --- App ---

app = FastAPI(title="Secure Bookstore API")

# --- Auth Endpoints ---


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.post("/register")
async def register_user(
    username: str,
    password: str,
    role: str = "user",
):
    if username in fake_users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    hashed = get_password_hash(password)
    fake_users_db[username] = {
        "username": username,
        "full_name": None,
        "email": None,
        "hashed_password": hashed,
        "disabled": False,
        "role": role,
    }
    return {"username": username, "hashed_password": hashed, "role": role}


# --- Book Endpoints ---


@app.get("/books", response_model=list[Book])
async def list_books(
    current_user: Annotated[UserInDB, Depends(get_current_active_user)],
):
    return books_db


@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(
    book: Book,
    current_user: Annotated[UserInDB, Depends(require_role("admin"))],
):
    global next_book_id
    new_book = {"id": next_book_id, **book.model_dump(exclude={"id"})}
    books_db.append(new_book)
    next_book_id += 1
    return new_book


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(
    book_id: int,
    current_user: Annotated[UserInDB, Depends(require_role("admin"))],
):
    for i, book in enumerate(books_db):
        if book["id"] == book_id:
            books_db.pop(i)
            return
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Book not found",
    )

Your API now enforces who can do what. Only admins can create or delete books, and regular users are blocked with a clear 403 response. Next up, you'll harden the input validation to make sure even admins can't submit malformed data.

Harden Input Validation with Pydantic

Your API now has authentication and role-based access control. Only logged-in users can browse books, and only admins can create or delete them.

There is still a gap. Right now, an admin could create a book with a completely empty title, a year of 99999, or a garbage ISBN. The API accepts anything as long as the types match. In this step, you will add strict Pydantic Field constraints that reject malformed data before it ever reaches your endpoint logic.

In this step, get ready to:
  • Upgrade the Book model with strict Field constraints on title, author, year, and ISBN.
  • Create separate input and output models to control exactly what data enters and leaves your API.
  • Test validation by sending invalid data and observing automatic 422 error responses.
Add Field constraints to the Book model

Pydantic's Field function lets you attach validation rules directly to each model attribute. Instead of writing manual checks inside your endpoint, the model itself rejects bad data before your code even runs.

  • In main.py, find this import line near the top of the file:
from pydantic import BaseModel
  • Add Field to the import, so the line now looks like this:
from pydantic import BaseModel, Field

Now you will replace the current Book model with a new BookCreate model that enforces strict constraints on every field.

  • Find the existing Book model in the # --- Models --- section:
class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int
  • Delete the entire Book class and replace it with this BookCreate model:
class BookCreate(BaseModel):
    # Title must be between 1 and 200 characters
    title: str = Field(min_length=1, max_length=200)
    # Author must be between 1 and 100 characters
    author: str = Field(min_length=1, max_length=100)
    # Year must be a realistic value
    year: int = Field(ge=1000, le=2026)
    # Optional 13-digit ISBN, validated with a regex pattern
    isbn: str | None = Field(default=None, pattern=r"^\d{13}$")

What do these constraints do?

  • min_length=1 on title and author blocks empty strings from sneaking in.
  • ge=1000, le=2026 on year means "greater than or equal to 1000" and "less than or equal to 2026". A year of 99999 or -5 is rejected automatically.
  • pattern=r"^\d{13}$" on isbn enforces exactly 13 digits. The field is optional (None is allowed), but if a value is provided, it must match the pattern.
  • Save main.py.

Your dev server will show errors at this point because the endpoints still reference the old Book model. You will fix this in the next substep.

Seeing a syntax error instead?

  • Make sure your import line reads from pydantic import BaseModel, Field. A missing Field import will cause a NameError.
  • Check the regex pattern string uses a raw string prefix: r"^\d{13}$". Without the r, Python interprets the backslash as an escape character.
  • Help me fix a syntax error in my BookCreate model.
Separate input and output models

The old Book model included an id field, which meant clients could send their own ID when creating a book. That is a security problem because clients should never control server-generated identifiers.

The fix is to use separate models for input and output. BookCreate handles incoming data (no id), and a new BookResponse model adds the id for API responses.

  • In main.py, add the BookResponse model directly below the BookCreate class you just created:
class BookResponse(BookCreate):
    # Server-generated ID, only included in responses
    id: int

Why does BookResponse extend BookCreate?

BookResponse inherits all the fields from BookCreate (title, author, year, isbn) and adds id on top. This way you define your fields once and reuse them across both models.

Now you need to update the book endpoints to use these new models. There are four small changes across two endpoints.

  • Scroll down to the # --- Book Endpoints --- section and find the GET /books endpoint. Find this line:
@app.get("/books", response_model=list[Book])
  • Replace Book with BookResponse, so the line now looks like this:
@app.get("/books", response_model=list[BookResponse])
  • Next, find the POST /books endpoint. Find this decorator line:
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
  • Replace response_model=Book with response_model=BookResponse, so the line now looks like this:
@app.post("/books", response_model=BookResponse, status_code=status.HTTP_201_CREATED)
  • Inside the same create_book function, find this parameter line:
    book: Book,
  • Replace Book with BookCreate, so the line now looks like this:
    book: BookCreate,
  • Finally, find this line inside create_book:
    new_book = {"id": next_book_id, **book.model_dump(exclude={"id"})}
  • Remove exclude={"id"} since BookCreate no longer has an id field. The line should now look like this:
    new_book = {"id": next_book_id, **book.model_dump()}
  • Save main.py.
  • Check your terminal where the dev server is running. It should reload without errors.

Dev server showing a NameError?

  • If you see NameError: name 'Book' is not defined, you still have a reference to the old Book class somewhere. Search your file for Book and make sure every reference is now either BookCreate or BookResponse.
  • Make sure BookResponse is defined after BookCreate. Python reads top to bottom, so BookResponse(BookCreate) fails if BookCreate has not been declared yet.
  • Help me fix a NameError after updating my book models.

✔️ Awesome, I've got everything!

Double check you have saved main.py and that your dev server reloaded without errors.

ⓧ I'd like to double check the full code

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

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pydantic import BaseModel, Field
from pwdlib import PasswordHash

# --- Security Configuration ---

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# --- Models ---


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


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


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None
    role: str = "user"


class UserInDB(User):
    hashed_password: str


class BookCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    author: str = Field(min_length=1, max_length=100)
    year: int = Field(ge=1000, le=2026)
    isbn: str | None = Field(default=None, pattern=r"^\d{13}$")


class BookResponse(BookCreate):
    id: int


# --- Fake Databases ---

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "admin",
    },
    "janedoe": {
        "username": "janedoe",
        "full_name": "Jane Doe",
        "email": "janedoe@example.com",
        "hashed_password": password_hash.hash("secret"),
        "disabled": False,
        "role": "user",
    },
}

books_db: list[dict] = [
    {"id": 1, "title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "year": 1925, "isbn": None},
    {"id": 2, "title": "To Kill a Mockingbird", "author": "Harper Lee", "year": 1960, "isbn": None},
]
next_book_id = 3

# --- Helper Functions ---


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


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


def get_user(db: dict, username: str) -> UserInDB | None:
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
    return None


def authenticate_user(db: dict, username: str, password: str) -> UserInDB | bool:
    user = get_user(db, username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


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=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


# --- Dependencies ---


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> UserInDB:
    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])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


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


def require_role(*allowed_roles: str):
    async def role_checker(
        current_user: Annotated[UserInDB, Depends(get_current_active_user)],
    ) -> UserInDB:
        if current_user.role not in allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="You do not have permission to perform this action",
            )
        return current_user
    return role_checker


# --- App ---

app = FastAPI(title="Secure Bookstore API")

# --- Auth Endpoints ---


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.post("/register")
async def register_user(
    username: str,
    password: str,
    role: str = "user",
):
    if username in fake_users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    hashed = get_password_hash(password)
    fake_users_db[username] = {
        "username": username,
        "full_name": None,
        "email": None,
        "hashed_password": hashed,
        "disabled": False,
        "role": role,
    }
    return {"username": username, "hashed_password": hashed, "role": role}


# --- Book Endpoints ---


@app.get("/books", response_model=list[BookResponse])
async def list_books(
    current_user: Annotated[UserInDB, Depends(get_current_active_user)],
):
    return books_db


@app.post("/books", response_model=BookResponse, status_code=status.HTTP_201_CREATED)
async def create_book(
    book: BookCreate,
    current_user: Annotated[UserInDB, Depends(require_role("admin"))],
):
    global next_book_id
    new_book = {"id": next_book_id, **book.model_dump()}
    books_db.append(new_book)
    next_book_id += 1
    return new_book


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(
    book_id: int,
    current_user: Annotated[UserInDB, Depends(require_role("admin"))],
):
    for i, book in enumerate(books_db):
        if book["id"] == book_id:
            books_db.pop(i)
            return
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Book not found",
    )
Test validation with invalid data

Time to see your constraints in action. You will send deliberately invalid data through Swagger docs and watch FastAPI reject it with detailed error messages.

  • Open your browser to http://127.0.0.1:8000/docs.
  • Click the Authorize button at the top right of the page.
  • Enter johndoe as the username and secret as the password.
  • Click Authorize, then Close.

First, try submitting a book with an empty title.

  • Expand the POST /books endpoint and click Try it out.
  • In the request body, set title to an empty string "", author to "Test", and year to 2024.
  • Click Execute.

You should see a 422 Validation Error response. The error body tells you exactly which field failed and why, including a message like String should have at least 1 character.

Now try an impossible year.

  • Change the title to "Test Book" and set year to 3000.
  • Click Execute.

Another 422 Validation Error. This time the message indicates the year must be less than or equal to 2026.

Finally, try a malformed ISBN.

  • Set year back to 2024 and add an isbn field with the value "abc".
  • Click Execute.

A third 422 Validation Error. The ISBN must be exactly 13 digits, and "abc" does not match the pattern.

Why is automatic validation so powerful?

Without Pydantic Field constraints, you would need to write manual if checks at the start of every endpoint. Each one is a chance to forget a check or get the logic wrong.

With Field constraints, the validation runs before your endpoint code. FastAPI returns a standardized 422 response with a detailed JSON body that tells the client exactly which field failed and why.

Now confirm that valid data still works.

  • In the POST /books request body, enter a valid book with title set to "1984", author set to "George Orwell", year set to 1949, and remove the isbn field (or set it to null).
  • Click Execute.

You should see a 201 Created response with the book data and a server-generated id field. Valid data passes through, invalid data is blocked.

Getting a 401 Unauthorized instead of 422?

  • Your token may have expired. Click the Authorize button again, click Logout, and re-enter your credentials (johndoe / secret).
  • If you restarted the dev server, the in-memory token is invalidated. Log in again through Authorize.
  • Help me troubleshoot a 401 error when testing input validation.

Your API now rejects bad data at the door. Every field has clear rules, and FastAPI enforces them automatically with descriptive error messages. Next up, you will tackle the Secret Mission and add rate limiting and CORS to round out your API's security.

Secret mission

Add Rate Limiting and CORS

Your API is authenticated and authorized, but it's still vulnerable to brute-force attacks and cross-origin abuse. In this secret mission, you'll add rate limiting with slowapi to throttle login attempts and configure CORS middleware to control which frontends can talk to your API.

Before you go

Clean Up Your Resources

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

Resources you used:

  • Project folder on your Desktop containing main.py.
  • Python virtual environment (venv folder inside your project).
  • Installed Python packages: fastapi, pyjwt, pwdlib (and slowapi if you completed the Secret Mission).
  • FastAPI development server process.

Keep everything running

No action needed. Choose this if you are still actively building on your bookstore API or plan to extend it soon.

  • Your FastAPI dev server keeps running in the terminal where you started it.
  • Your virtual environment stays active in that terminal session.
  • All your project files remain on your Desktop.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your files and virtual environment stay intact so you can pick up where you left off.

  • In the terminal running your FastAPI server, press Ctrl+C to stop the server.
  • 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 everything you built in this project.

  • In the terminal running your FastAPI server, press Ctrl+C to stop the server.
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the entire project folder by running:
cd ~/Desktop
rm -rf secure-bookstore-api

On Windows, you can also delete the project folder from File Explorer by right-clicking the folder and selecting Delete.

Mission accomplished

That's a Wrap!

Nice work! You just built a production-grade secured REST API from scratch using FastAPI. Your bookstore API now requires authentication, enforces permissions, and validates every piece of incoming data.

You've learned how to:

  • Hashed passwords with Argon2 via pwdlib and implemented JWT authentication using the OAuth2 Password flow to protect your API endpoints.
  • Built role-based access control that restricts write operations to admin users only, using FastAPI's dependency injection system.
  • Hardened input validation with Pydantic Field constraints to block malformed or dangerous requests before they reach your application logic.
  • Secret Mission: Added rate limiting with slowapi to prevent brute-force attacks and configured CORS middleware for cross-origin security.

Ready to quiz yourself?