Build a Secure Marketplace API

Introduction

30 Second Summary

Every time you buy something online, there's an API behind the scenes checking who you are, what you're allowed to do, and whether your request is legitimate. If any of those checks fail, real money and real data are at risk.

In this project, you will build a production-grade marketplace API with JWT authentication, role-based access control, and a full security stack using Python, FastAPI, and PostgreSQL.

What You'll Build

A secure REST API for a marketplace where users browse products, sellers manage listings, and admins have full control over everything.

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

• A PostgreSQL database with SQLAlchemy ORM models and Alembic migrations managing your schema.
• JWT authentication with short-lived access tokens, rotating refresh tokens, and Argon2id password hashing.
• A three-role RBAC system (User, Seller, Admin) with rate limiting, CORS, security headers, and a pytest test suite.
• Secret Mission: Add product search with pagination, seller analytics, and organized API documentation.

Are there any prerequisites?

You should be comfortable with basic Python (functions, classes, decorators) and familiar with HTTP concepts like GET/POST requests and status codes. No prior FastAPI or database experience is required.

Set Up Docker and Python Environment

A secure API needs a solid foundation before you write any business logic. You need a database to persist data, a clean project structure, and properly managed configuration.

In this step, you'll spin up a PostgreSQL database using Docker Compose, create the project folder layout, and install all the Python dependencies your marketplace API needs.

In this step, get ready to:

• Create the project folder structure and start PostgreSQL via Docker Compose.
• Set up a virtual environment and install all project dependencies.
• Configure environment variables with pydantic-settings.

Create the project structure and Docker Compose for PostgreSQL

Every Python project needs a well-organized folder structure. This marketplace API separates code into layers: core utilities, database models, API routes, and tests.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or Git Bash (Windows) and press Enter to open it.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and its subdirectories by running these commands:

mkdir marketplace-api
cd marketplace-api
mkdir -p app/core app/models app/api tests alembic

• Create empty __init__.py files so Python recognizes each folder as a package:

touch app/__init__.py app/core/__init__.py app/models/__init__.py app/api/__init__.py tests/__init__.py

• Confirm your folder structure by running:

ls app app/core app/models app/api

You should see __init__.py listed inside each subfolder.

Don't see the folders?

Make sure you ran cd marketplace-api before the mkdir command. If the folders already exist, the mkdir -p flag prevents errors.

help me troubleshoot my project folder structure

Now you need a database for your marketplace data. Docker Compose lets you define and run multi-container applications with a single YAML file.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Visual Studio Code and press Enter to open it.
• Click File in the top menu bar.
• Click Open Folder.
• Select the marketplace-api folder on your Desktop, then click Open.
• Create a new file called docker-compose.yml in the root of your marketplace-api folder by right-clicking in the Explorer sidebar and selecting New File.
• Paste the following into docker-compose.yml:

version: '3.8'
services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: marketplace
      POSTGRES_PASSWORD: marketplace_pass
      POSTGRES_DB: marketplace_db
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data
volumes:
  postgres_data:

What does this file do?

• image: postgres:16 pulls the official PostgreSQL 16 image from Docker Hub.
• The environment block sets the database username, password, and database name that PostgreSQL creates on first start.
• ports: '5432:5432' maps the container's PostgreSQL port to your machine so your Python app can connect.
• The volumes section stores database data in a named volume so it persists across container restarts.

• Save docker-compose.yml with Cmd+S (macOS) or Ctrl+S (Windows).

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before moving on.

ⓧ I'd like to double check the full code

version: '3.8'
services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: marketplace
      POSTGRES_PASSWORD: marketplace_pass
      POSTGRES_DB: marketplace_db
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data
volumes:
  postgres_data:

• Start the PostgreSQL container by running this command in your terminal:

docker compose up -d

What does this command do?

The docker compose up command reads your docker-compose.yml and creates the services defined in it. The -d flag runs containers in the background (detached mode) so your terminal stays free.

• Verify the container is running by running:

docker compose ps

You should see a container with postgres:16 listed and a status showing Up or Running.

Container not running?

Make sure Docker Desktop is running in the background. If it's not, open it and wait for the engine to start, then run docker compose up -d again.

If you see a port conflict error, another process is using port 5432. Stop the other process or change the port mapping in docker-compose.yml to '5433:5432'.

help me troubleshoot my Docker Compose PostgreSQL container

Create a virtual environment and install dependencies

A virtual environment isolates your project's Python packages from your system Python. This prevents version conflicts between different projects on your machine.

• Create your virtual environment by running this command from inside the marketplace-api folder:

macOS

python3 -m venv venv

• Activate the virtual environment:

source venv/bin/activate

Windows

python -m venv venv

• Activate the virtual environment:

venv\Scripts\activate

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

Don't see (venv) in your prompt?

On Windows, if you get a permissions error, run this first:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

Then try the activate command again.

help me activate my Python virtual environment

Your project needs several Python packages for the API, database, security, and testing layers.

• Create a new file called requirements.txt in the root of marketplace-api (right-click in the VS Code Explorer sidebar and select New File).
• Paste this list of dependencies into requirements.txt:

fastapi
uvicorn[standard]
sqlalchemy
alembic
argon2-cffi
python-jose[cryptography]
pydantic-settings
slowapi
httpx
pytest
pytest-asyncio
psycopg2-binary
python-multipart

What are these packages?

• fastapi is the web framework for building your API endpoints.
• sqlalchemy and alembic handle database models and migrations.
• argon2-cffi provides password hashing. python-jose[cryptography] handles JWT token creation and verification.
• slowapi adds rate limiting. pydantic-settings manages environment variable configuration.
• pytest and httpx power the test suite. psycopg2-binary connects Python to PostgreSQL.

• Save requirements.txt.

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before running the install command.

ⓧ I'd like to double check the full code

fastapi
uvicorn[standard]
sqlalchemy
alembic
argon2-cffi
python-jose[cryptography]
pydantic-settings
slowapi
httpx
pytest
pytest-asyncio
psycopg2-binary
python-multipart

• Install all the dependencies by running:

pip install -r requirements.txt

This downloads and installs all 13 packages and their sub-dependencies. It may take a minute.

• Confirm the installation succeeded by running:

pip list | grep fastapi

You should see fastapi with a version number in the output.

Seeing install errors?

If psycopg2-binary fails, make sure your virtual environment is activated (you should see (venv) in your prompt).

If you see a "pip not found" error, try pip3 install -r requirements.txt instead.

help me fix my pip install errors

Configure environment variables with pydantic-settings

Hardcoding secrets like database passwords or token keys directly in your code is a security risk. Environment variables keep sensitive values out of your source code and let you change configuration without editing Python files.

• Create a new file called .env in the root of your marketplace-api folder.
• Paste the following into .env:

DATABASE_URL=postgresql://marketplace:marketplace_pass@localhost:5432/marketplace_db
SECRET_KEY=supersecretkey1234567890abcdef
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7
CORS_ORIGINS=["http://localhost:3000"]
RATE_LIMIT_LOGIN=5/minute
RATE_LIMIT_REGISTER=3/minute

What do these variables control?

• DATABASE_URL tells your app how to connect to the PostgreSQL container you just started.
• SECRET_KEY is used to sign JWT tokens. In production, this would be a long random string.
• ACCESS_TOKEN_EXPIRE_MINUTES and REFRESH_TOKEN_EXPIRE_DAYS control how long tokens stay valid.
• RATE_LIMIT_LOGIN and RATE_LIMIT_REGISTER cap how many login and registration attempts an IP can make per minute.

• Save .env.

You also need a .env.example file that shows other developers what variables are required without exposing real secrets.

• Create a new file called .env.example in the root of marketplace-api.
• Paste the following into .env.example:

DATABASE_URL=postgresql://user:password@localhost:5432/marketplace_db
SECRET_KEY=your-secret-key-here
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7
CORS_ORIGINS=["http://localhost:3000"]
RATE_LIMIT_LOGIN=5/minute
RATE_LIMIT_REGISTER=3/minute

• Save .env.example.

Why two .env files?

The .env file holds your actual secrets and should never be committed to version control. The .env.example file acts as a template so anyone cloning your repo knows which variables to set up.

✔️ Awesome, I've got everything!

Great. Double check both files are saved before continuing.

ⓧ I'd like to double check the full code

Here is .env:

DATABASE_URL=postgresql://marketplace:marketplace_pass@localhost:5432/marketplace_db
SECRET_KEY=supersecretkey1234567890abcdef
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7
CORS_ORIGINS=["http://localhost:3000"]
RATE_LIMIT_LOGIN=5/minute
RATE_LIMIT_REGISTER=3/minute

