Secure a Book Collection API

Build a FastAPI REST API secured with JWT auth and password hashing.

Introduction

30 Second Summary

Every time you log into an app, a security system works behind the scenes to verify who you are. Without it, anyone could access your account or tamper with your data.

In this project, you will build a Book Collection REST API from scratch and secure it with password hashing, token-based authentication, and input validation.

What You'll Build

A Python REST API built with FastAPI, secured with JWT authentication, Argon2 password hashing, and Pydantic input validation.

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

  • A working API with full CRUD endpoints for creating, reading, and deleting books.
  • A JWT authentication system with securely hashed passwords protecting your API.
  • Validated Pydantic models that automatically reject malformed data before it reaches your endpoints.
  • A rate limiting layer that blocks API abuse (Secret Mission).

Are there any prerequisites?

You need basic Python knowledge like writing functions, working with dictionaries, and using imports. No prior web development or API experience is required.

Set Up Your Development Environment

Building a secure API requires a few specific tools working together. Python runs your server code. FastAPI provides the web framework. Supporting libraries handle password hashing, token generation, and input validation.

In this step, you will get all of these tools installed and connected. By the end, you will have a running server with live documentation you can explore right in your browser.

In this step, get ready to:
  • Install Python and create the project folder.
  • Set up a virtual environment and install all required packages.
  • Launch a FastAPI server and explore the Swagger UI.
Install Python and create the project folder
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Check that Python is installed by running this command:
python --version

I see a version number

You should see something like Python 3.12.x. Confirm the version is 3.10 or higher. If it is, you are ready to move on.

Command not found

On macOS, the command might be python3 instead. Try running:

python3 --version

If that shows a version number of 3.10 or higher, use python3 in place of python for the rest of this project.

If neither command works, download and install Python from the official website.

  • Go to python.org/downloads.
  • Click the download button for the latest version (3.10 or higher).
  • Run the installer. On Windows, check the box that says Add Python to PATH before clicking Install Now.
  • Close and reopen your terminal, then run python --version again to confirm it works.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir secure-books-api
cd secure-books-api

Your terminal prompt should now show secure-books-api, confirming you are inside the project folder.

Create a virtual environment and install packages

A virtual environment is an isolated space where your project's packages live. It keeps this project's dependencies separate from other Python projects on your machine.

  • Create a virtual environment by running this command:
python -m venv venv

What does this command do?

python -m venv tells Python to run its built-in virtual environment module. The second venv is the name of the folder it creates. This folder holds a local copy of Python and pip just for your project.

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

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

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

Don't see (venv) in your prompt?

  • Make sure you ran the activation command for the correct operating system. macOS uses source while Windows uses a direct path to the activate script.
  • On Windows, if you see a permissions error in PowerShell, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned first, then try the activation command again.
  • Confirm you are inside the secure-books-api folder where you created the virtual environment.

Help me activate my Python virtual environment.

  • Install all the required packages by running this command:
pip install "fastapi[standard]" "uvicorn[standard]" pyjwt "pwdlib[argon2]" python-multipart

What does each package do?

  • fastapi[standard] is the web framework for building your API. The [standard] extra bundles recommended tools including the development server.
  • uvicorn[standard] is the server that runs your FastAPI app. Uvicorn listens for incoming requests and sends them to your code.
  • pyjwt creates and verifies JSON Web Tokens (JWTs). You will use these tokens to authenticate users.
  • pwdlib[argon2] handles password hashing with the Argon2 algorithm. It turns plaintext passwords into unreadable hashes.
  • python-multipart is required for parsing form data. Your login endpoint will use form fields to accept usernames and passwords.

The install may take a minute. When it finishes, you should see Successfully installed followed by a list of package names at the end of the output.

Installation failed?

  • Check that (venv) still appears in your prompt. If it disappeared, re-run the activation command before trying the install again.
  • If you see a permissions error, make sure you are installing inside the virtual environment, not globally.
  • If the pwdlib[argon2] package fails to build, make sure your Python version is 3.10 or higher.

Help me fix a pip install error in my virtual environment.

Create a minimal FastAPI app

With all your packages installed, it is time to write your first line of API code. You will create a minimal app with a single endpoint to confirm everything works.

  • Create a new file called main.py in your secure-books-api folder using your code editor, then copy in the following code:
from fastapi import FastAPI

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

# --- Root endpoint ---
@app.get("/")
def root():
    return {"message": "Welcome to the Secure Book Collection API"}

What does this code do?

  • FastAPI(title="Secure Book Collection API") creates your application instance. The title shows up in the auto-generated documentation page.
  • @app.get("/") registers the function below it as a handler for GET requests to the root URL (/).
  • The root() function returns a Python dictionary. FastAPI automatically converts it to a JSON response.
  • Save main.py.
  • Start the development server by running this command:
uvicorn main:app --reload

What should I see?

Your terminal should display several lines of output, ending with something like:

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

The --reload flag tells Uvicorn to watch your files. Whenever you save a change, the server restarts automatically.

Server not starting?

  • If you see ModuleNotFoundError, check that (venv) is in your prompt. You may need to re-activate your virtual environment.
  • If you see a SyntaxError, double check that your main.py code matches the code block above exactly.
  • If port 8000 is already in use, close any other servers running on that port and try again.

Help me get my Uvicorn server running.

Awesome, I've got everything!

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

I'd like to double check the full code

from fastapi import FastAPI

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

# --- Root endpoint ---
@app.get("/")
def root():
    return {"message": "Welcome to the Secure Book Collection API"}
  • Open your browser and go to http://127.0.0.1:8000/docs.

You should see a page titled Secure Book Collection API with your GET / endpoint listed below the title.

What is Swagger UI?

This page is called Swagger UI. FastAPI generates it automatically from your code. You can click on any endpoint to expand it, fill in parameters, and send real requests to your API. No separate tool needed.

You will use this page throughout the project to test every endpoint you build.

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

