Secure a Python REST API with FastAPI

Build a FastAPI book catalog API secured with JWT auth and role-based access.

Introduction

30 Second Summary

Every time you log in to an app, a chain of security checks runs behind the scenes to verify who you are. Building these security layers from scratch is one of the most valuable backend skills you can have.

In this project, you will build a secure REST API using Python and FastAPI. You will start with an unprotected book catalog, then layer on real-world security one piece at a time: password hashing, JWT authentication, role-based access control, rate limiting, and security headers.

What You'll Build

A secure book catalog API built with FastAPI that demonstrates production-grade authentication, authorization, and abuse prevention.

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

A FastAPI application with 6+ endpoints secured by JWT authentication.
A user system with Argon2 password hashing and role-based access control (admin vs. reader).
Rate limiting, CORS configuration, and security headers middleware to prevent common attacks.
Secret Mission: Add a token refresh mechanism with short-lived access tokens and longer-lived refresh tokens.

Are there any prerequisites?

You should have basic Python knowledge (functions, classes, dictionaries). Familiarity with HTTP concepts like GET, POST, and status codes is helpful but not required.

Set Up Your Project and Install Dependencies

Before writing any security code, you need a clean workspace with the right tools installed. A Python virtual environment keeps this project's packages separate from the rest of your system, so nothing conflicts with your other projects.

In this step, you will create your project folder, install FastAPI and its companion security libraries, and confirm everything works by launching a basic API server.

In this step, get ready to:

Create a project folder and Python virtual environment.
Install FastAPI, PyJWT, pwdlib, and SlowAPI.
Launch a basic API server and view its interactive documentation.

Create your project folder and virtual environment

A virtual environment is an isolated Python environment that stores packages for one project only. This prevents version conflicts between different projects on your machine.

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 by running:

mkdir secure-api

Confirm the folder was created by running:

ls

You should see secure-api in the output.

Open the secure-api folder in VS Code by running:

code secure-api

VS Code opens with the secure-api folder as your workspace. You can close the system terminal now.

Seeing 'command not found' for code?

The code command needs to be installed separately. Open VS Code manually instead.

Press Cmd+Space (macOS) or the Windows key (Windows) to open search.
Type Visual Studio Code and press Enter.
Click File in the top menu bar.
Click Open Folder.
Navigate to your Desktop, select the secure-api folder, and click Open.

help me set up the code command in my terminal

In VS Code, open a new terminal by pressing Ctrl+` (Windows) or Cmd+` (macOS).

The integrated terminal opens at the bottom of VS Code, already inside the secure-api folder.

Create a Python virtual environment by running:

python -m venv venv

Confirm the virtual environment was created by running:

ls

You should see a venv folder in the output.

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

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

Your terminal prompt now starts with (venv), which means the virtual environment is active.

Don't see (venv) in your prompt?

On Windows, if you see a "running scripts is disabled" error, your PowerShell execution policy is blocking the activation script. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned and try activating again.
Make sure you are inside the secure-api folder. Run pwd to check your current directory.
If your system uses python3 instead of python, delete the venv folder and recreate it with python3 -m venv venv.

help me activate my Python virtual environment

Install the project dependencies

Your project needs four packages. Each one handles a different layer of API security.

Install all four packages at once by running:

pip install "fastapi[standard]" pyjwt "pwdlib[argon2]" slowapi

What does each package do?

fastapi[standard] installs FastAPI along with Uvicorn, the server that runs your API.
pyjwt handles creating and verifying JWT tokens for authentication.
pwdlib[argon2] provides password hashing with Argon2, the strongest hashing algorithm available today.
slowapi adds rate limiting to protect your endpoints from brute-force attacks.

When the install finishes, you should see a "Successfully installed" message listing all the packages and their dependencies.

Seeing 'pip: command not found'?

Make sure your virtual environment is active. Your terminal prompt should start with (venv).
If your system uses pip3 instead of pip, try running the command with pip3 instead.

help me fix my pip install issue

Create a minimal FastAPI app

With everything installed, you will write a tiny API to confirm the server starts correctly. This is a single health-check endpoint that returns a simple status message.

In the Explorer sidebar on the left, hover over the SECURE-API heading.
Click the New File icon (a page with a plus sign).
Type main.py and press Enter.
Paste the following code into main.py:

from fastapi import FastAPI

# Create the application instance with a descriptive title
app = FastAPI(title="Secure Book Catalog API")

# A simple endpoint to confirm the server is running
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

What does this code do?

FastAPI(title="Secure Book Catalog API") creates your API application. The title appears in the auto-generated documentation.
@app.get("/health") registers a GET endpoint at the /health path. When someone sends a GET request to that URL, FastAPI calls the function below the decorator.
The function returns a Python dictionary, which FastAPI automatically converts to a JSON response.

Save main.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete main.py file so far. Make sure yours matches exactly.

from fastapi import FastAPI

app = FastAPI(title="Secure Book Catalog API")

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

Start the development server by running this command in your VS Code terminal:

fastapi dev main.py

What does this command do?

fastapi dev starts FastAPI's built-in development server powered by Uvicorn. It watches your files and automatically restarts the server whenever you save a change.

You should see output in the terminal that includes a line like Uvicorn running on http://127.0.0.1:8000.

Open your browser and paste http://127.0.0.1:8000/docs into the address bar.

You should see the Swagger UI documentation page with the title "Secure Book Catalog API" at the top. Your GET /health endpoint is listed on the page.

Click on the GET /health row to expand it.
Click Try it out.
Click Execute.

You should see a response body of {"status": "healthy"} with a 200 status code. Your API is running.

Page won't load or server won't start?

Make sure the server is still running in your terminal. If you see an error, check that your virtual environment is active (look for (venv) in the prompt).
If you see ModuleNotFoundError: No module named 'fastapi', your packages were installed outside the virtual environment. Activate the environment and run the pip install command again.
Check that the URL in your browser is exactly http://127.0.0.1:8000/docs with no extra characters.

help me get my FastAPI server running

Your development environment is ready and your first endpoint is live. Next up, you will build out the full book catalog API with create, read, update, and delete operations.

Build the Unprotected Book Catalog API

Your project is set up and the dev server is running. Now it is time to build something real.

In this step, you will create a fully functional book catalog API with five endpoints for creating, reading, updating, and deleting books. The API will work perfectly, but it will have one massive problem: zero security. Anyone who finds your server can do anything they want.

In this step, get ready to:

Define Pydantic models for your book data.
Create five CRUD endpoints for the book catalog.
Test all endpoints through the Swagger UI.

Define Pydantic models and set up the data store

Every API endpoint needs to know the exact shape of the data it accepts and returns. FastAPI uses Pydantic models to define that shape and automatically reject any request that does not match.

You will create two models. BookCreate is the shape of data a user sends when adding a book. Book extends it by adding an auto-generated id field for the API response.

Replace the entire contents of main.py with the following code:

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

# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
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=2100)
    isbn: str = Field(min_length=10, max_length=13)