Here is .env.example:

DATABASE_URL=postgresql://user:password@localhost:5432/marketplace_db
SECRET_KEY=your-secret-key-here
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7
CORS_ORIGINS=["http://localhost:3000"]
RATE_LIMIT_LOGIN=5/minute
RATE_LIMIT_REGISTER=3/minute

Now you'll create the Python configuration class that reads these variables automatically. The pydantic-settings library validates and type-checks your environment variables at startup.

• In VS Code, open the app/core folder in the Explorer sidebar.
• Create a new file called config.py inside app/core/.
• Paste the following into app/core/config.py:

from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
    # Database connection string
    DATABASE_URL: str
    # Secret key for signing JWT tokens
    SECRET_KEY: str
    # Token expiration settings
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    # CORS allowed origins
    CORS_ORIGINS: List[str] = ["http://localhost:3000"]
    # Rate limiting for auth endpoints
    RATE_LIMIT_LOGIN: str = "5/minute"
    RATE_LIMIT_REGISTER: str = "3/minute"

    class Config:
        env_file = ".env"

# Create a single settings instance used across the app
settings = Settings()

What does this code do?

• Settings extends BaseSettings from pydantic-settings, which automatically reads values from your .env file.
• Each field has a type annotation (str, int, List[str]). If the .env value can't be converted to the declared type, the app crashes immediately with a clear error.
• Fields with default values (like ACCESS_TOKEN_EXPIRE_MINUTES: int = 30) are optional in the .env file. Fields without defaults (like DATABASE_URL: str) are required.
• The settings = Settings() line at the bottom creates a single instance that every other module imports.

• Save app/core/config.py.
• Verify the configuration loads correctly by running this in your terminal:

python -c "from app.core.config import settings; print(settings.DATABASE_URL)"

You should see postgresql://marketplace:marketplace_pass@localhost:5432/marketplace_db printed in the terminal.

Seeing a ModuleNotFoundError?

Make sure you're running the command from the marketplace-api folder (not from inside app/).

If you see No module named 'pydantic_settings', your virtual environment may not be activated. Check for (venv) in your prompt and re-run the activate command if needed.

help me fix my pydantic-settings config import error

✔️ Awesome, I've got everything!

Great. Your configuration is loading correctly from the .env file.

ⓧ I'd like to double check the full code

from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
    DATABASE_URL: str
    SECRET_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    CORS_ORIGINS: List[str] = ["http://localhost:3000"]
    RATE_LIMIT_LOGIN: str = "5/minute"
    RATE_LIMIT_REGISTER: str = "3/minute"

    class Config:
        env_file = ".env"

settings = Settings()

Your environment is ready. You have a running PostgreSQL database, all your dependencies installed, and a type-safe configuration system. Next up, you'll define the database models and run your first migration.

Build Database Models and Migrations

Your PostgreSQL database is running and your environment is configured. Now it's time to define the actual data structure your marketplace will use.

Every marketplace needs to store two core things: users (who buy and sell) and products (what gets listed). In this step, you'll define those database tables as Python classes, then use a migration tool to create them in PostgreSQL automatically.

In this step, get ready to:

• Create the SQLAlchemy database connection layer.
• Define User and Product models with a foreign key relationship.
• Initialize Alembic and run your first database migration.

Set up the SQLAlchemy database connection

Before you can define database tables, your app needs a way to talk to PostgreSQL. SQLAlchemy is an ORM (Object-Relational Mapper) that lets you write Python classes instead of raw SQL. It needs three things: an engine (the connection), a session factory (how you open/close conversations with the DB), and a Base class (the parent all your models inherit from).

• Create a new file called database.py inside app/core/ and add the following code:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

# Create the database engine using the URL from .env
engine = create_engine(settings.DATABASE_URL)

# Session factory - each request gets its own session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class that all models inherit from
Base = declarative_base()

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

What does this code do?

This file sets up the three pillars of your database layer.

• engine creates the actual connection to PostgreSQL using the DATABASE_URL from your .env file.
• SessionLocal is a factory that creates new database sessions. Each API request gets its own session so they don't interfere with each other.
• get_db() is a FastAPI dependency. It opens a session at the start of a request and closes it when the request finishes, even if an error occurs.

• Save app/core/database.py.

Seeing an import error?

If your editor shows a red underline on from app.core.config import settings, make sure your terminal's working directory is the marketplace-api/ folder (not a subfolder). The import path resolves relative to the project root.

Help me fix my SQLAlchemy database setup.

✔️ Awesome, I've got everything!

Great. Make sure you've saved app/core/database.py before moving on.

ⓧ I'd like to double check the full code

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Define User and Product models with relationships

Your marketplace has two main entities: users who interact with the platform and products that sellers list. Each entity becomes a Python class that maps directly to a PostgreSQL table.

The User model also needs a role system. A user can be a regular buyer, a seller who lists products, or an admin with full control. You'll represent this with a Python enum.

• Create a new file called user.py inside app/models/ and add the following code:

import enum
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Enum
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.core.database import Base

# Define the three roles a user can have
class UserRole(str, enum.Enum):
    user = "user"
    seller = "seller"
    admin = "admin"

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    role = Column(Enum(UserRole), default=UserRole.user, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    # A user can have many products (if they're a seller)
    products = relationship("Product", back_populates="seller")

What does this code do?

UserRole is a Python enum that restricts the role column to exactly three values: user, seller, or admin. PostgreSQL enforces this constraint at the database level.

The products relationship tells SQLAlchemy that one User can own many Products. The back_populates argument creates a two-way link so you can navigate from a product back to its seller.

• Save app/models/user.py.

Red underline on Base import?

If from app.core.database import Base shows an error, confirm you saved database.py in the previous substep. The import path depends on that file existing.

Help me fix my User model imports.

• Now create a file called product.py inside app/models/ and add the following code:

from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.core.database import Base

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    description = Column(String)
    price = Column(Float, nullable=False)
    seller_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    is_available = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    # Link back to the seller who listed this product
    seller = relationship("User", back_populates="products")

What does this code do?

The seller_id column uses a ForeignKey pointing to users.id. This creates a hard link between a product and the user who listed it. PostgreSQL will reject any product that references a non-existent user.

The seller relationship is the other side of the back_populates you defined on the User model. Now you can access product.seller to get the user who owns it.

• Save app/models/product.py.
• Finally, open app/models/__init__.py and add these exports so other parts of the app can import models easily:

from app.models.user import User, UserRole
from app.models.product import Product

Why export models from __init__.py?

This lets you write from app.models import User, Product instead of importing from each file individually. It also ensures all models are registered with Base.metadata when the package is imported, which is critical for Alembic's autogeneration to detect them.

• Save app/models/__init__.py.

✔️ Awesome, I've got everything!

Great. Make sure all three model files are saved before continuing.

ⓧ I'd like to double check the full code

import enum
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Enum
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.core.database import Base

class UserRole(str, enum.Enum):
    user = "user"
    seller = "seller"
    admin = "admin"

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    role = Column(Enum(UserRole), default=UserRole.user, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    products = relationship("Product", back_populates="seller")

from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.core.database import Base

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    description = Column(String)
    price = Column(Float, nullable=False)
    seller_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    is_available = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    seller = relationship("User", back_populates="products")

from app.models.user import User, UserRole
from app.models.product import Product

Initialize Alembic and generate the first migration

You have the models, but PostgreSQL doesn't know about them yet. Alembic is a database migration tool that translates your Python models into actual SQL tables. It tracks every change you make to the database schema over time, so you can upgrade, rollback, or share your database structure with teammates.

• Initialize Alembic in your project by running this command from the marketplace-api/ folder:

alembic init alembic

What does this command do?

This creates an alembic.ini config file in your project root and populates the alembic/ folder with env.py (the migration runtime), a versions/ directory (where migration scripts live), and a script.py.mako template.

• Confirm the initialization worked by running:

ls alembic

You should see env.py, versions, script.py.mako, and README in the output.

Seeing 'Directory already exists and is not empty'?

If the alembic/ directory has leftover files from a previous attempt, delete its contents first. Run rm -rf alembic then try alembic init alembic again.

Help me fix my Alembic initialization.

By default, Alembic doesn't know about your models or your database URL. You need to configure alembic/env.py to point at your app's metadata and connection string.

• Open alembic/env.py in your editor.
• Find this line near the top of the file:

target_metadata = None

• Replace that line and the comments above it with the following block:

# Import your app's models and settings
from app.core.config import settings
from app.models import User, Product
from app.core.database import Base

# Point Alembic at your database
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)

# Tell Alembic about your models' tables
target_metadata = Base.metadata

What does this configuration do?

Importing User and Product ensures their table definitions are registered with Base.metadata. Without this, Alembic would see an empty metadata and generate no tables.

config.set_main_option("sqlalchemy.url", ...) overrides the placeholder URL in alembic.ini with your real database connection from .env. This keeps credentials in one place.

• Save alembic/env.py.
• Generate the initial migration by running this command:

alembic revision --autogenerate -m "create users and products tables"

What does this command do?

Alembic compares your Python models (in Base.metadata) against the current state of your database (which is empty). It generates a migration script that creates the users and products tables with all the columns you defined.

You should see output indicating a new migration file was created inside alembic/versions/.

• Apply the migration to create the tables in PostgreSQL by running:

alembic upgrade head

You should see output showing the migration being applied, ending with a message that the upgrade is complete.

• Verify the tables now exist in PostgreSQL by running:

docker compose exec db psql -U marketplace -d marketplace_db -c "\dt"

You should see a table listing that includes both users and products (plus an alembic_version table that tracks migration state).

Migration failed or tables missing?

• If you see a connection error, confirm your PostgreSQL container is running with docker compose ps. The db service should show a status of running.
• If the autogenerate step created an empty migration (no create_table calls inside it), check that alembic/env.py imports both User and Product before setting target_metadata.
• If you see a "target database is not up to date" error, run alembic upgrade head before generating new revisions.

Help me fix my Alembic migration.

Your database schema is locked in. Next up, you'll build the authentication system that protects these tables with JWT tokens and secure password hashing.

Build the Authentication System

Your database is up and running with user and product tables ready to store data. Now it's time to build the security layer that controls who can access the API.

Every production API needs authentication ("who are you?") before authorization ("what can you do?"). In this step, you'll build a complete authentication system using Argon2 password hashing and JWT tokens.

In this step, get ready to:

• Implement password hashing and JWT token utilities.
• Build user registration and login endpoints.
• Add token refresh with rotation for long-lived sessions.

Implement Argon2 password hashing and JWT utilities

Storing passwords as plain text is one of the most dangerous security mistakes an API can make. Argon2id won the Password Hashing Competition because it resists both GPU-based attacks and side-channel attacks.

For session management, JWTs (JSON Web Tokens) let the server verify a user's identity without storing session state. You'll create short-lived access tokens (30 minutes) and longer-lived refresh tokens (7 days).

• Create a new file called app/core/security.py and add the following code:

import uuid
from datetime import datetime, timedelta
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
from jose import jwt, JWTError
from app.core.config import settings

# Argon2id hasher with secure defaults (memory=65536, time=3, parallelism=4)
ph = PasswordHasher()

def hash_password(password: str) -> str:
    """Hash a plain-text password using Argon2id."""
    return ph.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    """Verify a plain-text password against its Argon2id hash."""
    try:
        return ph.verify(hashed, plain)
    except VerifyMismatchError:
        return False

def create_access_token(subject: int) -> str:
    """Create a short-lived access token (30 min) with a unique jti."""
    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": str(subject), "exp": expire, "type": "access", "jti": str(uuid.uuid4())}
    return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")

def create_refresh_token(subject: int) -> str:
    """Create a long-lived refresh token (7 days) with a unique jti."""
    expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
    payload = {"sub": str(subject), "exp": expire, "type": "refresh", "jti": str(uuid.uuid4())}
    return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")

def decode_token(token: str) -> dict:
    """Decode and validate a JWT token."""
    return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])