You should see a response body containing {"message": "Welcome to the Secure Book Collection API"}. Your API is alive and responding.

Your development environment is ready and your first endpoint is live. Next up, you will build out the full Book Collection API with endpoints to create, list, fetch, and delete books.

Build the Book Collection API

In the last step, you set up your development environment and confirmed that FastAPI is running. Now it's time to build the actual API that manages your book collection.

You'll define data models that describe what a book looks like, then create endpoints to add, list, fetch, and delete books. By the end of this step, you'll have a fully working REST API you can interact with through your browser.

In this step, get ready to:
  • Define Pydantic data models for your book collection.
  • Create four API endpoints to add, list, fetch, and delete books.
  • Test all endpoints using the interactive Swagger UI.
Define your data models

Before building endpoints, you need to define what a "book" looks like in your API. Pydantic models let you describe the structure of your data as Python classes.

You'll create two models: one for incoming requests (without an ID) and one for responses (with an ID assigned by the server).

  • Open main.py and replace its entire contents with the following code:
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

# --- App Setup ---
app = FastAPI(title="Secure Book Collection API")

# --- In-Memory Storage ---
books_db = {}  # Stores books as {id: Book}
book_id_counter = 0  # Tracks the next available book ID


# --- Pydantic Models ---

# The shape of data a client sends when creating a book (no id needed)
class BookCreate(BaseModel):
    title: str
    author: str
    year: int


# The shape of data the API returns (includes the server-assigned id)
class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int

What does this code do?

  • BookCreate is the shape of data a client sends when adding a book. It has title, author, and year but no id, because the server assigns the ID automatically.
  • Book includes the id field. This is the shape the API sends back in responses.
  • books_db is a dictionary that stores books in memory. book_id_counter tracks the next available ID so each book gets a unique number.

Why two models instead of one?

Separating input and output models is a common API pattern. The client should never have to guess what ID to use. The server controls that. Having a dedicated BookCreate model makes it clear exactly what the client needs to provide.

  • Save main.py.
  • Check your terminal where uvicorn is running. You should see a log line that says WARNING: WatchFiles detected changes in 'main.py'. Reloading... followed by Application startup complete with no errors.

Seeing an import error?

  • Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
  • Check that you ran pip install "fastapi[standard]" in the previous step. If not, run it now.

Why is my FastAPI import failing?

Create endpoints to manage books

Your models define the shape of data. Now you need endpoints that let clients interact with that data over HTTP. Each endpoint maps to a specific action: creating, listing, fetching, or deleting books.

  • In main.py, add the following code below your model definitions to create the first two endpoints:
# --- Book Endpoints ---

# Create a new book and store it in memory
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
def create_book(book: BookCreate):
    global book_id_counter
    book_id_counter += 1
    # Convert BookCreate to Book by adding the generated id
    new_book = Book(id=book_id_counter, **book.model_dump())
    books_db[book_id_counter] = new_book
    return new_book


# Return all books in the collection
@app.get("/books", response_model=list[Book])
def list_books():
    return list(books_db.values())

What does this code do?

  • create_book accepts a BookCreate body, assigns it a unique ID, stores it in books_db, and returns the full Book.
  • status.HTTP_201_CREATED tells the client that a new resource was successfully created, instead of just a generic 200 OK.
  • model_dump() converts the Pydantic model into a dictionary so its fields can be unpacked into the Book constructor with **.
  • list_books returns every book stored in memory as a list.
  • Save main.py.
  • Refresh http://127.0.0.1:8000/docs in your browser. You should see POST /books and GET /books listed in the Swagger UI.

Don't see the new endpoints?

  • Make sure the endpoint code is placed below your model definitions, not inside one of the classes.
  • Check that the @app.post and @app.get decorators are at the top level of the file, not indented under anything.

My endpoints aren't showing up in the Swagger UI docs page.

  • Add the following two endpoints below list_books to fetch and delete individual books:
# Fetch a single book by its ID
@app.get("/books/{book_id}", response_model=Book)
def get_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    return books_db[book_id]


# Remove a book from the collection
@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    del books_db[book_id]

What does this code do?

  • get_book looks up a book by its ID. If the ID doesn't exist in books_db, it raises a 404 Not Found error.
  • delete_book removes a book from the dictionary. It also returns a 404 if the book doesn't exist.
  • {book_id} in the URL path becomes a function parameter. FastAPI automatically converts it to an integer for you.
  • Save main.py.
  • Refresh /docs in your browser. You should now see all four endpoints: POST /books, GET /books, GET /books/{book_id}, and DELETE /books/{book_id}.

Seeing a syntax error in the terminal?

  • Check that the HTTPException import is in your first line: from fastapi import FastAPI, HTTPException, status.
  • Make sure each raise HTTPException(...) block is indented inside its if statement. Python is sensitive to indentation.

Help me fix a syntax error in my FastAPI book endpoints.

Awesome, I've got everything!

Double check you've saved main.py and that the server reloaded without errors.

I'd like to double check the full code

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

# --- App Setup ---
app = FastAPI(title="Secure Book Collection API")

# --- In-Memory Storage ---
books_db = {}  # Stores books as {id: Book}
book_id_counter = 0  # Tracks the next available book ID


# --- Pydantic Models ---

# The shape of data a client sends when creating a book (no id needed)
class BookCreate(BaseModel):
    title: str
    author: str
    year: int


# The shape of data the API returns (includes the server-assigned id)
class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int


# --- Book Endpoints ---

# Create a new book and store it in memory
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
def create_book(book: BookCreate):
    global book_id_counter
    book_id_counter += 1
    # Convert BookCreate to Book by adding the generated id
    new_book = Book(id=book_id_counter, **book.model_dump())
    books_db[book_id_counter] = new_book
    return new_book


# Return all books in the collection
@app.get("/books", response_model=list[Book])
def list_books():
    return list(books_db.values())