class Book(BookCreate):
    id: int


# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
books_db: list[Book] = []
book_id_counter: int = 1

# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = FastAPI(title="Secure Book Catalog API")


# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

What does this code do?

BookCreate defines four fields with strict validation. Field(min_length=1, max_length=200) ensures a title is between 1 and 200 characters. Field(ge=1000, le=2100) constrains the year to a realistic range.

Book inherits from BookCreate and adds an id field. The user never chooses the ID. Your API assigns it automatically.

books_db is a plain Python list acting as your in-memory database. book_id_counter tracks the next ID to assign. In a production app you would use a real database, but a list keeps the focus on security concepts.

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

Seeing an error in the terminal?

Check that your import line includes HTTPException and status from fastapi, and BaseModel and Field from pydantic.
Make sure you replaced the entire file contents, including the old from fastapi import FastAPI line.

Still stuck? Ask for help.

Create the CRUD endpoints

CRUD stands for Create, Read, Update, and Delete. These are the four basic operations any API needs to manage a resource. You will build five endpoints that cover all four operations for books.

In main.py, find the # Health check section comment near the bottom of the file.
Add the following read endpoints above that comment:

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


@app.get("/books/{book_id}", response_model=Book)
async def get_book(book_id: int):
    for book in books_db:
        if book.id == book_id:
            return book
    raise HTTPException(status_code=404, detail="Book not found")

What do these endpoints do?

GET /books returns the entire books_db list. The response_model=list[Book] tells FastAPI to format the response as a list of Book objects.
GET /books/{book_id} searches through the list for a matching ID. If it finds one, it returns the book. If not, it raises a 404 Not Found error.

Save main.py.
Refresh http://127.0.0.1:8000/docs in your browser. You should see the two new GET /books endpoints listed alongside the existing /health endpoint.

Don't see the new endpoints in /docs?

Make sure you pasted the code above the # Health check comment, not below it.
Check your terminal for errors. A syntax error or indentation issue will prevent the server from reloading.

Need a hand? Ask for help.

Add the create endpoint below the get_book function you just wrote, above the # Health check section:

@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(book_data: BookCreate):
    global book_id_counter
    book = Book(id=book_id_counter, **book_data.model_dump())
    book_id_counter += 1
    books_db.append(book)
    return book

What does this endpoint do?

status_code=status.HTTP_201_CREATED tells FastAPI to return a 201 status code instead of the default 200. A 201 signals that a new resource was successfully created.
book_data.model_dump() converts the Pydantic model to a dictionary. The ** unpacks it so each field becomes a keyword argument to the Book constructor.
global book_id_counter is needed because the function modifies a module-level variable. Without it, Python would treat book_id_counter as a local variable inside the function.

Save main.py and check your terminal for a clean reload.
Refresh /docs. You should now see a green POST /books endpoint in the list.

Seeing a NameError for book_id_counter?

Make sure the global book_id_counter line is indented inside the function body. If it is at the module level, Python will raise an error.

Ask for help debugging.

Add the update and delete endpoints below the create_book function, still above the # Health check section:

@app.put("/books/{book_id}", response_model=Book)
async def update_book(book_id: int, book_data: BookCreate):
    for i, book in enumerate(books_db):
        if book.id == book_id:
            updated = Book(id=book_id, **book_data.model_dump())
            books_db[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Book not found")


@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=404, detail="Book not found")

What do these endpoints do?

PUT /books/{book_id} finds the book by ID, creates a new Book with the updated fields, and replaces the old entry in the list.
DELETE /books/{book_id} finds the book by ID and removes it. The 204 No Content status code means the operation succeeded but there is no response body to return.
Both endpoints use enumerate() to loop through the list with both the index and the book. The index is needed to replace or remove the item from books_db.

Save main.py.
Refresh /docs. You should now see all five book endpoints: two GET, one POST, one PUT, and one DELETE.

Server won't reload or endpoints are missing?

Make sure each endpoint function is at the module level (no extra indentation). They should start at the same column as the existing health_check function.
Confirm there are no duplicate function names. Each endpoint function must have a unique name.

Ask for help debugging.

✔️ Awesome, I've got everything!

Double check you have saved main.py and all five endpoints are visible at /docs.

ⓧ I'd like to double check the full code

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

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

# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
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=2100)
    isbn: str = Field(min_length=10, max_length=13)


class Book(BookCreate):
    id: int


# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
books_db: list[Book] = []
book_id_counter: int = 1

# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = FastAPI(title="Secure Book Catalog API")


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


@app.get("/books/{book_id}", response_model=Book)
async def get_book(book_id: int):
    for book in books_db:
        if book.id == book_id:
            return book
    raise HTTPException(status_code=404, detail="Book not found")


@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(book_data: BookCreate):
    global book_id_counter
    book = Book(id=book_id_counter, **book_data.model_dump())
    book_id_counter += 1
    books_db.append(book)
    return book


@app.put("/books/{book_id}", response_model=Book)
async def update_book(book_id: int, book_data: BookCreate):
    for i, book in enumerate(books_db):
        if book.id == book_id:
            updated = Book(id=book_id, **book_data.model_dump())
            books_db[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Book not found")


@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=404, detail="Book not found")


# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

Test all endpoints through the Swagger UI

Your endpoints exist, but you have not proven they actually work. The Swagger UI at /docs lets you send real requests to your API and see the responses without any external tools.

Open http://127.0.0.1:8000/docs in your browser.
Click on POST /books to expand it.
Click Try it out.
Replace the example request body with this:

{
  "title": "The Pragmatic Programmer",
  "author": "David Thomas",
  "year": 1999,
  "isbn": "9780135957059"
}

Click Execute. You should see a 201 response with the book data and an auto-assigned id of 1.

Now test the read endpoints.

Expand GET /books, click Try it out, then click Execute. You should see a list containing the book you just created.
Expand GET /books/{book_id}, click Try it out, enter 1 for the book_id field, then click Execute. You should see the single book returned.

Now test update and delete.

Expand PUT /books/{book_id}, click Try it out, enter 1 for book_id, and change the year to 2019 in the request body.
Click Execute. You should see a 200 response with the updated book data showing "year": 2019.
Expand DELETE /books/{book_id}, click Try it out, enter 1 for book_id, and click Execute. You should see a 204 response with no body.
Run GET /books one more time. The response should be an empty list [], confirming the book was deleted.

Every single one of those operations worked. You did not log in. You did not provide a password or token. Anyone who finds this server can create, modify, and delete every book in the catalog.

Why build the insecure version first?

Starting without security is deliberate. It lets you see each CRUD operation working in isolation before layering on authentication and authorization. When something breaks later, you will know the base API was solid.

The OWASP API Security Top 10 lists Broken Authentication and Broken Function Level Authorization as two of the most common API vulnerabilities. You just experienced both firsthand.

Your book catalog API is fully functional. The problem is that it is fully open. In the next step, you will start locking it down by adding password hashing and user registration.