What does this code do?

This module contains five security functions that the rest of your API will depend on.

• hash_password() takes a plain-text password and returns an Argon2id hash. The PasswordHasher uses secure defaults automatically.
• verify_password() compares a plain password against a stored hash. If they don't match, it catches VerifyMismatchError and returns False instead of crashing.
• create_access_token() and create_refresh_token() each embed a type claim so the API can distinguish between them, and a jti (JWT ID) for tracking individual tokens.
• decode_token() validates the signature and expiration automatically. If either fails, python-jose raises JWTError.

• Save app/core/security.py.
• Verify it works by running this quick import test:

python -c "from app.core.security import hash_password; print(hash_password('test'))"

You should see a long string starting with $argon2id$ printed in the terminal.

Seeing an ImportError?

• Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
• Check that you're running the command from the marketplace-api/ root folder, not from inside app/.
• If you see an error about argon2 or jose not being found, re-run pip install -r requirements.txt.

Help me debug my security.py import error

Build registration and login endpoints

With the security utilities ready, you can now build the endpoints that users interact with. You'll define Pydantic schemas for request/response validation, then wire up the registration and login routes.

• Create the new package directories by running these commands:

mkdir -p app/schemas app/api/v1
touch app/schemas/__init__.py app/api/__init__.py app/api/v1/__init__.py

• Confirm the directories exist by running:

ls app/schemas app/api/v1

You should see __init__.py listed inside both app/schemas and app/api/v1.

• Create a new file called app/schemas/auth.py and add the following validation schemas:

from pydantic import BaseModel, EmailStr
from datetime import datetime
from app.models.user import UserRole

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

# Schema for returning user data (excludes password)
class UserResponse(BaseModel):
    id: int
    email: str
    role: UserRole
    is_active: bool
    created_at: datetime
    class Config:
        from_attributes = True

# Schema for login requests
class LoginRequest(BaseModel):
    email: EmailStr
    password: str

# Schema for token responses (register, login, refresh)
class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

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

What do these schemas do?

Pydantic schemas define exactly what shape of data the API accepts and returns. FastAPI uses these to automatically validate incoming requests and generate documentation.

• UserCreate requires a valid email format (EmailStr validates this automatically) and a password string.
• UserResponse returns user data without the hashed password. The from_attributes = True setting lets Pydantic read directly from SQLAlchemy model instances.
• TokenResponse is the standard OAuth2 token response shape with both token types and a token_type field.

Getting an error about EmailStr?

EmailStr requires the email-validator package. If you see an import error, install it by running:

pip install email-validator

Help me fix my EmailStr import error

• Save app/schemas/auth.py.
• Now create app/api/v1/auth.py with the registration and login routes:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import hash_password, verify_password, create_access_token, create_refresh_token
from app.models.user import User
from app.schemas.auth import UserCreate, LoginRequest, TokenResponse

router = APIRouter()

@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate, db: Session = Depends(get_db)):
    # Check if email already exists
    existing_user = db.query(User).filter(User.email == user_data.email).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Email already registered"
        )
    # Create new user with hashed password
    new_user = User(
        email=user_data.email,
        hashed_password=hash_password(user_data.password)
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    # Return token pair
    return TokenResponse(
        access_token=create_access_token(new_user.id),
        refresh_token=create_refresh_token(new_user.id)
    )

@router.post("/login", response_model=TokenResponse)
def login(credentials: LoginRequest, db: Session = Depends(get_db)):
    # Find user by email
    user = db.query(User).filter(User.email == credentials.email).first()
    if not user or not verify_password(credentials.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid email or password"
        )
    # Return token pair
    return TokenResponse(
        access_token=create_access_token(user.id),
        refresh_token=create_refresh_token(user.id)
    )

What do these endpoints do?

Both endpoints follow the same pattern: validate input, check credentials, return tokens.

• POST /register checks for duplicate emails (returns 409 if found), hashes the password with Argon2id, saves the user, and returns a token pair so the user is immediately logged in.
• POST /login looks up the user by email, verifies the password against the stored hash, and returns tokens. The error message is intentionally vague ("Invalid email or password") to avoid revealing whether an email exists.
• Depends(get_db) is FastAPI's dependency injection. It creates a database session for the request and automatically closes it when the response is sent.

• Save app/api/v1/auth.py.

Implement token refresh with rotation

Access tokens expire after 30 minutes for security. Instead of forcing users to log in again, the client sends its refresh token to get a new token pair.

Token rotation means every refresh creates a brand-new refresh token too. If an attacker steals an old refresh token, it's already been replaced.

• In app/api/v1/auth.py, find the existing imports at the top of the file:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import hash_password, verify_password, create_access_token, create_refresh_token
from app.models.user import User
from app.schemas.auth import UserCreate, LoginRequest, TokenResponse

• Replace the imports with this updated version that adds JWTError, decode_token, and RefreshRequest:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from jose import JWTError
from app.core.database import get_db
from app.core.security import hash_password, verify_password, create_access_token, create_refresh_token, decode_token
from app.models.user import User
from app.schemas.auth import UserCreate, LoginRequest, TokenResponse, RefreshRequest

• Now add the refresh endpoint at the bottom of app/api/v1/auth.py, below the login function:

@router.post("/refresh", response_model=TokenResponse)
def refresh(request: RefreshRequest):
    # Decode and validate the refresh token
    try:
        payload = decode_token(request.refresh_token)
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid refresh token"
        )
    # Verify this is a refresh token, not an access token
    if payload.get("type") != "refresh":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token type"
        )
    # Generate new token pair (rotation)
    user_id = int(payload["sub"])
    return TokenResponse(
        access_token=create_access_token(user_id),
        refresh_token=create_refresh_token(user_id)
    )