# Fetch a single book by its ID
@app.get("/books/{book_id}", response_model=Book)
def get_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    return books_db[book_id]


# Remove a book from the collection
@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    del books_db[book_id]
Test your endpoints in Swagger UI

Your API has four endpoints, but you haven't tried them yet. Swagger UI lets you send real requests to your API directly from your browser, no extra tools needed.

  • 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 Great Gatsby",
  "author": "F. Scott Fitzgerald",
  "year": 1925
}
  • Click Execute.

You should see a 201 response with the book data and a server-assigned id of 1.

  • Create a second book by clicking Try it out again on POST /books and using this request body:
{
  "title": "1984",
  "author": "George Orwell",
  "year": 1949
}
  • Click Execute. This book should get an id of 2.

Now test the read endpoints.

  • Expand GET /books, click Try it out, then click Execute.

You should see both books returned in a list.

  • Expand GET /books/{book_id}, click Try it out.
  • Enter 1 in the book_id field and click Execute.

You should see only "The Great Gatsby" returned.

Now test the delete endpoint and confirm the 404 error handling works.

  • Expand DELETE /books/{book_id}, click Try it out.
  • Enter 1 in the book_id field and click Execute.

You should see a 204 response with no body. The book has been deleted.

  • Go back to GET /books/{book_id}, enter 1 again, and click Execute.

This time you should see a 404 error with the message Book not found. Your error handling is working correctly.

What just happened?

You just tested all four CRUD operations: Create (POST), Read (GET), and Delete (DELETE). The 404 error for a deleted book proves that your API correctly validates whether a resource exists before returning it.

Your book collection API is up and running. Right now, anyone can add or delete books without any restrictions. Next, you'll add user registration with secure password hashing to control who can modify the collection.

Add User Registration with Password Hashing

Your book collection API is up and running with four endpoints that create, list, fetch, and delete books. But there is a serious problem.

Right now, anyone who finds your API can delete every book in the collection. You need a way to identify who is making requests so you can control who gets to make changes. That starts with user registration and secure password storage.

In this step, get ready to:
  • Create Pydantic models to represent users.
  • Set up Argon2 password hashing with helper functions.
  • Build a registration endpoint that securely stores new users.
Create the user models

You need two separate models for users. One accepts the username and password during registration. The other represents the data you send back in API responses, deliberately excluding the password so it never leaks.

  • In main.py, find the line book_id_counter = 0 in your storage section.
  • Add a new users dictionary on the line right above it by copying this code:
users_db = {}
  • Below your Book model class, add these two new user model classes:
# Represents incoming registration data (username + password)
class UserCreate(BaseModel):
    username: str
    password: str


# Represents outgoing user data (password is never included)
class UserOut(BaseModel):
    username: str

What do these models do?

  • UserCreate is the shape of data coming IN. When someone registers, they send a username and a password.
  • UserOut is the shape of data going OUT. It only contains the username, so passwords and hashes are never accidentally exposed in API responses.
  • users_db is an in-memory dictionary that stores registered users. It maps each username to their hashed password.
  • Save main.py.

You will not see a visible change yet. The next substep adds the password hashing functions that these models depend on.

Set up password hashing

Storing passwords as plain text is one of the most dangerous mistakes in web development. If anyone gains access to your storage, every user's password is immediately exposed.

Password hashing solves this by running the password through a one-way mathematical function. The result is an unreadable string that cannot be reversed back into the original password.

  • At the top of main.py, add this new import alongside your existing imports:
from pwdlib import PasswordHash
  • Below the app = FastAPI(...) line, add the password hashing setup:
# --- Password Hashing ---
password_hash = PasswordHash.recommended()
  • Below your model classes, add these two helper functions:
# --- Helper Functions ---
def get_password_hash(password: str) -> str:
    # Turn a plain-text password into an Argon2 hash
    return password_hash.hash(password)


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

What does this code do?

  • PasswordHash.recommended() configures pwdlib to use the Argon2 algorithm. Argon2 won the Password Hashing Competition and is more resistant to GPU-based attacks than older algorithms like bcrypt.
  • get_password_hash() takes a plain-text password and returns a long, unreadable hash string. You call this when a user registers.
  • verify_password() takes a plain-text password and a stored hash, then returns True if they match. You call this when a user logs in.

Why Argon2 instead of bcrypt?

Bcrypt has been the go-to password hashing algorithm for years, but Argon2 is purpose-built to resist modern attacks. It uses large amounts of memory during hashing, which makes it extremely expensive to crack with GPUs or specialized hardware. It is also the algorithm recommended by FastAPI's official documentation.

  • Save main.py.
  • Check your terminal where uvicorn is running. You should see a line like Application startup complete confirming the server reloaded without errors.

Seeing an import error?

  • Make sure you installed pwdlib with the Argon2 extra. The correct install command is pip install "pwdlib[argon2]" (not just pip install pwdlib).
  • Confirm your virtual environment is still active. You should see (venv) at the beginning of your terminal prompt.
  • Check that the import line reads from pwdlib import PasswordHash exactly. The class name is PasswordHash with a capital P and H.

Still stuck? Ask for help.

Create the registration endpoint

With the models and hashing helpers in place, you can now build the endpoint that ties them together. When a user registers, the API hashes their password before storing it and returns only the username.

  • In main.py, add the registration endpoint below your helper functions and above your book endpoints:
# --- Auth Endpoints ---
@app.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def register_user(user: UserCreate):
    # Reject duplicate usernames
    if user.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already registered",
        )
    # Hash the password and store it
    users_db[user.username] = get_password_hash(user.password)
    # Return only the username (never the password or hash)
    return UserOut(username=user.username)