Add Password Hashing and User Registration

In the last step, you built a fully working book catalog API with five CRUD endpoints. The problem is that anyone who finds the URL can create, edit, or delete books. There is no concept of a user, a password, or any kind of identity.

Before we can lock down those endpoints, we need a user system. In this step, you will set up password hashing with Argon2, define user models, and create a registration endpoint.

In this step, get ready to:

Set up password hashing with Argon2 so passwords are never stored in plain text.
Define Pydantic models for users with role-based fields.
Create a registration endpoint that securely stores new users.

Set up password hashing and user models

Storing passwords in plain text is one of the most dangerous security mistakes in web development. If an attacker gains access to your database, every user's password is immediately compromised.

Instead, we will hash passwords using Argon2, the winner of the Password Hashing Competition. Hashing is a one-way transformation. The original password can never be recovered from the hash.

In main.py, find the two import lines at the very top of the file.
Add the password hashing import and setup by copying this code directly below the from pydantic import BaseModel, Field line and above the # Pydantic models comment:

from pwdlib import PasswordHash

# ---------------------------------------------------------------------------
# Password hashing
# ---------------------------------------------------------------------------
# Create a hasher that uses Argon2 (the strongest modern algorithm)
password_hash = PasswordHash.recommended()
# Pre-compute a dummy hash to prevent timing attacks during login
DUMMY_HASH = password_hash.hash("dummypassword")


def verify_password(plain_password: str, hashed_password: str) -> bool:
    # Compare a plain-text password against a stored hash
    return password_hash.verify(plain_password, hashed_password)


def get_password_hash(password: str) -> str:
    # Hash a plain-text password for storage
    return password_hash.hash(password)

What does this code do?

PasswordHash.recommended() creates a hasher pre-configured with Argon2. Argon2 is specifically designed to be slow and memory-intensive, which makes brute-force attacks extremely expensive for attackers.

DUMMY_HASH is a pre-computed hash used during login. When someone tries to log in with a username that does not exist, we still run the hash verification against this dummy value. This prevents attackers from measuring response times to figure out which usernames are valid.

Now we need data models that define what a user looks like. Pydantic models control exactly which fields go in and which fields come out of each endpoint.

In the # Pydantic models section, add these three user models above the existing BookCreate class:

# Accepts registration input: username, password, and an optional role
class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    password: str = Field(min_length=8, max_length=128)
    role: str = "reader"


# The safe user model returned in API responses (no password)
class User(BaseModel):
    username: str
    role: str


# Internal model with the hashed password (never sent to clients)
class UserInDB(User):
    hashed_password: str

Why three separate models?

UserCreate is what the client sends during registration. It includes the plain-text password and enforces length constraints.
User is what the API sends back. It deliberately excludes the password so it never leaks to the client.
UserInDB extends User and adds the hashed password. This model is used internally for storage and authentication only.

Scroll down to the # In-memory data stores section.
Add this line above the existing books_db line:

users_db: dict[str, UserInDB] = {}

Save main.py.
Check your terminal. The dev server should auto-restart without any errors.

Seeing an import error in the terminal?

Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
Confirm you installed pwdlib with the Argon2 extra. The correct install command is pip install "pwdlib[argon2]".
Check that the from pwdlib import PasswordHash line is at the top of the file with the other imports, not indented inside a class or function.

Help me debug my pwdlib import error

Create the registration endpoint

With the models and hashing helpers in place, we need an endpoint where new users can sign up. The endpoint will accept a username, password, and optional role, then store the user with a hashed password.

In main.py, find the # Endpoints - Books comment.
Add the registration endpoint by copying this code above that comment:

# ---------------------------------------------------------------------------
# Endpoints - Auth
# ---------------------------------------------------------------------------
@app.post("/register", response_model=User)
async def register(user_data: UserCreate):
    # Reject duplicate usernames
    if user_data.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    # Hash the password before storing it
    hashed = get_password_hash(user_data.password)
    user_in_db = UserInDB(
        username=user_data.username,
        role=user_data.role,
        hashed_password=hashed,
    )
    # Store the user in our in-memory database
    users_db[user_data.username] = user_in_db
    return user_in_db

What does this endpoint do?

The endpoint first checks if the username already exists and returns a 400 Bad Request if it does.
get_password_hash() hashes the plain-text password with Argon2 before storing it. The original password is never saved.
response_model=User tells FastAPI to use the User model (without the hashed password) when building the response. Even though we return user_in_db which has the hash, FastAPI strips it out automatically.

Save main.py.
Check your terminal for a clean restart with no errors.
Refresh http://127.0.0.1:8000/docs in your browser. You should see a new POST /register endpoint in the Swagger UI.

Don't see the /register endpoint?

Make sure the registration code is placed after the app = FastAPI(...) line. Endpoints defined before the app is created will not be registered.
Check that the decorator says @app.post("/register", response_model=User) with no typos in the path.
Look at your terminal for syntax errors. A missing colon or mismatched indentation can silently prevent the server from loading.

Help me figure out why my /register endpoint is not showing up

Test registration through Swagger UI

Time to make sure the registration flow works end to end. You will create two test users: an admin and a reader. These two roles will become important when we add authorization in a later step.

In the Swagger UI at http://127.0.0.1:8000/docs, click the POST /register endpoint to expand it.
Click Try it out.
Replace the request body with this JSON to create an admin user:

{
  "username": "adminuser",
  "password": "supersecure123",
  "role": "admin"
}

Click Execute.

You should see a 200 response with a JSON body that only contains username and role. No hashed password anywhere in the response.

Why is the hashed password missing from the response?

This is the response_model=User setting at work. Even though the function returns user_in_db (which includes the hash), FastAPI filters the response through the User model. Since User only defines username and role, everything else is automatically stripped out.

Now register a second user with the default reader role. Replace the request body with:

{
  "username": "readeruser",
  "password": "mypassword123",
  "role": "reader"
}

Click Execute.

You should see another 200 response with "role": "reader". You now have two users stored in memory: one admin and one reader.

Try registering the same username twice

Click Execute again without changing the body. You should see a 400 Bad Request error with the message "Username already exists". That is the duplicate check in your registration endpoint doing its job.

✔️ Awesome, I've got everything!

Great. Double check that you have saved main.py and your server is running cleanly.

ⓧ 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, Field
from pwdlib import PasswordHash

# ---------------------------------------------------------------------------
# Password hashing
# ---------------------------------------------------------------------------
password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")


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)


# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    password: str = Field(min_length=8, max_length=128)
    role: str = "reader"


class User(BaseModel):
    username: str
    role: str


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=2100)
    isbn: str = Field(min_length=10, max_length=13)


class Book(BookCreate):
    id: int


# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
users_db: dict[str, UserInDB] = {}
books_db: list[Book] = []
book_id_counter: int = 1

# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = FastAPI(title="Secure Book Catalog API")