Why check the token type?

Both access and refresh tokens are signed with the same secret key. Without the type check, a client could send an access token to the refresh endpoint and get new tokens.

The jti (JWT ID) gives each token a unique identifier. In a production system, you'd store used jti values in a database or Redis to detect token reuse attacks.

• Save app/api/v1/auth.py.
• Now create app/main.py to wire the router into the FastAPI application:

from fastapi import FastAPI
from app.api.v1 import auth

app = FastAPI(title="Marketplace API")
app.include_router(auth.router, prefix="/api/v1/auth", tags=["Auth"])

What does include_router do?

include_router() mounts the auth router at the /api/v1/auth prefix. Every route you defined (/register, /login, /refresh) becomes /api/v1/auth/register, etc.

The tags=["Auth"] parameter groups these endpoints together in the auto-generated Swagger documentation.

• Save app/main.py.
• Start the development server by running:

fastapi dev app/main.py

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

You should see FastAPI's interactive Swagger UI with three endpoints listed under the Auth section: POST /api/v1/auth/register, POST /api/v1/auth/login, and POST /api/v1/auth/refresh.

Server won't start?

• If you see a ModuleNotFoundError, make sure you're running the command from the marketplace-api/ root folder and your virtual environment is activated.
• If port 8000 is already in use, stop any other servers running on that port.
• Check that app/api/__init__.py and app/api/v1/__init__.py both exist. Python needs these files to treat the directories as importable packages.

Help me debug why my FastAPI server won't start

✔️ Awesome, I've got everything!

Great. Press Ctrl+C in the terminal to stop the dev server for now.

ⓧ I'd like to double check the full code

Here are the complete files you should have at this point.

import uuid
from datetime import datetime, timedelta
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
from jose import jwt, JWTError
from app.core.config import settings

ph = PasswordHasher()

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

def verify_password(plain: str, hashed: str) -> bool:
    try:
        return ph.verify(hashed, plain)
    except VerifyMismatchError:
        return False

def create_access_token(subject: int) -> str:
    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": str(subject), "exp": expire, "type": "access", "jti": str(uuid.uuid4())}
    return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")

def create_refresh_token(subject: int) -> str:
    expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
    payload = {"sub": str(subject), "exp": expire, "type": "refresh", "jti": str(uuid.uuid4())}
    return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")

def decode_token(token: str) -> dict:
    return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])

from pydantic import BaseModel, EmailStr
from datetime import datetime
from app.models.user import UserRole

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

class UserResponse(BaseModel):
    id: int
    email: str
    role: UserRole
    is_active: bool
    created_at: datetime
    class Config:
        from_attributes = True

class LoginRequest(BaseModel):
    email: EmailStr
    password: str

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

class RefreshRequest(BaseModel):
    refresh_token: str

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from jose import JWTError
from app.core.database import get_db
from app.core.security import hash_password, verify_password, create_access_token, create_refresh_token, decode_token
from app.models.user import User
from app.schemas.auth import UserCreate, LoginRequest, TokenResponse, RefreshRequest

router = APIRouter()

@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate, db: Session = Depends(get_db)):
    # Check if email already exists
    existing_user = db.query(User).filter(User.email == user_data.email).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Email already registered"
        )
    # Create new user with hashed password
    new_user = User(
        email=user_data.email,
        hashed_password=hash_password(user_data.password)
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    # Return token pair
    return TokenResponse(
        access_token=create_access_token(new_user.id),
        refresh_token=create_refresh_token(new_user.id)
    )

@router.post("/login", response_model=TokenResponse)
def login(credentials: LoginRequest, db: Session = Depends(get_db)):
    # Find user by email
    user = db.query(User).filter(User.email == credentials.email).first()
    if not user or not verify_password(credentials.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid email or password"
        )
    # Return token pair
    return TokenResponse(
        access_token=create_access_token(user.id),
        refresh_token=create_refresh_token(user.id)
    )

@router.post("/refresh", response_model=TokenResponse)
def refresh(request: RefreshRequest):
    # Decode and validate the refresh token
    try:
        payload = decode_token(request.refresh_token)
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid refresh token"
        )
    # Verify this is a refresh token, not an access token
    if payload.get("type") != "refresh":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token type"
        )
    # Generate new token pair (rotation)
    user_id = int(payload["sub"])
    return TokenResponse(
        access_token=create_access_token(user_id),
        refresh_token=create_refresh_token(user_id)
    )

from fastapi import FastAPI
from app.api.v1 import auth

app = FastAPI(title="Marketplace API")
app.include_router(auth.router, prefix="/api/v1/auth", tags=["Auth"])

Your authentication system is fully operational. Users can register, log in, and refresh their sessions without re-entering credentials. Next up, you'll build the protected routes that use these tokens to enforce role-based access control.

Add Protected Routes and Role-Based Access Control

Your authentication system can issue tokens, but tokens alone do not protect anything. You need middleware that intercepts every request, validates the token, and identifies who is making the call.

In this step, you will build the authorization layer. FastAPI's dependency injection system lets you create reusable guards that extract the current user from a token and enforce role-based rules across all your endpoints.

In this step, get ready to:

• Create authentication dependencies that decode JWTs and enforce active-user checks.
• Build User endpoints with profile retrieval and admin-only listing.
• Build Product endpoints with seller ownership enforcement.

Create authentication dependencies

Every protected endpoint needs to answer two questions: who is making this request, and are they allowed to do this? You will centralize both answers in a single file that any route can import.

• Create a new file app/api/deps.py by running:

touch app/api/deps.py

• Open app/api/deps.py in your editor and paste this code:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import decode_token
from app.models.user import User, UserRole
from typing import List

# Tells FastAPI where the token endpoint lives (for Swagger UI "Authorize" button)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User:
    """Decode the JWT and return the matching User from the database."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_token(token)
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
        # Only accept access tokens, not refresh tokens
        if payload.get("type") != "access":
            raise credentials_exception
    except Exception:
        raise credentials_exception
    user = db.query(User).filter(User.id == int(user_id)).first()
    if user is None:
        raise credentials_exception
    return user

def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
    """Ensure the authenticated user account is still active."""
    if not current_user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Inactive user"
        )
    return current_user

class RoleRequired:
    """Dependency factory that restricts access to specific roles."""
    def __init__(self, allowed_roles: List[UserRole]):
        self.allowed_roles = allowed_roles

    def __call__(self, current_user: User = Depends(get_current_active_user)) -> User:
        if current_user.role not in self.allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return current_user

What does this code do?

This file defines three layers of access control that stack on top of each other:

• get_current_user extracts the Bearer token from the request header, decodes it with your decode_token utility, and fetches the matching User from the database. If anything fails, it raises a 401.
• get_current_active_user depends on get_current_user and adds one more check: is the account still active? If not, it raises a 403.
• RoleRequired is a callable class. You instantiate it with a list of allowed roles (e.g., seller and admin). When FastAPI calls it, it checks whether the user's role is in that list.

Why a callable class for roles?

FastAPI's Depends() accepts any callable. A class with a __call__ method lets you pass configuration (the allowed roles list) at definition time, then use the configured instance as a dependency on any route.

• Save app/api/deps.py.

Seeing an import error?

Make sure app/api/__init__.py exists (it can be empty). Python needs it to treat app/api/ as a package.

Also confirm your virtual environment is active. If fastapi or sqlalchemy are not found, reactivate it with source venv/bin/activate (macOS) or venv\Scripts\activate (Windows).

Help me debug my import error in app/api/deps.py

Build User endpoints

With the auth dependencies ready, you can now create routes that respond differently based on who is logged in. The User endpoints let any authenticated user view their own profile, but only an admin can list all users.

• Create app/api/v1/users.py and paste this code:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.api.deps import get_current_active_user, RoleRequired
from app.models.user import User, UserRole
from app.schemas.auth import UserResponse
from typing import List

router = APIRouter()

@router.get("/me", response_model=UserResponse)
def get_my_profile(current_user: User = Depends(get_current_active_user)):
    """Return the currently authenticated user's profile."""
    return current_user