What does this endpoint do?

  • response_model=UserOut tells FastAPI to use the UserOut model for the response. Even if you accidentally include the password in the return value, FastAPI strips it out.
  • The duplicate check prevents two users from registering with the same username. It returns a 400 Bad Request if the username is already taken.
  • get_password_hash(user.password) hashes the password before storing it. The dictionary stores the username as the key and the hash as the value.
  • Save main.py.
  • Open http://127.0.0.1:8000/docs in your browser to see the Swagger UI. You should see a new POST /register endpoint listed alongside your book endpoints.

Now test the registration flow to see password hashing in action.

  • Click the POST /register row to expand it.
  • Click Try it out.
  • Replace the request body with the following:
{
  "username": "alice",
  "password": "secretpass123"
}
  • Click Execute.

The response body should show a 201 Created status with only the username returned. No password or hash appears in the response.

What should I see?

The response body should look like this:

{
  "username": "alice"
}

Even though you sent a password, the API only returns the username. Behind the scenes, your password was hashed into a long string like $argon2id$v=19$m=65536... and stored in users_db. That hash cannot be reversed back into the original password.

  • Try registering the same username alice again by clicking Execute a second time.

This time you should see a 400 Bad Request response with the message Username already registered. Your duplicate check is working.

Not seeing the /register endpoint?

  • Make sure you saved main.py. Unsaved changes will not trigger uvicorn's auto-reload.
  • Hard-refresh the Swagger UI page with Cmd+Shift+R (macOS) or Ctrl+Shift+R (Windows). A normal refresh sometimes caches the old endpoint list.
  • Check your terminal for any red error text. A common cause is placing the endpoint inside another function by mistake. Make sure @app.post("/register"...) is at the top level of your file, not indented inside another function.

Still stuck? Ask for help.

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 full main.py as it should look at the end of this step.

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

# --- App Setup ---
app = FastAPI(title="Secure Book Collection API")

# --- Password Hashing ---
password_hash = PasswordHash.recommended()

# --- In-Memory Storage ---
users_db = {}
books_db = {}
book_id_counter = 0


# --- Pydantic Models ---
class BookCreate(BaseModel):
    title: str
    author: str
    year: int


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


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


class UserOut(BaseModel):
    username: str


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


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


# --- Auth Endpoints ---
@app.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def register_user(user: UserCreate):
    if user.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already registered",
        )
    users_db[user.username] = get_password_hash(user.password)
    return UserOut(username=user.username)


# --- Book Endpoints ---
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
def create_book(book: BookCreate):
    global book_id_counter
    book_id_counter += 1
    new_book = Book(id=book_id_counter, **book.model_dump())
    books_db[book_id_counter] = new_book
    return new_book


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


@app.get("/books/{book_id}", response_model=Book)
def get_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    return books_db[book_id]


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    del books_db[book_id]

Your API can now register users and store their passwords as secure, irreversible hashes. Next up, you will build a login system that issues JWT tokens so authenticated users can prove who they are on every request.

Implement JWT Authentication

Your users can register with securely hashed passwords. Nobody can read those passwords, even if they access your storage directly.

The next challenge is proving who someone is when they come back to use the API. You need a way for users to log in once and then carry proof of their identity with every request. That proof is a JSON Web Token (JWT).

In this step, get ready to:
  • Configure JWT signing with a secret key and a token creation function.
  • Build a login endpoint that verifies credentials and issues access tokens.
  • Create a dependency function that validates tokens and identifies the current user.
Set up your JWT configuration

A JWT is a compact, signed token that encodes a user's identity and an expiration time. Your server creates one when a user logs in, and the user sends it back with every future request to prove who they are.

What is a JSON Web Token?

A JWT has three parts separated by dots: a header (algorithm info), a payload (user data and expiration), and a signature (proof the token has not been tampered with). Your server signs the token with a secret key. When a request comes in with a token, the server verifies the signature to confirm the token is legitimate.

To sign tokens, you need a secret key that only your server knows. Anyone with this key can forge valid tokens, so it must be random and long.

  • Generate a secret key by running this command in your terminal:
python -c "import secrets; print(secrets.token_hex(32))"
  • Copy the long hex string that appears in your terminal output. You will paste it into your code in a moment.

What does this command do?

The secrets module generates cryptographically secure random values. token_hex(32) produces a 64-character hex string (32 random bytes). This is strong enough to prevent brute-force guessing.

Now you need to add new imports and configuration to main.py. The JWT system depends on several libraries you installed earlier.

  • In main.py, replace the import lines at the very top of your file with these updated imports:
from datetime import datetime, timedelta, timezone

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

What's new in these imports?

  • datetime, timedelta, and timezone handle token expiration timestamps.
  • jwt is the PyJWT library for encoding and decoding JSON Web Tokens.
  • Depends enables FastAPI's dependency injection system, which you will use to protect routes.
  • OAuth2PasswordBearer and OAuth2PasswordRequestForm are FastAPI's built-in OAuth2 helpers for handling login forms and extracting tokens from headers.
  • InvalidTokenError is the base exception PyJWT raises when a token is expired, tampered with, or malformed.
  • Next, find the # --- App Setup --- section. Add these JWT configuration lines directly below it:
# --- Security Configuration ---
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
  • Replace your-secret-key-here with the hex string you copied from the terminal earlier. Keep the quotes around it.
  • Now add the OAuth2 scheme setup below your # --- Password Hashing --- section. Copy the following:
# --- OAuth2 Scheme ---
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

What does this configuration do?

  • SECRET_KEY is the private key your server uses to sign and verify every JWT. Keep it secret.
  • ALGORITHM is set to HS256, which uses HMAC with SHA-256 for signing. This is the standard symmetric algorithm for JWTs.
  • ACCESS_TOKEN_EXPIRE_MINUTES controls how long a token stays valid. After 30 minutes, the user must log in again.
  • oauth2_scheme tells FastAPI to look for a Bearer token in the Authorization header. The tokenUrl="token" tells Swagger UI which endpoint handles login.
  • Save main.py.
  • Check your terminal where uvicorn is running. You should see a reload message with no errors.