# ---------------------------------------------------------------------------
# Endpoints - Auth
# ---------------------------------------------------------------------------
@app.post("/register", response_model=User)
async def register(user_data: UserCreate):
    if user_data.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    hashed = get_password_hash(user_data.password)
    user_in_db = UserInDB(
        username=user_data.username,
        role=user_data.role,
        hashed_password=hashed,
    )
    users_db[user_data.username] = user_in_db
    return user_in_db


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


@app.get("/books/{book_id}", response_model=Book)
async def get_book(book_id: int):
    for book in books_db:
        if book.id == book_id:
            return book
    raise HTTPException(status_code=404, detail="Book not found")


@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(book_data: BookCreate):
    global book_id_counter
    book = Book(id=book_id_counter, **book_data.model_dump())
    book_id_counter += 1
    books_db.append(book)
    return book


@app.put("/books/{book_id}", response_model=Book)
async def update_book(book_id: int, book_data: BookCreate):
    for i, book in enumerate(books_db):
        if book.id == book_id:
            updated = Book(id=book_id, **book_data.model_dump())
            books_db[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Book not found")


@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=404, detail="Book not found")


# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

You now have a user system with secure password hashing. Nobody can read the original passwords, even if they get access to your data store. Next up, you will turn those registered users into authenticated sessions by adding JWT tokens.

Implement JWT Authentication

You have user registration with securely hashed passwords. But passwords alone don't protect your API.

Right now, anyone can hit every book endpoint without logging in. You need a way for users to prove their identity on every request. That is exactly what JWT authentication does. A user logs in once, your API issues a signed token, and every subsequent request carries that token as proof.

In this step, get ready to:

Set up JWT security configuration and token models.
Create a login endpoint that issues signed JWT tokens.
Protect all book endpoints so only authenticated users can access them.

Set up JWT configuration

A JSON Web Token (JWT) is a compact, signed string your API hands out when a user logs in. The token contains claims like the username and an expiration time, all signed with a secret key that only your server knows.

Before writing any token logic, you need three configuration values and a secret key. Let's generate the key first.

Generate a random secret key by running this command in your VS Code terminal:

python -c "import secrets; print(secrets.token_hex(32))"

You should see a long string of random hexadecimal characters in the terminal. Copy this string and keep it handy for the next code block.

What is a secret key?

This key is the password your server uses to sign and verify tokens. Anyone who has this key can forge valid tokens, so in a production app you would store it in an environment variable instead of your source code.

Now you'll update your imports and add the security configuration to main.py.

In main.py, find the three import lines at the very top of the file:

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

Replace those three lines with this expanded import block:

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

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

Next, find the password hashing section header that says # Password hashing. Add this new security configuration section directly above it:

# ---------------------------------------------------------------------------
# Security configuration
# ---------------------------------------------------------------------------
# Generate your own key: python -c "import secrets; print(secrets.token_hex(32))"
SECRET_KEY = "replace-with-your-generated-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

Replace replace-with-your-generated-secret-key-here with the hex string you generated earlier. Keep the surrounding quotes.

What do these configuration values do?

SECRET_KEY is the key used to sign every JWT your server creates. Only your server should know this value.
ALGORITHM specifies HS256, a symmetric signing algorithm that uses the same key to sign and verify.
ACCESS_TOKEN_EXPIRE_MINUTES sets how long each token stays valid. After 30 minutes the token expires and the user must log in again.

Next you need Pydantic models to represent the token data your API sends back, plus an OAuth2 scheme that tells FastAPI where to find the token on incoming requests.

In main.py, find the Pydantic models section header that says # Pydantic models. Add these two new models directly below that header, above the existing UserCreate class:

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


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

What do these models do?

Token defines the shape of the JSON your login endpoint returns: an access_token string and a token_type (always "bearer").
TokenData holds the claims extracted from a decoded token. You will use it later when verifying incoming tokens.

Now find the # App setup section. Add this new section directly above it, after the in-memory data stores:

# ---------------------------------------------------------------------------
# OAuth2 scheme
# ---------------------------------------------------------------------------
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

What is OAuth2PasswordBearer?

This tells FastAPI that your API uses the OAuth2 Password flow. The tokenUrl="token" parameter points to the login endpoint you are about to create. FastAPI uses this to automatically add an Authorize button to the Swagger UI.

Save main.py.

Seeing an import error in the terminal?

Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
Check that import jwt appears on its own line (not import PyJWT). The package is called PyJWT on pip but you import it as jwt.

Why is my import failing after adding JWT dependencies?

Create the login endpoint and auth helpers

With the configuration in place, you need three helper functions: one to look up a user, one to authenticate credentials, and one to mint a new JWT. These helpers keep the login endpoint clean and testable.

In main.py, find the # App setup section. Add this new section directly below it (after the app = FastAPI(...) line), above the # Endpoints - Auth section:

# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------
def get_user(username: str) -> UserInDB | None:
    return users_db.get(username)


def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user(username)
    if not user:
        # Still verify against a dummy hash to prevent timing attacks
        verify_password(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

What do these helpers do?

get_user() looks up a user by username in the in-memory users_db dictionary.
authenticate_user() finds the user and verifies the password against the stored hash. If the user does not exist, it still runs a password check against DUMMY_HASH so the response time is the same either way. This prevents attackers from guessing valid usernames based on how fast the server responds.

Directly below the authenticate_user function, add the token creation helper:

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})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

What does create_access_token do?

This function takes a dictionary of claims (like the username) and adds an exp expiration claim. It then signs the whole payload with your SECRET_KEY using jwt.encode(). The resulting string is the JWT that clients will send back on every request.

Now it is time to create the login endpoint that uses these helpers.

Find the # Endpoints - Auth section. Directly below the existing register endpoint function, add the login endpoint:

@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    # Authenticate the user with the provided credentials
    user = authenticate_user(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 signed JWT with the username as the subject claim
    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 the /token endpoint do?

OAuth2PasswordRequestForm automatically parses the username and password from the login form. FastAPI handles the form parsing for you.
If authentication fails, the endpoint returns a 401 Unauthorized response with a WWW-Authenticate: Bearer header. This header tells the client it needs a bearer token.
The "sub" (subject) claim is the standard JWT field for identifying who the token belongs to.

Save main.py.
Refresh http://127.0.0.1:8000/docs in your browser. You should see a new POST /token endpoint in the list and an Authorize button at the top right.

Don't see the Authorize button?

Check your terminal for errors. A missing comma or mismatched parenthesis in the new code will prevent the server from reloading.
Make sure oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") exists above the app = FastAPI(...) line. FastAPI reads the scheme at startup to generate the Authorize UI.

My FastAPI Swagger UI is not showing the Authorize button after adding OAuth2PasswordBearer.

Protect the book endpoints with token verification

Your API can now issue tokens, but it does not check them yet. Every book endpoint is still wide open. You need a dependency that intercepts every request, pulls the token from the Authorization: Bearer header, and decodes it to identify the user.

In main.py, add a new section between the # Auth helpers section and the # Endpoints - Auth section. Copy the following code:

# ---------------------------------------------------------------------------
# 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:
        # Decode the JWT and extract the username 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(token_data.username)
    if user is None:
        raise credentials_exception
    return user

What does get_current_user do?

This is a FastAPI dependency. Any endpoint that includes it as a parameter automatically requires a valid token.

Depends(oauth2_scheme) extracts the bearer token from the request's Authorization header.
jwt.decode() verifies the signature and checks the expiration. If anything is wrong, it raises InvalidTokenError.
If the token is valid, the function returns the full UserInDB object so the endpoint knows who is making the request.

Now you will wire this dependency into every book endpoint. Each endpoint gains a current_user parameter that FastAPI resolves before the endpoint runs. If the token is missing or invalid, FastAPI returns a 401 automatically.

Find the # Endpoints - Books section. Replace all five book endpoint functions with the updated versions below:

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


@app.get("/books/{book_id}", response_model=Book)
async def get_book(
    book_id: int,
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    for book in books_db:
        if book.id == book_id:
            return book
    raise HTTPException(status_code=404, detail="Book not found")


@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(
    book_data: BookCreate,
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    global book_id_counter
    book = Book(id=book_id_counter, **book_data.model_dump())
    book_id_counter += 1
    books_db.append(book)
    return book


@app.put("/books/{book_id}", response_model=Book)
async def update_book(
    book_id: int,
    book_data: BookCreate,
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    for i, book in enumerate(books_db):
        if book.id == book_id:
            updated = Book(id=book_id, **book_data.model_dump())
            books_db[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Book not found")


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

What changed in each endpoint?

Every book function now has a current_user: Annotated[UserInDB, Depends(get_current_user)] parameter. The function body is unchanged.

Annotated is a typing construct that lets you attach metadata to a type hint. Here, it tells FastAPI: "this parameter is a UserInDB, and you should resolve it by calling get_current_user first." If the dependency fails, the endpoint never runs.

Save main.py.

Time to see the authentication in action. Since your server stores data in memory and auto-reloads on save, you need to re-register a user before testing.

Refresh http://127.0.0.1:8000/docs in your browser.
Expand the POST /register endpoint and click Try it out.
Register a test user with the username testuser and password testpass123. Click Execute.
Now expand the GET /books endpoint and click Try it out, then Execute.

You should see a 401 Unauthorized response with the message "Not authenticated". Your endpoints are now locked down.

Click the Authorize button at the top right of the Swagger UI page.
Enter testuser in the username field.
Enter testpass123 in the password field.
Click Authorize, then click Close.
Try the GET /books endpoint again. Click Try it out, then Execute.

You should now see a 200 OK response with an empty list []. The endpoint works when you are authenticated.

Still seeing 401 after authorizing?

Make sure you registered the user after the last server reload. Every time you save main.py, the server restarts and the in-memory data is wiped.
Check that the username and password you entered in the Authorize dialog match the ones you used for registration. Both are case-sensitive.
Verify that your SECRET_KEY is a quoted string in main.py. A missing quote or stray whitespace will cause decode failures.

I'm still getting 401 Unauthorized after clicking Authorize in FastAPI Swagger UI. Help me debug.

Double check that your full main.py matches the code below.

✔️ Awesome, I've got everything!

Make sure you have saved main.py and your server is running without errors in the terminal.

ⓧ 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, Request, Response, 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 = "replace-with-your-generated-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# ---------------------------------------------------------------------------
# Password hashing
# ---------------------------------------------------------------------------
password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")


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)


# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
class Token(BaseModel):
    access_token: str
    token_type: str


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


class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    password: str = Field(min_length=8, max_length=128)
    role: str = "reader"


class User(BaseModel):
    username: str
    role: str


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=2100)
    isbn: str = Field(min_length=10, max_length=13)


class Book(BookCreate):
    id: int


# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
users_db: dict[str, UserInDB] = {}
books_db: list[Book] = []
book_id_counter: int = 1

# ---------------------------------------------------------------------------
# OAuth2 scheme
# ---------------------------------------------------------------------------
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = FastAPI(title="Secure Book Catalog API")


# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------
def get_user(username: str) -> UserInDB | None:
    return users_db.get(username)


def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user(username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    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})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


# ---------------------------------------------------------------------------
# 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(token_data.username)
    if user is None:
        raise credentials_exception
    return user


# ---------------------------------------------------------------------------
# Endpoints - Auth
# ---------------------------------------------------------------------------
@app.post("/register", response_model=User)
async def register(user_data: UserCreate):
    if user_data.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    hashed = get_password_hash(user_data.password)
    user_in_db = UserInDB(
        username=user_data.username,
        role=user_data.role,
        hashed_password=hashed,
    )
    users_db[user_data.username] = user_in_db
    return user_in_db


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    user = authenticate_user(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")


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


@app.get("/books/{book_id}", response_model=Book)
async def get_book(
    book_id: int,
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    for book in books_db:
        if book.id == book_id:
            return book
    raise HTTPException(status_code=404, detail="Book not found")


@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(
    book_data: BookCreate,
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    global book_id_counter
    book = Book(id=book_id_counter, **book_data.model_dump())
    book_id_counter += 1
    books_db.append(book)
    return book


@app.put("/books/{book_id}", response_model=Book)
async def update_book(
    book_id: int,
    book_data: BookCreate,
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    for i, book in enumerate(books_db):
        if book.id == book_id:
            updated = Book(id=book_id, **book_data.model_dump())
            books_db[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Book not found")


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


# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

Your API now requires authentication for every book operation. Next up, you will add role-based authorization so that only admin users can create, update, and delete books.

Add Role-Based Authorization

In the previous step, you locked down your API with JWT authentication. Every endpoint now requires a valid token before it responds. But there is a problem: any authenticated user can do anything.

A reader and an admin have the exact same power right now. A reader could delete every book in the catalog. In this step, you will add role-based authorization so that readers can only browse the catalog, while only admins can create, update, or delete books.

In this step, get ready to:

Create a reusable require_role dependency factory that checks user roles.
Apply role restrictions so only admins can modify books.
Test that a reader gets blocked and an admin gets through.

Create the require_role dependency factory

Authentication answers "who are you?" but authorization answers "what are you allowed to do?" You need a function that checks a user's role before letting them access an endpoint.

A dependency factory is a function that returns another function. You call require_role(["admin"]) and it gives you back a dependency tailored to that specific list of roles.

In main.py, scroll to the Dependencies section and find the get_current_user function.
Add the following require_role function directly below get_current_user, before the # Endpoints - Auth comment:

def require_role(allowed_roles: list[str]):
    async def role_checker(
        current_user: Annotated[UserInDB, Depends(get_current_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

What does this code do?

require_role is a factory function. When you call require_role(["admin"]), it creates and returns a new inner function called role_checker that is locked to the "admin" role.

The inner function uses Depends(get_current_user) to first authenticate the user. Then it checks whether that user's role is in the allowed_roles list. If the role does not match, it raises a 403 Forbidden error.

Save main.py.

You will not see a change yet. The next substep wires this function into the book endpoints.

Seeing a syntax error in the terminal?

Make sure the entire role_checker function is indented inside require_role. It is a nested function, so it needs one extra level of indentation.
Check that return role_checker is at the same indent level as async def role_checker. It belongs to the outer function, not the inner one.
help me fix a syntax error in my require_role function

Apply role restrictions to the book endpoints

Right now, all five book endpoints use Depends(get_current_user). This means any logged-in user can create, update, and delete books. You will keep the two read endpoints open to all authenticated users and restrict the three write endpoints to admins only.

In main.py, scroll down to the book endpoints section.
Find this comment:

# ---------------------------------------------------------------------------
# Endpoints - Books
# ---------------------------------------------------------------------------

Update the comment to clarify that these are the read-only endpoints:

# ---------------------------------------------------------------------------
# Endpoints - Books (read: any authenticated user)
# ---------------------------------------------------------------------------

Next, you will add a new section comment to separate the write endpoints from the read endpoints.

Find the @app.post("/books" decorator line for the create_book function.
Add this section comment on a new blank line above it:

# ---------------------------------------------------------------------------
# Endpoints - Books (write: admin only)
# ---------------------------------------------------------------------------

Now for the real security change. You need to swap the dependency on the three write endpoints from get_current_user to require_role(["admin"]).

Inside the create_book function, find this parameter line:

    current_user: Annotated[UserInDB, Depends(get_current_user)],

Replace Depends(get_current_user) with Depends(require_role(["admin"])), so the line now looks like this:

    current_user: Annotated[UserInDB, Depends(require_role(["admin"]))],

Make the same replacement inside the update_book function.
Make the same replacement inside the delete_book function.
Save main.py.

Your terminal should show that the dev server reloaded automatically. The two GET endpoints still accept any authenticated user. The three write endpoints now require the admin role.

Server showing an error after saving?

Make sure you only changed the dependency on create_book, update_book, and delete_book. The list_books and get_book functions should still use Depends(get_current_user).
Double check that the square brackets and parentheses match exactly: Depends(require_role(["admin"])). It is easy to miss a closing bracket.
help me fix the require_role dependency on my book endpoints

Why a factory function instead of a simple check?

You could hardcode a role check inside each endpoint, but that means repeating the same logic everywhere. A factory function lets you write the check once and customize it per endpoint.

If you later add an editor role that can update but not delete, you just change the decorator to require_role(["admin", "editor"]). No rewriting of the check logic.

Test role-based authorization in Swagger UI

Time to prove that your role restrictions actually work. You will log in as a reader, get blocked from creating a book, then log in as an admin and succeed.

Why do I need to re-register users?

Your API stores users in memory, not in a database. Every time you save main.py, the dev server reloads and the in-memory data resets. Any users you registered before the save are gone.

Open http://127.0.0.1:8000/docs in your browser.
Expand the POST /register endpoint and click Try it out.
Register a reader user by pasting this into the request body and clicking Execute:

{
  "username": "alice",
  "password": "password123",
  "role": "reader"
}

{
  "username": "bob",
  "password": "password456",
  "role": "admin"
}

Both registrations should return a 200 response with the username and role. Now log in as the reader.

Click the green Authorize button at the top of the page (the lock icon).
Enter alice in the username field.
Enter password123 in the password field.
Click Authorize, then click Close.
Expand the POST /books endpoint and click Try it out.
Paste this book into the request body and click Execute:

{
  "title": "Clean Code",
  "author": "Robert Martin",
  "year": 2008,
  "isbn": "9780132350884"
}

You should see a 403 Forbidden response with the message "You do not have permission to perform this action". The reader was authenticated but not authorized.

What is the difference between 401 and 403?

A 401 Unauthorized means "I don't know who you are." The request had no valid token at all.

A 403 Forbidden means "I know who you are, but you are not allowed to do this." The token was valid, but the user's role does not have permission.

Now log in as the admin and try the same request.

Click the Authorize button again.
Click Logout to clear the reader's token.
Enter bob in the username field.
Enter password456 in the password field.
Click Authorize, then click Close.
Go back to POST /books and click Execute with the same book data.

You should see a 201 Created response with the book data including an id field. The admin role was authorized to create the book.

Reader is not getting 403?

Make sure you clicked Logout and re-authorized between switching users. Swagger UI keeps the old token until you explicitly log out.
Verify that you registered alice with "role": "reader" and not "role": "admin". Check the response from the POST /register call to confirm.
Confirm that all three write endpoints (create_book, update_book, delete_book) use Depends(require_role(["admin"])) and not Depends(get_current_user).
help me debug why my reader user is not getting a 403 Forbidden response

What OWASP risk does this prevent?

This addresses OWASP API Security Top 10 risk API5: Broken Function Level Authorization. Without role checks, any authenticated user can call admin-only functions like deleting data.

The require_role dependency enforces that only users with the right role can reach sensitive operations. This is one of the most impactful API security patterns you can implement.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and that the dev server is running without errors.

ⓧ I'd like to double check the full code

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

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

import jwt
from fastapi import Depends, FastAPI, HTTPException, Request, Response, 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 = "replace-with-your-generated-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# ---------------------------------------------------------------------------
# Password hashing
# ---------------------------------------------------------------------------
password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")


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)


# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
class Token(BaseModel):
    access_token: str
    token_type: str


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


class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    password: str = Field(min_length=8, max_length=128)
    role: str = "reader"


class User(BaseModel):
    username: str
    role: str


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=2100)
    isbn: str = Field(min_length=10, max_length=13)


class Book(BookCreate):
    id: int


# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
users_db: dict[str, UserInDB] = {}
books_db: list[Book] = []
book_id_counter: int = 1

# ---------------------------------------------------------------------------
# OAuth2 scheme
# ---------------------------------------------------------------------------
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = FastAPI(title="Secure Book Catalog API")


# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------
def get_user(username: str) -> UserInDB | None:
    return users_db.get(username)


def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user(username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    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})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


# ---------------------------------------------------------------------------
# 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(token_data.username)
    if user is None:
        raise credentials_exception
    return user


def require_role(allowed_roles: list[str]):
    async def role_checker(
        current_user: Annotated[UserInDB, Depends(get_current_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


# ---------------------------------------------------------------------------
# Endpoints - Auth
# ---------------------------------------------------------------------------
@app.post("/register", response_model=User)
async def register(user_data: UserCreate):
    if user_data.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    hashed = get_password_hash(user_data.password)
    user_in_db = UserInDB(
        username=user_data.username,
        role=user_data.role,
        hashed_password=hashed,
    )
    users_db[user_data.username] = user_in_db
    return user_in_db


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    user = authenticate_user(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")


# ---------------------------------------------------------------------------
# Endpoints - Books (read: any authenticated user)
# ---------------------------------------------------------------------------
@app.get("/books", response_model=list[Book])
async def list_books(
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    return books_db


@app.get("/books/{book_id}", response_model=Book)
async def get_book(
    book_id: int,
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    for book in books_db:
        if book.id == book_id:
            return book
    raise HTTPException(status_code=404, detail="Book not found")


# ---------------------------------------------------------------------------
# Endpoints - Books (write: admin only)
# ---------------------------------------------------------------------------
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(
    book_data: BookCreate,
    current_user: Annotated[UserInDB, Depends(require_role(["admin"]))],
):
    global book_id_counter
    book = Book(id=book_id_counter, **book_data.model_dump())
    book_id_counter += 1
    books_db.append(book)
    return book


@app.put("/books/{book_id}", response_model=Book)
async def update_book(
    book_id: int,
    book_data: BookCreate,
    current_user: Annotated[UserInDB, Depends(require_role(["admin"]))],
):
    for i, book in enumerate(books_db):
        if book.id == book_id:
            updated = Book(id=book_id, **book_data.model_dump())
            books_db[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Book not found")


@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=404, detail="Book not found")


# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

Your API now enforces both authentication and authorization. Users must prove who they are with a token, and their role determines what they can do. Next up, you will add rate limiting, CORS, and security headers to harden the API against abuse.

Add Rate Limiting, CORS, and Security Headers

Your API now authenticates users with JWT tokens and enforces role-based access. But there is still a critical gap.

An attacker could hammer your login endpoint thousands of times per second to guess passwords. A malicious frontend hosted on another domain could silently call your API. Your responses lack the security headers that browsers rely on to prevent clickjacking and content sniffing. In this step, you will close all three gaps.

In this step, get ready to:

Add rate limiting to prevent brute-force login attacks.
Configure CORS to control which origins can access your API.
Add security headers to every response.

Add rate limiting with SlowAPI

Rate limiting caps how many times a client can hit an endpoint within a time window. Without it, an attacker can fire thousands of login attempts per second to brute-force passwords.

What is SlowAPI?

SlowAPI is the most popular rate limiting library for FastAPI. It uses a simple decorator to set limits like "5 requests per minute" on individual endpoints.

In main.py, find the line from pwdlib import PasswordHash near the top of the file.
Add these three import lines directly below it:

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

Scroll down to the # OAuth2 scheme comment section.
Add this new rate limiter section directly above it:

# ---------------------------------------------------------------------------
# Rate limiter
# ---------------------------------------------------------------------------
limiter = Limiter(key_func=get_remote_address)

Find the line app = FastAPI(title="Secure Book Catalog API") in the # App setup section.
Add these two lines directly below it:

app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

What does this code do?

Limiter(key_func=get_remote_address) creates a rate limiter that identifies clients by their IP address.

app.state.limiter stores the limiter on the app so SlowAPI can access it during requests. The add_exception_handler line tells FastAPI to return a proper 429 Too Many Requests response when a client exceeds the limit.

You won't see a change yet. Next, you will apply the rate limit to the login endpoint.

The login endpoint is the most important place to rate limit. It is the one attackers target for brute-force password guessing. You need to modify the existing /token endpoint to add the rate limiting decorator and a required parameter.

Scroll down to the /token endpoint.
Find these lines:

@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:

Replace those lines with the following, which adds the @limiter.limit decorator and a new request: Request parameter:

@app.post("/token", response_model=Token)
@limiter.limit("5/minute")
async def login_for_access_token(
    request: Request,
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:

Why does SlowAPI need request: Request?

SlowAPI reads the client's IP address from the Request object to track how many times that IP has called the endpoint. If request: Request is missing from the function signature, SlowAPI silently fails and the rate limit does nothing.

The Request import already exists at the top of your file (from the fastapi import line), so you just need to add it as a parameter here.

Save main.py.
Check your terminal. The FastAPI dev server should auto-reload without errors.

Seeing an import error or crash?

Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
Confirm slowapi is installed. Run pip install slowapi again if needed.
Check that you imported _rate_limit_exceeded_handler with the leading underscore. It is easy to miss.

Still stuck? Ask for help.

Configure CORS middleware

CORS (Cross-Origin Resource Sharing) controls which websites are allowed to call your API. Without CORS configuration, a browser blocks any frontend hosted on a different domain from making requests to your backend.

At the top of main.py, find the line from fastapi import Depends, FastAPI, HTTPException, Request, Response, status.
Add this import line directly below it:

from fastapi.middleware.cors import CORSMiddleware

Scroll down to the # App setup section.
Add the CORS middleware block below the app.add_exception_handler line you just added:

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

What does this code do?

allow_origins is the list of domains allowed to call your API. Here it is set to http://localhost:3000, which is where a local React or Next.js frontend typically runs.

allow_credentials=True lets the browser send cookies and authorization headers with cross-origin requests. allow_methods and allow_headers are set to "*" to allow all HTTP methods and headers.

Why not use allow_origins=["*"]?

Setting allow_origins to a wildcard "*" means any website in the world can call your API. Combined with allow_credentials=True, this is a security vulnerability because a malicious site could make authenticated requests on behalf of your users.

Always use an explicit allowlist of trusted origins in production.

Save main.py and confirm the server reloads without errors in your terminal.

You won't see a visible change in Swagger UI since CORS only affects cross-origin browser requests. The next chunk adds the final security layer.

Add security headers and test rate limiting

Security headers tell the browser how to handle your API's responses. They prevent attacks like clickjacking and content-type sniffing without any changes to your endpoint logic.

At the top of main.py, find the line from pwdlib import PasswordHash.
You already added the SlowAPI imports below that line. Add this new import line below the SlowAPI imports:

from starlette.middleware.base import BaseHTTPMiddleware

Scroll down to the # App setup section.
Add the security headers middleware class and registration below the CORS middleware block you just added:

# Security headers middleware
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        response.headers["Strict-Transport-Security"] = (
            "max-age=63072000; includeSubDomains; preload"
        )
        return response


app.add_middleware(SecurityHeadersMiddleware)

What do these headers do?

X-Content-Type-Options: nosniff prevents browsers from guessing a file's content type, which stops attacks that disguise malicious scripts as harmless files.
X-Frame-Options: DENY blocks your API from being embedded in an iframe, preventing clickjacking attacks.
Referrer-Policy: strict-origin-when-cross-origin limits how much URL information is shared when navigating to other sites.
Strict-Transport-Security tells browsers to only connect over HTTPS for the next 2 years.

Save main.py and confirm the server reloads without errors.

Server crashing after adding the middleware class?

Verify you imported BaseHTTPMiddleware from starlette.middleware.base, not from fastapi.
Make sure the dispatch method is indented inside the SecurityHeadersMiddleware class.
Check that app.add_middleware(SecurityHeadersMiddleware) is outside the class, at the module level.

Need help debugging?

Now let's verify that rate limiting actually works. You need to register a user first (since the in-memory database resets every time the server reloads), then hit the login endpoint repeatedly.

Open http://127.0.0.1:8000/docs in your browser.
Expand the POST /register endpoint and click Try it out.
Register a test user with username testuser, password testpassword123, and click Execute.
Expand the POST /token endpoint and click Try it out.
Enter testuser in the username field and testpassword123 in the password field.
Click Execute six times in quick succession.

The first five requests return a 200 status with a JWT token. The sixth request returns a 429 Too Many Requests response. Your rate limiter is working.

Not seeing a 429 response?

Double-check that the @limiter.limit("5/minute") decorator is placed between @app.post("/token", response_model=Token) and the async def line.
Confirm your login_for_access_token function has request: Request as its first parameter. Without it, SlowAPI cannot read the client's IP and silently skips the limit.
Make sure you clicked Execute at least 6 times within the same minute.

Still not working? Ask for help.

What OWASP risks did you just address?

OWASP is the Open Worldwide Application Security Project. Its API Security Top 10 is the industry standard list of the most critical API vulnerabilities.

Rate limiting addresses API4: Unrestricted Resource Consumption. CORS with an explicit allowlist prevents unauthorized cross-origin access. Security headers protect against client-side attacks like clickjacking and content sniffing (API8: Security Misconfiguration).

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Here is the complete main.py file at the end of this step. Compare it line by line with yours.

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

import jwt
from fastapi import Depends, FastAPI, HTTPException, Request, Response, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pydantic import BaseModel, Field
from pwdlib import PasswordHash
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
from starlette.middleware.base import BaseHTTPMiddleware

# ---------------------------------------------------------------------------
# Security configuration
# ---------------------------------------------------------------------------
# Generate your own key: python -c "import secrets; print(secrets.token_hex(32))"
SECRET_KEY = "replace-with-your-generated-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# ---------------------------------------------------------------------------
# Password hashing
# ---------------------------------------------------------------------------
password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")


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)


# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
class Token(BaseModel):
    access_token: str
    token_type: str


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


class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    password: str = Field(min_length=8, max_length=128)
    role: str = "reader"


class User(BaseModel):
    username: str
    role: str


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=2100)
    isbn: str = Field(min_length=10, max_length=13)


class Book(BookCreate):
    id: int


# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
users_db: dict[str, UserInDB] = {}
books_db: list[Book] = []
book_id_counter: int = 1

# ---------------------------------------------------------------------------
# Rate limiter
# ---------------------------------------------------------------------------
limiter = Limiter(key_func=get_remote_address)

# ---------------------------------------------------------------------------
# OAuth2 scheme
# ---------------------------------------------------------------------------
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = FastAPI(title="Secure Book Catalog API")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# Security headers middleware
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        response.headers["Strict-Transport-Security"] = (
            "max-age=63072000; includeSubDomains; preload"
        )
        return response


app.add_middleware(SecurityHeadersMiddleware)


# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------
def get_user(username: str) -> UserInDB | None:
    return users_db.get(username)


def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user(username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    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})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


# ---------------------------------------------------------------------------
# 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(token_data.username)
    if user is None:
        raise credentials_exception
    return user


def require_role(allowed_roles: list[str]):
    async def role_checker(
        current_user: Annotated[UserInDB, Depends(get_current_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


# ---------------------------------------------------------------------------
# Endpoints - Auth
# ---------------------------------------------------------------------------
@app.post("/register", response_model=User)
async def register(user_data: UserCreate):
    if user_data.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )
    hashed = get_password_hash(user_data.password)
    user_in_db = UserInDB(
        username=user_data.username,
        role=user_data.role,
        hashed_password=hashed,
    )
    users_db[user_data.username] = user_in_db
    return user_in_db


@app.post("/token", response_model=Token)
@limiter.limit("5/minute")
async def login_for_access_token(
    request: Request,
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    user = authenticate_user(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")


# ---------------------------------------------------------------------------
# Endpoints - Books (read: any authenticated user)
# ---------------------------------------------------------------------------
@app.get("/books", response_model=list[Book])
async def list_books(
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    return books_db


@app.get("/books/{book_id}", response_model=Book)
async def get_book(
    book_id: int,
    current_user: Annotated[UserInDB, Depends(get_current_user)],
):
    for book in books_db:
        if book.id == book_id:
            return book
    raise HTTPException(status_code=404, detail="Book not found")


# ---------------------------------------------------------------------------
# Endpoints - Books (write: admin only)
# ---------------------------------------------------------------------------
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
async def create_book(
    book_data: BookCreate,
    current_user: Annotated[UserInDB, Depends(require_role(["admin"]))],
):
    global book_id_counter
    book = Book(id=book_id_counter, **book_data.model_dump())
    book_id_counter += 1
    books_db.append(book)
    return book


@app.put("/books/{book_id}", response_model=Book)
async def update_book(
    book_id: int,
    book_data: BookCreate,
    current_user: Annotated[UserInDB, Depends(require_role(["admin"]))],
):
    for i, book in enumerate(books_db):
        if book.id == book_id:
            updated = Book(id=book_id, **book_data.model_dump())
            books_db[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Book not found")


@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=404, detail="Book not found")


# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

Secret mission

Add Token Refresh

Your API's access tokens last 30 minutes right now. In production, that's a long window for a stolen token to cause damage. Can you add a token refresh mechanism with short-lived access tokens and longer-lived refresh tokens?

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:

secure-api project folder on your Desktop.
Python virtual environment (venv) inside the project folder.
main.py application file.
Installed pip packages: FastAPI, PyJWT, pwdlib, and SlowAPI.
FastAPI dev server (running in your terminal).

Keep everything running

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

Your FastAPI dev server stays running in the terminal. You can keep testing endpoints at http://127.0.0.1:8000/docs.
All in-memory data (users, books) persists as long as the server is running.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your project files and virtual environment stay intact for when you come back.

Go to the terminal where the FastAPI server is running.
Press Ctrl+C to stop the server.
When you come back, open a terminal in the secure-api folder, reactivate the virtual environment, and run fastapi dev main.py again.
In-memory data (registered users and books) resets every time you restart the server. You will need to re-register users and re-create books.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes everything you built in this project.

Go to the terminal where the FastAPI server is running and press Ctrl+C to stop it.
Deactivate the virtual environment by running deactivate in your terminal.
Delete the entire secure-api folder from your Desktop. This removes the virtual environment, installed packages, and all project files.

Mission accomplished

That's a Wrap!

You just built a production-grade secured REST API from scratch using FastAPI. Your API handles authentication, authorization, input validation, and abuse prevention, all testable through interactive Swagger docs.

You've learned how to:

Build a JWT-authenticated REST API with Argon2 password hashing, signed tokens, and the OAuth2 Password flow.
Create role-based authorization that restricts write operations to admin users while allowing readers to browse the catalog.
Add production security layers including rate limiting with SlowAPI, CORS configuration with an explicit origin allowlist, and custom security headers middleware.
Secret Mission: Add a token refresh mechanism with short-lived access tokens and long-lived refresh tokens.

Ready to quiz yourself?