@router.patch("/me", response_model=UserResponse)
def update_my_profile(
    email: str = None,
    current_user: User = Depends(get_current_active_user),
    db: Session = Depends(get_db)
):
    """Update the current user's email address."""
    if email:
        current_user.email = email
        db.commit()
        db.refresh(current_user)
    return current_user

@router.get("", response_model=List[UserResponse])
def list_users(
    db: Session = Depends(get_db),
    current_user: User = Depends(RoleRequired([UserRole.admin]))
):
    """List all users. Admin only."""
    return db.query(User).all()

What does this code do?

• GET /me uses get_current_active_user as a dependency. If the token is valid and the user is active, it returns the user object.
• PATCH /me accepts an optional email query parameter and commits the change to the database.
• GET /users uses RoleRequired([UserRole.admin]). Any non-admin user hitting this endpoint gets a 403 Forbidden response.

• Save app/api/v1/users.py.

Build Product endpoints with seller ownership

Products are where ownership enforcement gets interesting. A seller can only edit or delete their own products. An admin can modify any product. A regular user cannot create products at all.

• Create the Pydantic schemas for product validation. Create app/schemas/product.py and paste:

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class ProductCreate(BaseModel):
    """Schema for creating a new product listing."""
    title: str
    description: Optional[str] = None
    price: float

class ProductUpdate(BaseModel):
    """Schema for partial product updates. All fields optional."""
    title: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    is_available: Optional[bool] = None

class ProductResponse(BaseModel):
    """Schema for product data returned to clients."""
    id: int
    title: str
    description: Optional[str]
    price: float
    seller_id: int
    is_available: bool
    created_at: datetime

    class Config:
        from_attributes = True

What does this code do?

• ProductCreate validates incoming data when a seller lists a new product. Only title and price are required.
• ProductUpdate makes every field optional so sellers can update just one attribute at a time.
• ProductResponse shapes what the API sends back. The from_attributes = True config tells Pydantic to read directly from SQLAlchemy model attributes.

• Save app/schemas/product.py.
• Now create app/api/v1/products.py and paste:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.api.deps import get_current_active_user, RoleRequired
from app.models.user import User, UserRole
from app.models.product import Product
from app.schemas.product import ProductCreate, ProductUpdate, ProductResponse
from typing import List

router = APIRouter()

@router.get("", response_model=List[ProductResponse])
def list_products(db: Session = Depends(get_db)):
    """Public endpoint: list all available products."""
    return db.query(Product).filter(Product.is_available == True).all()

@router.post("", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
def create_product(
    product: ProductCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(RoleRequired([UserRole.seller, UserRole.admin]))
):
    """Create a new product. Seller or Admin only."""
    db_product = Product(**product.model_dump(), seller_id=current_user.id)
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    return db_product

@router.patch("/{product_id}", response_model=ProductResponse)
def update_product(
    product_id: int,
    product: ProductUpdate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
):
    """Update a product. Seller can edit own products, Admin can edit any."""
    db_product = db.query(Product).filter(Product.id == product_id).first()
    if not db_product:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
    # Ownership check: only the seller who created it or an admin
    if db_product.seller_id != current_user.id and current_user.role != UserRole.admin:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to edit this product")
    for key, value in product.model_dump(exclude_unset=True).items():
        setattr(db_product, key, value)
    db.commit()
    db.refresh(db_product)
    return db_product

@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_product(
    product_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
):
    """Delete a product. Seller can delete own products, Admin can delete any."""
    db_product = db.query(Product).filter(Product.id == product_id).first()
    if not db_product:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
    # Ownership check
    if db_product.seller_id != current_user.id and current_user.role != UserRole.admin:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to delete this product")
    db.delete(db_product)
    db.commit()

How does ownership enforcement work?

The PATCH and DELETE endpoints use get_current_active_user (not RoleRequired) because any active user might attempt to edit a product. The ownership logic is handled manually inside the function body.

If the product's seller_id does not match the current user's ID AND the current user is not an admin, the request gets a 403. This pattern gives sellers full control over their own listings while giving admins global override power.

• Save app/api/v1/products.py.

Now register both new routers in the application entry point.

• In app/main.py, find this line:

from app.api.v1 import auth

• Replace it with:

from app.api.v1 import auth, users, products

• Then find the app.include_router line for auth and add the two new routers below it so the full file looks like this:

from fastapi import FastAPI
from app.api.v1 import auth, users, products

app = FastAPI(title="Marketplace API")
app.include_router(auth.router, prefix="/api/v1/auth", tags=["Auth"])
app.include_router(users.router, prefix="/api/v1/users", tags=["Users"])
app.include_router(products.router, prefix="/api/v1/products", tags=["Products"])

What does this code do?

Each include_router call mounts a router at a URL prefix. Requests to /api/v1/users/me are handled by the users router. Requests to /api/v1/products go to the products router. The tags parameter organizes the Swagger UI docs into labeled sections.

• Save app/main.py.
• Start the dev server by running:

fastapi dev app/main.py

• Open http://127.0.0.1:8000/docs in your browser. You should see three sections in the Swagger UI: Auth, Users, and Products.

Server won't start?

If you see a ModuleNotFoundError, check that you have __init__.py files in app/api/, app/api/v1/, app/schemas/, and app/models/. Each can be empty but must exist.

If you see a database connection error, confirm your PostgreSQL container is running with docker compose ps.

Help me debug my FastAPI startup error

✔️ Awesome, I've got everything!

Great. Double check you have saved all four files.

ⓧ I'd like to double check the full code

Here is the full app/api/deps.py:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import decode_token
from app.models.user import User, UserRole
from typing import List

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_token(token)
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
        if payload.get("type") != "access":
            raise credentials_exception
    except Exception:
        raise credentials_exception
    user = db.query(User).filter(User.id == int(user_id)).first()
    if user is None:
        raise credentials_exception
    return user

def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
    if not current_user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Inactive user"
        )
    return current_user

class RoleRequired:
    def __init__(self, allowed_roles: List[UserRole]):
        self.allowed_roles = allowed_roles

    def __call__(self, current_user: User = Depends(get_current_active_user)) -> User:
        if current_user.role not in self.allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return current_user

Here is the full app/schemas/product.py:

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class ProductCreate(BaseModel):
    title: str
    description: Optional[str] = None
    price: float

class ProductUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    is_available: Optional[bool] = None

class ProductResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    price: float
    seller_id: int
    is_available: bool
    created_at: datetime

    class Config:
        from_attributes = True

Here is the full app/api/v1/users.py:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.api.deps import get_current_active_user, RoleRequired
from app.models.user import User, UserRole
from app.schemas.auth import UserResponse
from typing import List

router = APIRouter()

@router.get("/me", response_model=UserResponse)
def get_my_profile(current_user: User = Depends(get_current_active_user)):
    return current_user

@router.patch("/me", response_model=UserResponse)
def update_my_profile(
    email: str = None,
    current_user: User = Depends(get_current_active_user),
    db: Session = Depends(get_db)
):
    if email:
        current_user.email = email
        db.commit()
        db.refresh(current_user)
    return current_user

@router.get("", response_model=List[UserResponse])
def list_users(
    db: Session = Depends(get_db),
    current_user: User = Depends(RoleRequired([UserRole.admin]))
):
    return db.query(User).all()

Here is the full app/api/v1/products.py:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.api.deps import get_current_active_user, RoleRequired
from app.models.user import User, UserRole
from app.models.product import Product
from app.schemas.product import ProductCreate, ProductUpdate, ProductResponse
from typing import List

router = APIRouter()

@router.get("", response_model=List[ProductResponse])
def list_products(db: Session = Depends(get_db)):
    return db.query(Product).filter(Product.is_available == True).all()