Seeing an import error?

  • If you see ModuleNotFoundError: No module named 'jwt', make sure your virtual environment is active and run pip install pyjwt again.
  • Double check that every import line is spelled exactly as shown. A missing letter in InvalidTokenError or OAuth2PasswordBearer causes an ImportError.

Still stuck? Ask for help.

Now you need a Token model to define the shape of login responses and a function that actually creates signed tokens.

  • Add the Token model below your UserOut class in the Pydantic models section:
class Token(BaseModel):
    access_token: str
    token_type: str
  • Add the create_access_token function below your verify_password function in the helper functions section:
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    # Copy the data so we don't modify the original
    to_encode = data.copy()
    # Set expiration time
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    # Sign and return the token
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

What does this code do?

  • The Token model defines the JSON shape your login endpoint returns: an access_token string and a token_type (always "bearer").
  • create_access_token takes a dictionary (containing the username) and stamps it with an expiration time.
  • jwt.encode() signs the payload with your SECRET_KEY using the HS256 algorithm and returns a compact token string.
  • The exp claim is a built-in JWT standard. PyJWT automatically rejects tokens whose exp timestamp is in the past.
  • Save main.py.
  • Check your terminal for a clean reload with no errors.

Seeing a syntax error on the | None type hint?

The timedelta | None syntax requires Python 3.10 or later. Run python --version to check. If you are on an older version, update Python before continuing.

Need help upgrading?

Create the login endpoint

You have a function that creates tokens. Now you need an endpoint where users send their username and password to receive one.

  • Add the login endpoint below your register_user endpoint in the auth endpoints section. Copy the following code:
@app.post("/token", response_model=Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    # Extract credentials from the form
    username = form_data.username
    password = form_data.password
    # Reject if user doesn't exist or password is wrong
    if username not in users_db or not verify_password(password, users_db[username]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # Create and return the access token
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")

What does this code do?

  • OAuth2PasswordRequestForm is a FastAPI dependency that parses a login form with username and password fields. Swagger UI generates this form automatically.
  • The endpoint checks if the username exists in users_db and if the password matches the stored hash. If either fails, it returns a 401 error.
  • The "sub" key stands for "subject" in the JWT standard. It holds the unique identifier for the token owner, which is the username in this case.
  • On success, the endpoint returns a Token with the signed JWT and the type "bearer", which tells clients to send it in the Authorization: Bearer <token> header.
  • Save main.py.
  • Open http://127.0.0.1:8000/docs in your browser. You should see a new POST /token endpoint in the list.

Test the full login flow now.

  • First, register a test user. Expand the POST /register endpoint and click Try it out.
  • Enter a username and password in the request body (for example, "testuser" and "password123"), then click Execute.
  • Now expand the POST /token endpoint and click Try it out.
  • Enter the same username and password in the username and password fields, then click Execute.

You should see a 200 response with a JSON body containing access_token (a long encoded string) and token_type set to "bearer". That encoded string is your JWT.

Getting a 401 Unauthorized response?

  • Make sure you registered the user first using POST /register before trying to log in. In-memory storage resets every time the server restarts.
  • Check that the username and password you entered in the /token form match exactly what you registered with. They are case-sensitive.
  • If you see a 422 error instead of 401, make sure you are filling in the username and password form fields, not the request body JSON.

Help me debug my login endpoint.

Build the token validation dependency

Your server can issue tokens, but it cannot verify them yet. You need a function that reads the token from an incoming request, decodes it, and identifies the user.

What is a FastAPI dependency?

A dependency is a function that FastAPI runs automatically before your endpoint code executes. When you add a dependency to an endpoint, FastAPI calls it first, and your endpoint only runs if the dependency succeeds. This is how you enforce authentication without repeating the same token-checking code in every endpoint.

  • Add the get_current_user function below your create_access_token function in the helper functions section:
async def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    # Prepare a reusable 401 error
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # Decode the token and extract the username
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except InvalidTokenError:
        raise credentials_exception
    # Verify the user still exists
    if username not in users_db:
        raise credentials_exception
    return username

What does this code do?

  • Depends(oauth2_scheme) tells FastAPI to extract the Bearer token from the request's Authorization header. If the header is missing, FastAPI immediately returns a 401 error before your function even runs.
  • jwt.decode() verifies the token's signature and checks that it has not expired. If the token is tampered with or past its expiration, PyJWT raises InvalidTokenError.
  • The function extracts the "sub" claim (the username) from the decoded payload. It also confirms the user still exists in the database before returning.
  • Any endpoint that adds Depends(get_current_user) as a parameter will require a valid token. You will wire this up in the next step.
  • Save main.py.
  • Check your terminal for a clean reload with no errors.

You will not see a visible change in Swagger UI yet. The get_current_user function only takes effect when you attach it to endpoints, which happens in the next step.

Server not reloading cleanly?

  • If you see NameError: name 'oauth2_scheme' is not defined, make sure the oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") line appears above the get_current_user function in your file.
  • If you see IndentationError, make sure the try and except blocks are indented consistently inside the function.

Help me fix my get_current_user function.

Awesome, I've got everything!

Great. Double check you have saved main.py and your server is running with no errors.

I'd like to double check the full code

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

from datetime import datetime, timedelta, timezone

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

# --- App Setup ---
app = FastAPI(title="Secure Book Collection API")

# --- Security Configuration ---
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# --- Password Hashing ---
password_hash = PasswordHash.recommended()

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

# --- In-Memory Storage ---
users_db = {}
books_db = {}
book_id_counter = 0


# --- Pydantic Models ---
class BookCreate(BaseModel):
    title: str
    author: str
    year: int


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


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


class UserOut(BaseModel):
    username: str


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


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


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


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


async def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    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
    except InvalidTokenError:
        raise credentials_exception
    if username not in users_db:
        raise credentials_exception
    return username


# --- Auth Endpoints ---
@app.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def register_user(user: UserCreate):
    if user.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already registered",
        )
    users_db[user.username] = get_password_hash(user.password)
    return UserOut(username=user.username)


@app.post("/token", response_model=Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    username = form_data.username
    password = form_data.password
    if username not in users_db or not verify_password(password, users_db[username]):
        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": username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


# --- Book Endpoints ---
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
def create_book(book: BookCreate):
    global book_id_counter
    book_id_counter += 1
    new_book = Book(id=book_id_counter, **book.model_dump())
    books_db[book_id_counter] = new_book
    return new_book


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


@app.get("/books/{book_id}", response_model=Book)
def get_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    return books_db[book_id]


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    del books_db[book_id]

Your API can now register users, hash their passwords, and issue signed JWT tokens on login. Next up, you will wire the get_current_user dependency into your book endpoints so only authenticated users can create and delete books.

Protect Your API Endpoints

You now have a working authentication system. Users can register, log in, and receive JWT tokens. But there is a problem: every endpoint is still wide open.

Anyone can create, delete, or modify books without logging in. In this step, you will wire the get_current_user function into your book endpoints so that only authenticated users can make changes.

In this step, get ready to:
  • Lock down the POST and DELETE book endpoints so they require a valid token.
  • Test the protection in Swagger UI by attempting requests with and without authentication.
  • Add a user profile endpoint that identifies who is making the request.
Add authentication to your book endpoints

FastAPI uses a pattern called dependency injection to run shared logic before an endpoint executes. By adding Depends(get_current_user) as a parameter, FastAPI automatically checks for a valid token before the endpoint code runs.

  • In main.py, scroll down to the book endpoints section and find the create_book function definition.
  • Find this line:
def create_book(book: BookCreate):
  • Replace the function signature with this updated version that includes current_user as a second parameter:
def create_book(book: BookCreate, current_user: str = Depends(get_current_user)):
  • Now find the delete_book function definition further down in the same file.
  • Find this line:
def delete_book(book_id: int):
  • Replace it with this version that adds the same current_user dependency:
def delete_book(book_id: int, current_user: str = Depends(get_current_user)):

What does Depends(get_current_user) do?

When FastAPI sees Depends(get_current_user) in a function signature, it runs get_current_user before the endpoint code executes.

If the token is missing, expired, or invalid, get_current_user raises a 401 error and the endpoint never runs. If the token is valid, the authenticated username is passed into current_user so the endpoint knows who made the request.

Why keep GET endpoints public?

The two GET endpoints (list all books and get a single book) stay unprotected. This is a common API design pattern where reading data is open to everyone, but creating, updating, or deleting data requires authentication.

  • Save main.py.
  • Check your terminal where uvicorn is running. You should see a log line showing that the server reloaded automatically.

Seeing a syntax error in the terminal?

  • Make sure the current_user parameter is separated from the previous parameter by a comma.
  • Double check that the parentheses in Depends(get_current_user) are balanced. There is one opening and one closing parenthesis around the function name.
  • Confirm you only changed the def line for each function. The rest of the function body stays the same.

Help me fix a syntax error after adding Depends(get_current_user) to my book endpoints.

Test the protected endpoints

Time to see the protection in action. You will first try to use a protected endpoint without logging in, then authorize and try again.

  • Open http://127.0.0.1:8000/docs in your browser to load the Swagger UI.
  • Expand the POST /books endpoint.
  • Click Try it out.
  • Enter a book in the request body (for example, a title, author, and year) and click Execute.

You should see a 401 Unauthorized response with the detail Not authenticated. The endpoint rejected your request because you did not provide a token.

  • Scroll to the top of the Swagger UI page and click the Authorize button (it has a lock icon).
  • Enter the username and password for the user you registered in the previous step.
  • Click Authorize in the dialog.
  • Click Close to dismiss the dialog.

What is the Authorize button doing?

Behind the scenes, Swagger UI sends your credentials to the /token endpoint and receives a JWT access token. It then attaches that token as a Bearer token in the Authorization header of every subsequent request you make from the page.

  • Go back to the POST /books endpoint.
  • Click Try it out again, enter a book, and click Execute.

This time you should see a 201 Created response with your new book in the response body. The token proved your identity and the endpoint let you through.

  • Test the GET /books endpoint to confirm it still works without authorization. Public endpoints remain open.

Still getting a 401 after authorizing?

  • Make sure the username and password you entered in the Authorize dialog match exactly what you used during registration.
  • If you restarted the uvicorn server since registering, your user data was lost because it is stored in memory. Register a new user and try again.
  • Check that the tokenUrl in your OAuth2PasswordBearer is set to "token", matching your /token endpoint path.

Help me debug why I am still getting 401 errors after clicking Authorize in Swagger UI.

Add a user profile endpoint

A common API feature is a "who am I" endpoint that returns information about the currently authenticated user. This confirms the token correctly identifies the person making the request.

  • In main.py, find the login_for_access_token function. Add the following endpoint directly below it, before the book endpoints section:
# Returns the username of the currently authenticated user
@app.get("/users/me")
def read_current_user(current_user: str = Depends(get_current_user)):
    return {"username": current_user}

What does this endpoint do?

  • Depends(get_current_user) extracts and validates the JWT token from the request header.
  • The decoded username is passed into current_user, which the endpoint returns as a simple dictionary.
  • This is the same dependency used on the book endpoints. Any endpoint that needs to know who is calling it can reuse it.
  • Save main.py.
  • Go back to the Swagger UI at http://127.0.0.1:8000/docs and refresh the page.
  • Make sure you are still authorized (if the page refreshed and logged you out, click Authorize again and re-enter your credentials).
  • Expand the GET /users/me endpoint, click Try it out, then click Execute.

You should see a 200 response with a JSON body containing your username. The token told the API exactly who you are.

Don't see the GET /users/me endpoint in Swagger UI?

  • Refresh the Swagger UI page. The server may not have reloaded yet.
  • Make sure you placed the new endpoint outside the login_for_access_token function, not inside it. It should be at the same indentation level as your other @app decorators.

Help me figure out why my GET /users/me endpoint is not showing up in Swagger UI.

Awesome, I've got everything!

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

I'd like to double check the full code

Here is the complete main.py as it should look at the end of this step. Compare it with yours to spot any differences.

from datetime import datetime, timedelta, timezone

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

# --- App Setup ---
app = FastAPI(title="Secure Book Collection API")

# --- Security Configuration ---
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# --- Password Hashing ---
password_hash = PasswordHash.recommended()

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

# --- In-Memory Storage ---
users_db = {}
books_db = {}
book_id_counter = 0


# --- Pydantic Models ---
class BookCreate(BaseModel):
    title: str
    author: str
    year: int


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


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


class UserOut(BaseModel):
    username: str


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


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


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


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


async def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    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
    except InvalidTokenError:
        raise credentials_exception
    if username not in users_db:
        raise credentials_exception
    return username


# --- Auth Endpoints ---
@app.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def register_user(user: UserCreate):
    if user.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already registered",
        )
    users_db[user.username] = get_password_hash(user.password)
    return UserOut(username=user.username)


@app.post("/token", response_model=Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    username = form_data.username
    password = form_data.password
    if username not in users_db or not verify_password(password, users_db[username]):
        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": username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.get("/users/me")
def read_current_user(current_user: str = Depends(get_current_user)):
    return {"username": current_user}


# --- Book Endpoints ---
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
def create_book(book: BookCreate, current_user: str = Depends(get_current_user)):
    global book_id_counter
    book_id_counter += 1
    new_book = Book(id=book_id_counter, **book.model_dump())
    books_db[book_id_counter] = new_book
    return new_book


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


@app.get("/books/{book_id}", response_model=Book)
def get_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    return books_db[book_id]


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int, current_user: str = Depends(get_current_user)):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    del books_db[book_id]

Your API now enforces authentication on every endpoint that changes data. Next, you will add input validation so that every request body is checked for correct formatting before it reaches your endpoint logic.

Add Input Validation with Pydantic

Your API endpoints are locked down with JWT authentication. Only registered users with valid tokens can create or delete books. But there is still a gap in your defenses.

Right now, an authenticated user could submit a book with an empty title, a year of 99999, or a username that is two characters long. Pydantic's Field constraints let you define exactly what valid data looks like. FastAPI checks every incoming request against these rules automatically.

In this step, get ready to:
  • Add Field constraints to the BookCreate model for title length, author length, and year range.
  • Add validation rules to the UserCreate model including a custom validator that rejects usernames with spaces.
  • Test that invalid requests are automatically rejected with detailed 422 error responses.
Enhance the BookCreate model with Field constraints

Pydantic's Field function lets you attach rules directly to each field in a model. FastAPI evaluates these rules before your endpoint code runs, so invalid data never reaches your application logic.

  • In main.py, find the Pydantic import line near the top of your file.
  • Find this line:
from pydantic import BaseModel
  • Replace it with this updated import that adds Field and field_validator:
from pydantic import BaseModel, Field, field_validator
  • Now scroll down to the BookCreate class in the # --- Pydantic Models --- section.
  • Find this class:
class BookCreate(BaseModel):
    title: str
    author: str
    year: int
  • Replace the entire class with this enhanced version that adds Field constraints and a new genre field:
class BookCreate(BaseModel):
    # Title must be 1-200 characters
    title: str = Field(min_length=1, max_length=200)
    # Author must be 1-100 characters
    author: str = Field(min_length=1, max_length=100)
    # Year must be between 1000 and 2026
    year: int = Field(ge=1000, le=2026)
    # Genre is optional, defaults to None
    genre: str | None = None

What do these Field constraints do?

  • min_length=1 prevents empty strings. A request with an empty title is rejected before your endpoint runs.
  • max_length=200 caps the title length. This protects against oversized payloads.
  • ge=1000 means "greater than or equal to" and le=2026 means "less than or equal to." Together, they constrain the year to a realistic range.
  • genre: str | None = None makes genre optional. If a request omits it, Pydantic fills in None automatically.

Since BookCreate now includes a genre field, the Book output model needs to match.

  • Find the Book class just below BookCreate.
  • Find this line:
    year: int
  • Add a genre field on the line below it, so the Book class now looks like this:
class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int
    genre: str | None = None
  • Save main.py.

You will not see a visible change yet. The next substep adds more validation, and then you will test everything together.

Add validation to the UserCreate model

Field constraints handle simple rules like length and range. For custom logic, Pydantic provides field_validator, a decorator that lets you write your own validation function for any field.

  • In main.py, find the UserCreate class.
  • Find this class:
class UserCreate(BaseModel):
    username: str
    password: str
  • Replace the entire class with this validated version:
class UserCreate(BaseModel):
    # Username must be 3-50 characters
    username: str = Field(min_length=3, max_length=50)
    # Password must be at least 8 characters
    password: str = Field(min_length=8)

    @field_validator("username")
    @classmethod
    def username_no_spaces(cls, v: str) -> str:
        if " " in v:
            raise ValueError("Username must not contain spaces")
        return v

What does this code do?

  • Field(min_length=3, max_length=50) ensures usernames are between 3 and 50 characters long.
  • Field(min_length=8) enforces a minimum password length. Short passwords are rejected before they ever get hashed.
  • @field_validator("username") decorates a custom validation method. This method checks for spaces in the username and raises a ValueError if any are found.
  • The @classmethod decorator is required by Pydantic for field_validator functions.
  • Save main.py.
  • Check your terminal. The server should auto-reload and show Application startup complete with no errors.

Seeing a startup error?

  • Check that your import line reads from pydantic import BaseModel, Field, field_validator. A missing import causes a NameError.
  • Make sure the @field_validator("username") and @classmethod decorators are both present and in that order (@field_validator first, @classmethod second).
  • Confirm the genre: str | None = None field appears in both BookCreate and Book.

Help me fix a startup error after adding Pydantic Field validation

Awesome, I've got everything!

Great. Double check you have saved main.py and the server reloaded without errors.

I'd like to double check the full code

Here is the complete main.py with all the validation you have added so far.

from datetime import datetime, timedelta, timezone

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

# --- App Setup ---
app = FastAPI(title="Secure Book Collection API")

# --- Security Configuration ---
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# --- Password Hashing ---
password_hash = PasswordHash.recommended()

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

# --- In-Memory Storage ---
users_db = {}
books_db = {}
book_id_counter = 0


# --- 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=2026)
    genre: str | None = None


class Book(BaseModel):
    id: int
    title: str
    author: str
    year: int
    genre: str | None = None


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

    @field_validator("username")
    @classmethod
    def username_no_spaces(cls, v: str) -> str:
        if " " in v:
            raise ValueError("Username must not contain spaces")
        return v


class UserOut(BaseModel):
    username: str


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


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


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


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


async def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    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
    except InvalidTokenError:
        raise credentials_exception
    if username not in users_db:
        raise credentials_exception
    return username


# --- Auth Endpoints ---
@app.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def register_user(user: UserCreate):
    if user.username in users_db:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already registered",
        )
    users_db[user.username] = get_password_hash(user.password)
    return UserOut(username=user.username)