@router.post("", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
def create_product(
    product: ProductCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(RoleRequired([UserRole.seller, UserRole.admin]))
):
    db_product = Product(**product.model_dump(), seller_id=current_user.id)
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    return db_product

@router.patch("/{product_id}", response_model=ProductResponse)
def update_product(
    product_id: int,
    product: ProductUpdate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
):
    db_product = db.query(Product).filter(Product.id == product_id).first()
    if not db_product:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
    if db_product.seller_id != current_user.id and current_user.role != UserRole.admin:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to edit this product")
    for key, value in product.model_dump(exclude_unset=True).items():
        setattr(db_product, key, value)
    db.commit()
    db.refresh(db_product)
    return db_product

@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_product(
    product_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
):
    db_product = db.query(Product).filter(Product.id == product_id).first()
    if not db_product:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
    if db_product.seller_id != current_user.id and current_user.role != UserRole.admin:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to delete this product")
    db.delete(db_product)
    db.commit()

Here is the full app/main.py:

from fastapi import FastAPI
from app.api.v1 import auth, users, products

app = FastAPI(title="Marketplace API")
app.include_router(auth.router, prefix="/api/v1/auth", tags=["Auth"])
app.include_router(users.router, prefix="/api/v1/users", tags=["Users"])
app.include_router(products.router, prefix="/api/v1/products", tags=["Products"])

Your API now enforces authentication and authorization on every protected route. Next up, you will add rate limiting, CORS, and security headers to harden the API against real-world attacks.

Harden Security

Your marketplace API has authentication, role-based access control, and full CRUD endpoints. But without additional hardening, an attacker could brute-force login credentials with thousands of requests per second.

In this step, you'll add three critical production security layers: rate limiting to throttle brute-force attacks, CORS and security headers to protect against cross-origin and injection attacks, and structured exception handling to prevent stack traces from leaking internal details.

In this step, get ready to:

• Add rate limiting to authentication endpoints.
• Configure CORS and security header middleware.
• Implement structured exception handling that never leaks stack traces.

Add rate limiting to auth endpoints

A brute-force attack hammers your login endpoint with thousands of password guesses per second. Rate limiting caps how many requests a single IP can make in a given time window, making brute-force attacks impractical.

• Open app/api/v1/auth.py in your editor.
• Add the following imports at the top of the file, below the existing imports:

from fastapi import Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from app.core.config import settings

# Create a rate limiter that tracks requests per IP address
limiter = Limiter(key_func=get_remote_address)

What does this code do?

The Limiter class from slowapi tracks how many requests each IP address makes. The get_remote_address function extracts the client's IP from the incoming request.

By default, slowapi uses an in-memory backend to store request counts. This means rate limits reset when the server restarts, which is fine for development.

• Add the @limiter.limit decorator to your register endpoint. Place it between the @router.post decorator and the function definition, and make sure the function accepts request: Request as its first parameter:

@router.post("/register")
@limiter.limit(settings.RATE_LIMIT_REGISTER)
async def register(request: Request, ...):
    # existing registration logic stays the same

• Add the same pattern to your login endpoint:

@router.post("/login")
@limiter.limit(settings.RATE_LIMIT_LOGIN)
async def login(request: Request, ...):
    # existing login logic stays the same

Why does the endpoint need a request parameter?

The slowapi library hooks into each request by reading the request: Request parameter. Without it, the limiter cannot identify the client or track the request count. The request argument must be explicitly named in the function signature.

What do the rate limit strings mean?

The value settings.RATE_LIMIT_LOGIN resolves to "5/minute" from your .env file. This means a single IP can attempt login at most 5 times per minute. Registration is even stricter at 3 per minute, since legitimate users rarely register multiple accounts.

• Now update app/main.py to register the rate limit exceeded handler. Replace the entire file content with the following:

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from starlette.middleware.base import BaseHTTPMiddleware
from app.core.config import settings
from app.api.v1 import auth, users, products

# Rate limiter with in-memory backend, keyed by client IP
limiter = Limiter(key_func=get_remote_address)
app = FastAPI(title="Marketplace API")

# Attach limiter to app state so slowapi can find it
app.state.limiter = limiter

# Return 429 Too Many Requests when rate limit is exceeded
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# Include routers
app.include_router(auth.router, prefix="/api/v1/auth", tags=["Auth"])
app.include_router(users.router, prefix="/api/v1/users", tags=["Users"])
app.include_router(products.router, prefix="/api/v1/products", tags=["Products"])

What does this code do?

This sets up the rate limiting infrastructure at the application level. The app.state.limiter = limiter line lets slowapi access the limiter from any request context.

The _rate_limit_exceeded_handler catches the RateLimitExceeded exception that slowapi raises when a client exceeds their limit. It automatically returns a 429 status code with rate limit headers.

• Save both app/api/v1/auth.py and app/main.py.
• Start the dev server by running:

fastapi dev app/main.py

• You should see the server start without errors. If you hit POST /api/v1/auth/login more than 5 times in a minute, you'll receive a 429 response.

Seeing an import error?

Make sure you installed slowapi via your requirements.txt in Step 1. If it's missing, run pip install slowapi with your virtual environment activated.

If you see an error about Request not being defined, confirm you added from fastapi import Request to your auth.py imports.

Help me debug this import error

Configure CORS and security middleware

Browsers enforce a same-origin policy that blocks requests from different domains. CORS middleware tells browsers which origins are allowed to call your API.

• In app/main.py, add the CORS middleware below the add_exception_handler line you just wrote:

# Allow cross-origin requests from configured frontends
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

What does this code do?

The CORSMiddleware adds the Access-Control-Allow-Origin header to responses. The allow_origins list comes from your .env file. In development it's http://localhost:3000. In production, you'd change this to your actual frontend domain.

The wildcard "*" for methods and headers means all HTTP methods and headers are allowed. The allow_credentials=True flag allows cookies and auth headers to be sent cross-origin.

Beyond CORS, production APIs should set security headers on every response to prevent common attacks like clickjacking and MIME-type sniffing.

• Add the SecurityHeadersMiddleware class below the CORS middleware in app/main.py:

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        # Prevent browsers from MIME-sniffing the content type
        response.headers["X-Content-Type-Options"] = "nosniff"
        # Block the page from being embedded in iframes (clickjacking protection)
        response.headers["X-Frame-Options"] = "DENY"
        # Enable browser XSS filtering
        response.headers["X-XSS-Protection"] = "1; mode=block"
        return response

app.add_middleware(SecurityHeadersMiddleware)

What do these security headers do?

• X-Content-Type-Options: nosniff prevents browsers from guessing the MIME type of a response, blocking attacks that disguise malicious scripts as images.
• X-Frame-Options: DENY prevents your API responses from being loaded inside an iframe, stopping clickjacking attacks.
• X-XSS-Protection: 1; mode=block activates the browser's built-in XSS filter, blocking the page if a cross-site scripting attack is detected.

• Save app/main.py.
• Restart the dev server and visit http://127.0.0.1:8000/docs in your browser.
• Open DevTools (Cmd+Option+I on macOS or F12 on Windows) and check the Network tab.
• Select any request and look at the response headers. You should see x-content-type-options: nosniff, x-frame-options: DENY, and x-xss-protection: 1; mode=block listed.

Not seeing the security headers?

Make sure app.add_middleware(SecurityHeadersMiddleware) is called after defining the class. If the class definition has an indentation error, Python won't recognize it as a valid middleware.

Also verify that from starlette.middleware.base import BaseHTTPMiddleware is at the top of your file.

Help me debug my middleware

Implement structured exception handling

When an unhandled error occurs, FastAPI's default behavior shows the full Python traceback in the response. In production, this leaks internal file paths, library versions, and code structure to attackers.

• Add the following exception handlers to app/main.py, below the app.add_middleware(SecurityHeadersMiddleware) line and above the router includes:

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    # Return structured 422 for malformed request bodies
    return JSONResponse(
        status_code=422,
        content={"detail": str(exc.errors()), "error_code": "VALIDATION_ERROR"}
    )

@app.exception_handler(Exception)
async def generic_exception_handler(request, exc):
    # Catch all unhandled exceptions - never leak stack traces
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error", "error_code": "INTERNAL_ERROR"}
    )

What does this code do?

The RequestValidationError handler catches malformed requests (wrong field types, missing required fields) and returns a consistent JSON structure with the error details and an error_code that frontend clients can use for programmatic error handling.

The generic Exception handler is the critical security layer. It catches any unexpected error and returns a generic message. The real error gets logged server-side, but the client only sees "Internal server error". No file paths, no library versions, no stack traces.

• Save app/main.py.
• Restart the dev server and visit http://127.0.0.1:8000/docs.
• Try sending a POST /api/v1/auth/register request with an invalid body (for example, an empty JSON {}).
• You should receive a 422 response with the structured format {"detail": "...", "error_code": "VALIDATION_ERROR"} instead of a raw traceback.

Still seeing raw tracebacks?

Make sure the exception handlers are decorated with @app.exception_handler(...) and not @router.exception_handler(...). These handlers must be on the app object itself to catch exceptions globally.

Also check that from fastapi.exceptions import RequestValidationError is imported from fastapi.exceptions, not from pydantic.

Help me fix my exception handlers

✔️ Awesome, I've got everything!

Great. Double check you've saved app/main.py and move on to testing.

ⓧ I'd like to double check the full code

Here is the complete app/main.py with all security hardening in place:

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from starlette.middleware.base import BaseHTTPMiddleware
from app.core.config import settings
from app.api.v1 import auth, users, products