@app.post("/token", response_model=Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    username = form_data.username
    password = form_data.password
    if username not in users_db or not verify_password(password, users_db[username]):
        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": username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.get("/users/me")
def read_current_user(current_user: str = Depends(get_current_user)):
    return {"username": current_user}


# --- Book Endpoints ---
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED)
def create_book(book: BookCreate, current_user: str = Depends(get_current_user)):
    global book_id_counter
    book_id_counter += 1
    new_book = Book(id=book_id_counter, **book.model_dump())
    books_db[book_id_counter] = new_book
    return new_book


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


@app.get("/books/{book_id}", response_model=Book)
def get_book(book_id: int):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    return books_db[book_id]


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int, current_user: str = Depends(get_current_user)):
    if book_id not in books_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Book not found"
        )
    del books_db[book_id]
Test validation with invalid data

Your models now enforce strict rules on every incoming request. Time to prove it works by deliberately sending bad data and watching FastAPI reject it.

  • Open Swagger UI in your browser at http://127.0.0.1:8000/docs.
  • Click the Authorize button and log in with the user you registered in a previous step.

First, test a book with an empty title.

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

You should see a 422 Unprocessable Entity response. The response body shows a detailed error explaining that the title string is too short.

What is a 422 response?

HTTP status code 422 means "Unprocessable Entity." The server understood the request format, but the data inside it fails validation. FastAPI returns 422 automatically whenever a Pydantic model rejects the input.