limiter = Limiter(key_func=get_remote_address)
app = FastAPI(title="Marketplace API")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"])

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        return response

app.add_middleware(SecurityHeadersMiddleware)

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(status_code=422, content={"detail": str(exc.errors()), "error_code": "VALIDATION_ERROR"})

@app.exception_handler(Exception)
async def generic_exception_handler(request, exc):
    return JSONResponse(status_code=500, content={"detail": "Internal server error", "error_code": "INTERNAL_ERROR"})

app.include_router(auth.router, prefix="/api/v1/auth", tags=["Auth"])
app.include_router(users.router, prefix="/api/v1/users", tags=["Users"])
app.include_router(products.router, prefix="/api/v1/products", tags=["Products"])

Your API is now hardened against brute-force attacks, cross-origin abuse, and information leakage. Next up, you'll write a test suite to prove these security layers work correctly.

Write the Test Suite

Your API has authentication, authorization, and a full set of endpoints. But how do you know they all work together correctly?

Automated tests catch bugs before they reach production. In this step, you will set up a test database, write authentication tests, and write authorization tests that verify every security layer you built.

In this step, get ready to:

• Set up test infrastructure with pytest fixtures and a SQLite test database.
• Write authentication tests covering registration, login, and token refresh.
• Write authorization tests validating RBAC and ownership enforcement.

Set up the test infrastructure

Tests need their own database so they never touch production data. You will override the app's database dependency to point at a local SQLite file that gets created fresh for every test function and torn down afterwards.

• Create the file tests/conftest.py inside your marketplace-api folder by running:

touch tests/conftest.py

• Paste the following test infrastructure code into tests/conftest.py:

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.core.database import Base, get_db
from app.models.user import UserRole
from app.core.security import hash_password, create_access_token

# Use a local SQLite file for fast, isolated tests
TEST_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="function")
def db():
    # Create all tables fresh for each test
    Base.metadata.create_all(bind=engine)
    session = TestingSessionLocal()
    yield session
    session.close()
    # Tear down all tables after the test completes
    Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def client(db):
    # Override the app's database dependency to use the test session
    def override_get_db():
        yield db
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def create_test_user(db, email, password, role=UserRole.user):
    """Helper to insert a user directly and return the user + auth headers."""
    from app.models.user import User
    user = User(email=email, hashed_password=hash_password(password), role=role)
    db.add(user)
    db.commit()
    db.refresh(user)
    token = create_access_token(user.id)
    return user, {"Authorization": f"Bearer {token}"}

What does this code do?

This file sets up three things that every test in the project will share:

• A db fixture that creates all database tables before each test and drops them after. Each test starts with a completely empty database.
• A client fixture that swaps the real PostgreSQL session for the test SQLite session using FastAPI's dependency override system. The TestClient lets you call your API endpoints without a running server.
• A create_test_user() helper that inserts a user directly into the database with a hashed password and returns both the user object and a ready-to-use auth header.

• Save tests/conftest.py.
• Confirm the file exists by running:

ls tests/

• You should see conftest.py in the output.

Don't see conftest.py?

Make sure the tests/ directory exists. If it does not, create it with mkdir tests from the marketplace-api folder, then try again.

Help me debug my test setup

Write authentication tests

Authentication tests verify that your register, login, and token refresh endpoints respond correctly for both valid and invalid inputs. Each test function calls one endpoint and asserts a single expected behaviour.

• Create tests/test_auth.py and paste the following code:

from conftest import create_test_user
from app.models.user import UserRole


def test_register_success(client):
    """POST /auth/register with valid data returns 201 and tokens."""
    response = client.post("/api/v1/auth/register", json={
        "email": "newuser@example.com",
        "password": "strongpass123"
    })
    assert response.status_code == 201
    data = response.json()
    assert "access_token" in data
    assert "refresh_token" in data


def test_register_duplicate_email(client, db):
    """Registering with an existing email returns 409."""
    create_test_user(db, "taken@example.com", "password123")
    response = client.post("/api/v1/auth/register", json={
        "email": "taken@example.com",
        "password": "anotherpass"
    })
    assert response.status_code == 409


def test_login_success(client, db):
    """Login with correct credentials returns 200 and both tokens."""
    create_test_user(db, "login@example.com", "correctpass")
    response = client.post("/api/v1/auth/login", json={
        "email": "login@example.com",
        "password": "correctpass"
    })
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert "refresh_token" in data


def test_login_wrong_password(client, db):
    """Login with wrong password returns 401."""
    create_test_user(db, "login@example.com", "correctpass")
    response = client.post("/api/v1/auth/login", json={
        "email": "login@example.com",
        "password": "wrongpass"
    })
    assert response.status_code == 401


def test_token_refresh_success(client, db):
    """Refresh with a valid refresh token returns 200 and new tokens."""
    create_test_user(db, "refresh@example.com", "password123")
    # Login first to get a refresh token
    login_response = client.post("/api/v1/auth/login", json={
        "email": "refresh@example.com",
        "password": "password123"
    })
    refresh_token = login_response.json()["refresh_token"]
    # Use the refresh token
    response = client.post("/api/v1/auth/refresh", json={
        "refresh_token": refresh_token
    })
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert "refresh_token" in data


def test_token_refresh_wrong_type(client, db):
    """Using an access token as a refresh token returns 401."""
    create_test_user(db, "refresh@example.com", "password123")
    login_response = client.post("/api/v1/auth/login", json={
        "email": "refresh@example.com",
        "password": "password123"
    })
    access_token = login_response.json()["access_token"]
    # Try to refresh using the access token instead
    response = client.post("/api/v1/auth/refresh", json={
        "refresh_token": access_token
    })
    assert response.status_code == 401


def test_protected_route_no_token(client):
    """Accessing a protected route without a token returns 401."""
    response = client.get("/api/v1/users/me")
    assert response.status_code == 401

What does this test file cover?

Seven test functions, each targeting one behaviour of the auth system:

• Registration returns tokens on success and rejects duplicate emails.
• Login returns tokens for valid credentials and rejects wrong passwords.
• Token refresh works with a genuine refresh token but rejects access tokens passed in its place.
• A protected route (/api/v1/users/me) rejects requests with no token at all.

• Save tests/test_auth.py.

Seeing an import error?

If you see ModuleNotFoundError: No module named 'conftest', make sure you are running pytest from the marketplace-api/ root folder. Pytest automatically adds the tests/ directory to the import path when it discovers the test files.

Help me fix my import error

Write authorization tests

Authorization tests prove that roles are enforced correctly. A User cannot create products. A Seller can create products but cannot access admin routes. A Seller cannot edit another Seller's product.

• Create tests/test_rbac.py and paste the following code:

from conftest import create_test_user
from app.models.user import UserRole


def test_user_cannot_create_product(client, db):
    """A user with the 'user' role cannot POST /products (403)."""
    _, headers = create_test_user(db, "user@example.com", "password", UserRole.user)
    response = client.post("/api/v1/products", json={
        "title": "Test Product",
        "description": "A test product",
        "price": 29.99
    }, headers=headers)
    assert response.status_code == 403


def test_seller_can_create_product(client, db):
    """A seller can POST /products successfully (201)."""
    _, headers = create_test_user(db, "seller@example.com", "password", UserRole.seller)
    response = client.post("/api/v1/products", json={
        "title": "Test Product",
        "description": "A test product",
        "price": 29.99
    }, headers=headers)
    assert response.status_code == 201


def test_seller_cannot_access_admin_users(client, db):
    """A seller cannot GET /users (admin-only route, returns 403)."""
    _, headers = create_test_user(db, "seller@example.com", "password", UserRole.seller)
    response = client.get("/api/v1/users", headers=headers)
    assert response.status_code == 403


def test_admin_can_access_users(client, db):
    """An admin can GET /users (200)."""
    _, headers = create_test_user(db, "admin@example.com", "password", UserRole.admin)
    response = client.get("/api/v1/users", headers=headers)
    assert response.status_code == 200


def test_seller_cannot_edit_other_sellers_product(client, db):
    """Seller A cannot PATCH Seller B's product (403)."""
    # Seller A creates a product
    _, headers_a = create_test_user(db, "seller_a@example.com", "password", UserRole.seller)
    create_response = client.post("/api/v1/products", json={
        "title": "Seller A Product",
        "description": "Belongs to Seller A",
        "price": 19.99
    }, headers=headers_a)
    product_id = create_response.json()["id"]
    # Seller B tries to edit Seller A's product
    _, headers_b = create_test_user(db, "seller_b@example.com", "password", UserRole.seller)
    response = client.patch(f"/api/v1/products/{product_id}", json={
        "title": "Hijacked Product"
    }, headers=headers_b)
    assert response.status_code == 403