The response body includes the exact field name, the rule that was violated, and a human-readable message. Your API clients can use this to show users exactly what went wrong.

Next, test a book with a year far in the future.

  • In the same POST /books endpoint, set title to "Future Book", author to "Test Author", and year to 99999.
  • Click Execute.

Another 422 response. This time the error explains that the year must be less than or equal to 2026.

Finally, test user registration with a short username.

  • Expand the POST /register endpoint and click Try it out.
  • Set username to "ab" and password to "testpassword".
  • Click Execute.

You should see a 422 response telling you the username must be at least 3 characters.

  • Try one more test. Set username to "bad user" (with a space) and password to "testpassword".
  • Click Execute.

This time the error message comes from your custom field_validator: "Username must not contain spaces." Your validation catches everything, from simple length checks to custom business rules.

Pro tip: valid requests still work

Try creating a book with valid data to confirm nothing is broken. Set title to "The Great Gatsby", author to "F. Scott Fitzgerald", year to 1925, and genre to "Fiction". You should get a 201 response with the new book and its genre.

Secret mission

Add Rate Limiting Middleware

Your API is secured with authentication and validation. But what stops someone from flooding your endpoints with thousands of requests? Build rate limiting middleware that blocks abusive clients automatically.

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-books-api project folder.
  • Python virtual environment (venv).
  • main.py application file.
  • Installed pip packages (FastAPI, Uvicorn, PyJWT, pwdlib, python-multipart).
  • Uvicorn development server (if still running).

Keep everything running

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

  • Your project folder, virtual environment, and installed packages will stay in place.
  • If the Uvicorn server is still running, you can keep using it at http://127.0.0.1:8000/docs.
  • Everything runs locally, so there are no costs for leaving it as-is.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your files and virtual environment will remain ready for next time.

  • Stop the Uvicorn server by pressing Ctrl + C in the terminal where it is running.
  • Deactivate your virtual environment by running deactivate in your terminal.
  • When you come back, navigate to your secure-books-api folder, reactivate the virtual environment, and run uvicorn main:app --reload again.

Delete - I don't want to use this again

Remove all project resources and start fresh.

  • Stop the Uvicorn server by pressing Ctrl + C in the terminal where it is running.
  • Deactivate the virtual environment by running deactivate.
  • Delete the entire secure-books-api folder. This removes your code, virtual environment, and all installed packages.
  • Python itself will remain installed on your system. You do not need to uninstall it.

Mission accomplished

Nicely Done!

You just built a fully secured REST API from scratch using FastAPI. Your Book Collection API started wide open, and you progressively layered on real security until only authenticated users with valid tokens could modify data.

Here's what you learned:

  • Built a complete REST API with CRUD endpoints using FastAPI and Pydantic models that validate every request automatically.
  • Secured user passwords with Argon2 hashing and implemented JWT access tokens for stateless authentication.
  • Protected API routes using FastAPI's dependency injection so only authenticated users can create or delete books.
  • Added rate limiting middleware to block API abuse using a sliding window algorithm.

Ready to quiz yourself?