What does this test file cover?

Five test functions validating the three-role RBAC system:

• Role enforcement: Users are blocked from seller-only endpoints. Sellers are blocked from admin-only endpoints.
• Role success: Sellers can create products. Admins can list all users.
• Ownership enforcement: Seller A cannot modify Seller B's product even though both have the seller role.

• Save tests/test_rbac.py.
• Run the full test suite by running:

pytest tests/

• You should see all 12 tests passing with green output.

Tests failing?

If you see failures related to missing tables, make sure your app/models/__init__.py imports both User and Product. SQLAlchemy's Base.metadata.create_all() only creates tables for models that have been imported.

If you see connection errors, confirm your Docker PostgreSQL container is still running. The test suite uses SQLite, but the app imports settings which validates the .env file at import time.

Help me debug my failing tests

✔️ Awesome, I've got everything!

Great. Make sure all three test files are saved.

ⓧ I'd like to double check the full code

Here is the complete tests/conftest.py:

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.core.database import Base, get_db
from app.models.user import UserRole
from app.core.security import hash_password, create_access_token

TEST_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="function")
def db():
    Base.metadata.create_all(bind=engine)
    session = TestingSessionLocal()
    yield session
    session.close()
    Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def client(db):
    def override_get_db():
        yield db
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def create_test_user(db, email, password, role=UserRole.user):
    from app.models.user import User
    user = User(email=email, hashed_password=hash_password(password), role=role)
    db.add(user)
    db.commit()
    db.refresh(user)
    token = create_access_token(user.id)
    return user, {"Authorization": f"Bearer {token}"}

Here is the complete tests/test_auth.py:

from conftest import create_test_user
from app.models.user import UserRole


def test_register_success(client):
    """POST /auth/register with valid data returns 201 and tokens."""
    response = client.post("/api/v1/auth/register", json={
        "email": "newuser@example.com",
        "password": "strongpass123"
    })
    assert response.status_code == 201
    data = response.json()
    assert "access_token" in data
    assert "refresh_token" in data


def test_register_duplicate_email(client, db):
    """Registering with an existing email returns 409."""
    create_test_user(db, "taken@example.com", "password123")
    response = client.post("/api/v1/auth/register", json={
        "email": "taken@example.com",
        "password": "anotherpass"
    })
    assert response.status_code == 409


def test_login_success(client, db):
    """Login with correct credentials returns 200 and both tokens."""
    create_test_user(db, "login@example.com", "correctpass")
    response = client.post("/api/v1/auth/login", json={
        "email": "login@example.com",
        "password": "correctpass"
    })
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert "refresh_token" in data


def test_login_wrong_password(client, db):
    """Login with wrong password returns 401."""
    create_test_user(db, "login@example.com", "correctpass")
    response = client.post("/api/v1/auth/login", json={
        "email": "login@example.com",
        "password": "wrongpass"
    })
    assert response.status_code == 401


def test_token_refresh_success(client, db):
    """Refresh with a valid refresh token returns 200 and new tokens."""
    create_test_user(db, "refresh@example.com", "password123")
    login_response = client.post("/api/v1/auth/login", json={
        "email": "refresh@example.com",
        "password": "password123"
    })
    refresh_token = login_response.json()["refresh_token"]
    response = client.post("/api/v1/auth/refresh", json={
        "refresh_token": refresh_token
    })
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert "refresh_token" in data


def test_token_refresh_wrong_type(client, db):
    """Using an access token as a refresh token returns 401."""
    create_test_user(db, "refresh@example.com", "password123")
    login_response = client.post("/api/v1/auth/login", json={
        "email": "refresh@example.com",
        "password": "password123"
    })
    access_token = login_response.json()["access_token"]
    response = client.post("/api/v1/auth/refresh", json={
        "refresh_token": access_token
    })
    assert response.status_code == 401


def test_protected_route_no_token(client):
    """Accessing a protected route without a token returns 401."""
    response = client.get("/api/v1/users/me")
    assert response.status_code == 401

Here is the complete tests/test_rbac.py:

from conftest import create_test_user
from app.models.user import UserRole


def test_user_cannot_create_product(client, db):
    """A user with the 'user' role cannot POST /products (403)."""
    _, headers = create_test_user(db, "user@example.com", "password", UserRole.user)
    response = client.post("/api/v1/products", json={
        "title": "Test Product",
        "description": "A test product",
        "price": 29.99
    }, headers=headers)
    assert response.status_code == 403


def test_seller_can_create_product(client, db):
    """A seller can POST /products successfully (201)."""
    _, headers = create_test_user(db, "seller@example.com", "password", UserRole.seller)
    response = client.post("/api/v1/products", json={
        "title": "Test Product",
        "description": "A test product",
        "price": 29.99
    }, headers=headers)
    assert response.status_code == 201


def test_seller_cannot_access_admin_users(client, db):
    """A seller cannot GET /users (admin-only route, returns 403)."""
    _, headers = create_test_user(db, "seller@example.com", "password", UserRole.seller)
    response = client.get("/api/v1/users", headers=headers)
    assert response.status_code == 403


def test_admin_can_access_users(client, db):
    """An admin can GET /users (200)."""
    _, headers = create_test_user(db, "admin@example.com", "password", UserRole.admin)
    response = client.get("/api/v1/users", headers=headers)
    assert response.status_code == 200


def test_seller_cannot_edit_other_sellers_product(client, db):
    """Seller A cannot PATCH Seller B's product (403)."""
    _, headers_a = create_test_user(db, "seller_a@example.com", "password", UserRole.seller)
    create_response = client.post("/api/v1/products", json={
        "title": "Seller A Product",
        "description": "Belongs to Seller A",
        "price": 19.99
    }, headers=headers_a)
    product_id = create_response.json()["id"]
    _, headers_b = create_test_user(db, "seller_b@example.com", "password", UserRole.seller)
    response = client.patch(f"/api/v1/products/{product_id}", json={
        "title": "Hijacked Product"
    }, headers=headers_b)
    assert response.status_code == 403

Your API is now battle-tested. Every authentication flow and authorization rule has automated proof that it works. Next up, you will tackle the Secret Mission and add search, analytics, and API documentation.

Secret mission

Search, Analytics, and API Documentation

Your marketplace API works, but production APIs need more. Add product search with filters and pagination, seller analytics, a health check endpoint, and organized Swagger documentation to make this API truly portfolio-worthy.

Clean Up Your Resources

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 using Docker, so there are no ongoing cloud costs.

Resources you used:

• PostgreSQL container and data volume (running via Docker Compose).
• The marketplace-api/ project folder on your Desktop (includes virtual environment, source code, and test database).

Keep everything running

No action needed. Choose this if you're still actively building or plan to continue developing the API.

• Your PostgreSQL container stays up and your data persists across restarts.
• The Docker container uses a small amount of memory while running.
• You can return to the project anytime by activating your virtual environment and running fastapi dev app/main.py.

Pause - I'll come back to this later

Shut down the running containers to free up memory while keeping your data and files intact.

• Open a terminal and navigate to your marketplace-api/ folder.
• Stop the PostgreSQL container by running:

docker compose stop

• Your database data is preserved in the Docker volume. When you're ready to return, run docker compose start to bring it back up.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is irreversible.

• Open a terminal and navigate to your marketplace-api/ folder.
• Remove the Docker containers and the PostgreSQL data volume by running:

docker compose down -v

• Deactivate your virtual environment if it is still active:

deactivate

• Move up one directory and delete the entire project folder.

macOS

cd ~/Desktop
rm -rf marketplace-api

Windows

cd ~/Desktop
rmdir /s /q marketplace-api

• Confirm the folder is gone by running ls on your Desktop. You should no longer see marketplace-api in the output.

Nice Work!

Nice Work!

You've just built a production-grade secure marketplace API with FastAPI and PostgreSQL.

Every security layer in this project maps directly to patterns used in real production systems.

You've learned how to:

• Implement JWT authentication with access and refresh token rotation and Argon2id password hashing to secure user credentials.
• Build a three-role RBAC system (User, Seller, Admin) with ownership enforcement on protected API routes backed by SQLAlchemy models and Alembic migrations.
• Deploy a fully tested marketplace API with Docker Compose, rate limiting, CORS, security headers, and a pytest test suite covering authentication and authorization.
• Secret Mission: Add product search with pagination, seller analytics, and organized API documentation with OpenAPI tags.

Ready to quiz yourself?