Build a High-Performance Fintech API

Build a secure async wallet API with JWT auth, RBAC, and load testing.

Introduction

30 Second Summary

Every time you check your bank balance or send money through an app, an API is handling that request alongside thousands of others. That API needs to be both lightning-fast and airtight with security.

In this project, you will build VaultPay, a fintech digital wallet API using FastAPI and SQLModel. Your API will handle user registration, authentication, balance queries, and fund transfers under high concurrency.

What You'll Build

A production-architected fintech wallet API that handles 10,000+ concurrent connections with JWT authentication, role-based access control, rate limiting, and tuned connection pooling.

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

  • A fully functional wallet API with user registration, login, balance checks, and fund transfers between accounts.
  • JWT authentication with bcrypt password hashing, RBAC, and rate limiting protecting every endpoint.
  • Load test results proving your API handles high concurrency with tuned connection pooling and multi-worker deployment.
  • Secret Mission: Add refresh token rotation with reuse detection to prevent stolen token attacks.

Are there any prerequisites?

You should have basic Python knowledge and familiarity with REST APIs. No prior experience with FastAPI or async Python is needed.

Before We Start

VaultPay is a fintech wallet API that handles user registration, authentication, balance queries, and fund transfers. To build an API that serves thousands of users simultaneously, you need the right tools and project structure in place first.

In this step, you'll install Python, set up your project environment with all the required packages, and create the file structure that organizes VaultPay's components.

In this step, get ready to:
  • Install and verify Python 3.10+.
  • Create a project directory with a virtual environment and install all dependencies.
  • Set up the project file structure for VaultPay's modules.
Check your Python installation

VaultPay requires Python 3.10 or higher because FastAPI depends on modern Python features like type hints and async/await.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Check your Python version by running this command:
python --version

✔️ I see version 3.10 or higher

Your output looks something like Python 3.10.x or higher. You're all set to continue.

ⓧ I see an older version

If you see a version lower than 3.10, you need to install a newer version.

On macOS, the system Python is often outdated. Even after installing a newer version, the python command may still point to the old one.

  • Download Python 3.10+ from python.org/downloads and run the installer.
  • After installation, try running:
python3 --version

If python3 --version shows 3.10 or higher, use python3 instead of python for all commands in this project.

ⓧ Command not found

Python is not installed on your machine yet.

  • Download Python 3.10+ from python.org/downloads and run the installer.
  • On Windows, check the Add Python to PATH checkbox during installation.
  • Close and reopen your terminal after installation.
  • Verify by running:
python --version

On macOS, if python still isn't recognized, try python3 --version instead. Use python3 for all commands in this project if that works.

Create the project directory and install dependencies

You'll create a dedicated folder on your Desktop and a virtual environment to keep VaultPay's packages isolated from the rest of your system.

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

What is a virtual environment?

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

  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

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

Not seeing (venv) in your prompt?

  • Make sure you ran the activation command for your operating system.
  • On macOS, if you used python3 earlier, use python3 -m venv venv to create the environment, then try activating again.
  • On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned first, then try again.

Still stuck? Help me activate my Python virtual environment.

  • Install all the project dependencies by running:
pip install "fastapi[standard]" sqlmodel "python-jose[cryptography]" bcrypt pydantic-settings slowapi aiosqlite locust

You should see packages downloading and installing, ending with a Successfully installed message followed by a list of package names.

What are these packages?

  • fastapi[standard] is the web framework for building the API, bundled with Uvicorn (the server).
  • sqlmodel combines SQLAlchemy and Pydantic for database models that also validate data.
  • python-jose[cryptography] handles JWT token creation and verification.
  • bcrypt provides secure password hashing.
  • slowapi adds rate limiting to protect endpoints from abuse.
  • aiosqlite lets Python talk to SQLite asynchronously.
  • locust is a load testing tool for simulating thousands of concurrent users.

Seeing installation errors?

  • If you see a permissions error, make sure your virtual environment is activated (you should see (venv) in your prompt).
  • If a specific package fails, try installing it separately with pip install <package-name>.
  • If you used python3 earlier, use pip3 instead of pip.

Help me fix my pip install errors for VaultPay.

Set up the file structure

VaultPay's code is organized into modules. Each file handles one responsibility: database setup, authentication, data models, request schemas, and route handlers.

  • Create the directory structure and all project files by running these commands:

macOS

mkdir -p app/routes
touch app/__init__.py app/main.py app/models.py app/auth.py app/database.py app/schemas.py
touch app/routes/__init__.py app/routes/auth_routes.py app/routes/wallet_routes.py app/routes/admin_routes.py
touch locustfile.py

Windows

New-Item -ItemType Directory -Path app\routes -Force
New-Item -ItemType File -Path app\__init__.py -Force
New-Item -ItemType File -Path app\main.py -Force
New-Item -ItemType File -Path app\models.py -Force
New-Item -ItemType File -Path app\auth.py -Force
New-Item -ItemType File -Path app\database.py -Force
New-Item -ItemType File -Path app\schemas.py -Force
New-Item -ItemType File -Path app\routes\__init__.py -Force
New-Item -ItemType File -Path app\routes\auth_routes.py -Force
New-Item -ItemType File -Path app\routes\wallet_routes.py -Force
New-Item -ItemType File -Path app\routes\admin_routes.py -Force
New-Item -ItemType File -Path locustfile.py -Force
  • Confirm the files exist by running:
ls app/
ls app/routes/

You should see __init__.py, main.py, models.py, auth.py, database.py, and schemas.py in the app/ directory.

The app/routes/ directory should contain __init__.py, auth_routes.py, wallet_routes.py, and admin_routes.py.

Missing files or folders?

  • Make sure you're inside the vaultpay folder when running the commands. Run pwd (macOS) or cd (Windows) to confirm your current directory.
  • Double-check the double underscores in __init__.py. These are two underscores on each side.

Help me fix my VaultPay project file structure.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Cursor (or VS Code) and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to Desktop, select the vaultpay folder, and click Open.

You should see the app folder with all its files, locustfile.py, and the venv folder in your editor's sidebar.

Your environment is ready and your project structure is in place. Next up, you'll define the database layer with async connection pooling to handle thousands of simultaneous connections.

Define the Database Layer with Async Connection Pooling

Your development environment is ready with all dependencies installed. Now it's time to define the data foundation for VaultPay.

A fintech API handles sensitive financial data. Users, wallets, and transaction records all need persistent storage that can keep up with thousands of simultaneous requests. In this step, you'll define the data models and configure an async database engine with connection pooling to support high concurrency.

In this step, get ready to:
  • Define SQLModel table classes for users, wallets, and transactions.
  • Configure the async database engine with connection pooling parameters.
  • Create the session dependency and lifespan for FastAPI integration.
Create the data models

SQLModel lets you define database tables as Python classes. Each class becomes a table, and each attribute becomes a column.

  • Open app/models.py in your editor and add the imports and User model by pasting this code:
from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)
    hashed_password: str
    role: str = Field(default="user")
    is_active: bool = Field(default=True)

What does this code do?

The table=True parameter tells SQLModel to create an actual database table from this class.

  • id is the primary key. Setting default=None lets the database auto-generate it.
  • email is marked unique=True so no two users share an email, and index=True speeds up lookups.
  • role defaults to "user" and will later support an "admin" role for access control.
  • You won't see a change yet. The next chunk adds the Wallet and Transaction models that reference this User table.
  • Add the Wallet and Transaction models below the User class in app/models.py:

class Wallet(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(foreign_key="user.id", unique=True)
    balance: float = Field(default=0.0)


class Transaction(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    from_wallet_id: int = Field(foreign_key="wallet.id")
    to_wallet_id: int = Field(foreign_key="wallet.id")
    amount: float
    timestamp: datetime = Field(default_factory=datetime.utcnow)

What does this code do?

These two models complete the data schema for VaultPay's financial operations.

  • Wallet links to a User via foreign_key="user.id". Each user gets exactly one wallet (unique=True).
  • Transaction records fund movements between two wallets. Both from_wallet_id and to_wallet_id are foreign keys to the wallet table.
  • timestamp uses default_factory=datetime.utcnow to automatically record when each transfer happens.
  • Save app/models.py.
  • Verify the models are valid Python by running this command:
python -c "from app.models import User, Wallet, Transaction; print('Models loaded successfully')"
  • You should see Models loaded successfully printed in your terminal.

Seeing an ImportError or SyntaxError?

  • Make sure your virtual environment is still active. You should see the environment name in your terminal prompt.
  • Check that your file is saved as app/models.py (not in the project root).
  • Confirm app/__init__.py exists. Python needs this file to treat app/ as a package.

Why can't Python find my models?

✔️ Awesome, I've got everything!

Great. Double check you've saved app/models.py and move on to the database engine configuration.

ⓧ I'd like to double check the full code

from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)
    hashed_password: str
    role: str = Field(default="user")
    is_active: bool = Field(default=True)


class Wallet(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(foreign_key="user.id", unique=True)
    balance: float = Field(default=0.0)


class Transaction(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    from_wallet_id: int = Field(foreign_key="wallet.id")
    to_wallet_id: int = Field(foreign_key="wallet.id")
    amount: float
    timestamp: datetime = Field(default_factory=datetime.utcnow)
Configure the async engine with connection pooling

A connection pool keeps a set of database connections open and ready to use. Without pooling, every request would open and close its own connection, which becomes a bottleneck under heavy load.

  • Open app/database.py and add the engine configuration by pasting this code:
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlmodel import SQLModel
from contextlib import asynccontextmanager
from fastapi import FastAPI

DATABASE_URL = "sqlite+aiosqlite:///./vaultpay.db"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    echo=False,
)

What does this code do?

This creates an async database engine using SQLAlchemy's create_async_engine with the aiosqlite driver for non-blocking database access.

  • pool_size=20 keeps 20 connections open and ready. Requests grab one instantly instead of waiting for a new connection.
  • max_overflow=10 allows 10 extra connections during traffic spikes (30 total at peak).
  • pool_pre_ping=True tests each connection before using it, preventing errors from stale connections.
  • pool_recycle=3600 replaces connections older than one hour to avoid timeout issues.

Why SQLAlchemy imports instead of SQLModel?

SQLModel builds on top of SQLAlchemy for its database operations. The async engine and session tools come from SQLAlchemy's async extension because SQLModel reuses them directly. Your models still use SQLModel's simpler syntax while the connection layer uses SQLAlchemy's mature async infrastructure.

  • You won't see a change yet. The next section adds the session factory and lifespan that put this engine to work.
Create the session dependency and lifespan

FastAPI uses dependencies to inject shared resources (like database sessions) into route handlers. The lifespan context manager handles startup and shutdown tasks for the application.

  • Add the session factory and lifespan below the engine in app/database.py:

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_session():
    async with async_session() as session:
        yield session


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield
    await engine.dispose()

What does this code do?

This section wires the engine into FastAPI's request lifecycle.

  • async_sessionmaker creates a factory that produces database sessions on demand. expire_on_commit=False prevents the session from trying to reload data after a commit, which would block the async event loop.
  • get_session() is an async generator that FastAPI injects into route handlers. Each request gets its own session that auto-closes when the request finishes.
  • lifespan() runs once on app startup to create all database tables, then disposes the engine cleanly on shutdown.
  • Save app/database.py.
  • Verify the database layer loads correctly by running this command:
python -c "from app.database import engine, get_session, lifespan; print('Database layer ready')"
  • You should see Database layer ready printed in your terminal.

Seeing a ModuleNotFoundError?

  • If the error mentions sqlalchemy or aiosqlite, your virtual environment may not be active. Reactivate it and try again.
  • If the error mentions app.database, make sure you're running the command from the project root directory (the folder containing app/).
  • Check that your app/__init__.py file exists. It can be empty but must be present.

Help me fix my database module import error.

✔️ Awesome, I've got everything!

Great. Double check you've saved app/database.py and move on.

ⓧ I'd like to double check the full code

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlmodel import SQLModel
from contextlib import asynccontextmanager
from fastapi import FastAPI

DATABASE_URL = "sqlite+aiosqlite:///./vaultpay.db"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    echo=False,
)

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_session():
    async with async_session() as session:
        yield session


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield
    await engine.dispose()

Your data layer is defined and verified. Next up, you'll build the authentication system with JWT tokens and bcrypt password hashing to secure every endpoint.

Implement JWT Authentication and Password Security

Your database layer is ready to store users, wallets, and transactions. Now you need a way to securely identify who is making each request.

In this step, you will build the authentication system that protects VaultPay. This includes hashing passwords so they are never stored in plain text, issuing JWT tokens that prove a user's identity, and creating API endpoints where users can register and log in.

In this step, get ready to:
  • Build password hashing and JWT token utilities.
  • Create Pydantic validation schemas and auth endpoints.
  • Implement the get_current_user dependency to protect routes.
Build password hashing and JWT token creation

Fintech applications must never store raw passwords. Instead, you hash them with a one-way algorithm so the original value cannot be recovered.

You also need a way to prove identity on subsequent requests without sending the password every time. JWT tokens solve this by encoding a user ID and expiration into a signed string.

  • Create the authentication utilities module by adding this code to app/auth.py:
import bcrypt
from jose import jwt, JWTError
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select

from app.database import get_session
from app.models import User

# Secret key for signing JWTs (use a strong random value in production)
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Tells FastAPI where to find the token in incoming requests
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")


def hash_password(password: str) -> str:
    # Hash the password with a random salt using bcrypt
    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")


def verify_password(password: str, hashed: str) -> bool:
    # Check a plaintext password against its stored hash
    return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))

What does this code do?

SECRET_KEY is the signing key for your tokens. Anyone who has this key can forge tokens, so in production you would generate a strong random string and store it as an environment variable.

OAuth2PasswordBearer configures FastAPI to look for a Bearer token in the Authorization header of incoming requests.

bcrypt hashes passwords with a random salt each time, so even two identical passwords produce different hashes.

You won't see a change yet. The next chunk adds the JWT creation function that completes the authentication utilities.

  • Below the verify_password function in app/auth.py, add the token creation function:
def create_access_token(data: dict) -> str:
    # Copy the payload so we don't modify the original
    to_encode = data.copy()
    # Set the token to expire after 30 minutes
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    # Sign and return the JWT string
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

What does this code do?

create_access_token takes a dictionary (like {"sub": user.id}), adds an expiration timestamp, and encodes it into a signed JWT string using the HS256 algorithm.

The 30-minute expiry means stolen tokens are only usable for a short window.

  • Save app/auth.py.
  • Verify the module loads correctly by running this command:
python -c "from app.auth import create_access_token; print(create_access_token({'sub': 1}))"

You should see a long encoded string starting with eyJ. That is your JWT.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. If you see an error about jose or bcrypt not being found, re-run your install command from Step 1.

Also confirm you are running the command from the project root directory (the folder containing the app/ folder).

Help me debug a ModuleNotFoundError when importing from app.auth

Create the validation schemas and auth endpoints

Before building the endpoints, you need Pydantic schemas that validate incoming request data. These schemas reject malformed requests before your business logic ever runs.

  • Add the authentication schemas to app/schemas.py by pasting this code:
from pydantic import BaseModel, EmailStr, field_validator


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

    @field_validator("password")
    @classmethod
    def password_min_length(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        return v


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


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

What does this code do?

UserRegister requires a valid email and a password with at least 8 characters. The @field_validator decorator runs custom validation logic before the request reaches your endpoint.

TokenResponse defines the shape of the JSON response returned after login or registration.

  • Now add the wallet and transfer schemas below the TokenResponse class in the same file:
class TransferRequest(BaseModel):
    to_user_id: int
    amount: float

    @field_validator("amount")
    @classmethod
    def amount_positive(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("Amount must be positive")
        return v


class BalanceResponse(BaseModel):
    user_id: int
    balance: float


class TransferResponse(BaseModel):
    message: str
    new_balance: float

What does this code do?

TransferRequest validates that the transfer amount is positive. Negative or zero transfers would be a bug or an attack vector.

BalanceResponse and TransferResponse define the structure of outgoing JSON responses. You will use these in the wallet endpoints in the next step.

  • Save app/schemas.py.

You won't see a visible change yet. These schemas are consumed by the auth routes you will build next.

  • Create the register endpoint by adding this code to app/routes/auth_routes.py:
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select

from app.database import get_session
from app.models import User, Wallet
from app.schemas import UserRegister, UserLogin, TokenResponse
from app.auth import hash_password, verify_password, create_access_token

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post("/register", response_model=TokenResponse, status_code=201)
async def register(data: UserRegister, session: AsyncSession = Depends(get_session)):
    # Check if the email is already taken
    result = await session.execute(select(User).where(User.email == data.email))
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Email already registered")

    # Create the user with a hashed password
    user = User(email=data.email, hashed_password=hash_password(data.password))
    session.add(user)
    await session.commit()
    await session.refresh(user)

    # Give every new user a wallet with a starting balance
    wallet = Wallet(user_id=user.id, balance=1000.0)
    session.add(wallet)
    await session.commit()

    # Return a JWT so the user is immediately authenticated
    token = create_access_token({"sub": user.id})
    return TokenResponse(access_token=token)

What does this code do?

The /auth/register endpoint accepts an email and password, validates them against the UserRegister schema, then creates a new user with a hashed password and a wallet with a $1,000 starting balance.

It returns a JWT immediately so the user does not need a separate login step after registration.

  • Add the login endpoint below the register function in the same file:
@router.post("/login", response_model=TokenResponse)
async def login(data: UserLogin, session: AsyncSession = Depends(get_session)):
    # Look up the user by email
    result = await session.execute(select(User).where(User.email == data.email))
    user = result.scalar_one_or_none()

    # Verify the password against the stored hash
    if not user or not verify_password(data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials",
        )

    # Issue a fresh access token
    token = create_access_token({"sub": user.id})
    return TokenResponse(access_token=token)

What does this code do?

The /auth/login endpoint looks up the user by email, then uses bcrypt.checkpw under the hood to compare the submitted password against the stored hash.

If the credentials are valid, it returns a fresh JWT. If not, it returns a 401 error. The generic error message prevents attackers from knowing whether the email or password was wrong.

  • Save app/routes/auth_routes.py.

You cannot test these endpoints in a browser yet because app/main.py has not been wired up. That happens in the next step.

Seeing a syntax error in auth_routes.py?

Double check that the register and login functions are at the same indentation level, and that both use async def.

Also make sure there is no extra indentation at the top of the file. The imports and router = APIRouter(...) line should be at the module level (no leading spaces).

Help me fix a syntax error in my auth_routes.py file

Implement the get_current_user dependency

The auth endpoints let users get a token. Now you need a way to verify that token on every protected request. FastAPI's dependency injection system makes this easy with a reusable function.

  • In app/auth.py, add the get_current_user dependency below the create_access_token function:
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_session),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # Decode the JWT and extract the user ID
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    # Fetch the user from the database
    user = await session.get(User, user_id)
    if user is None or not user.is_active:
        raise credentials_exception
    return user

What does this code do?

get_current_user is a FastAPI dependency. Any endpoint that includes current_user: User = Depends(get_current_user) in its signature will automatically have the JWT verified and the user object loaded.

If the token is missing, expired, or tampered with, FastAPI returns a 401 response before your endpoint code even runs.

  • Now add the role-based access control helper below get_current_user in app/auth.py:
def require_role(role: str):
    def role_checker(current_user: User = Depends(get_current_user)):
        if current_user.role != role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions",
            )
        return current_user
    return Depends(role_checker)

What does this code do?

require_role is a factory function. Calling require_role("admin") returns a dependency that first authenticates the user and then checks whether their role field matches.

This is RBAC (role-based access control). You will use it in the next step to restrict admin-only endpoints.

  • Save app/auth.py.
  • Verify everything imports cleanly by running:
python -c "from app.auth import get_current_user, require_role; print('Auth module OK')"

You should see Auth module OK printed with no errors.

Getting an ImportError?

Check that your app/__init__.py file exists (it can be empty). Without it, Python does not recognize app as a package.

Also confirm that get_current_user is defined after create_access_token in the file. Python reads top to bottom, so functions must be defined before they are referenced.

Help me fix an ImportError when importing get_current_user

✔️ Awesome, I've got everything!

Great. Double check you have saved both app/auth.py, app/schemas.py, and app/routes/auth_routes.py.

ⓧ I'd like to double check the full code

Here is the complete app/auth.py:

import bcrypt
from jose import jwt, JWTError
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select

from app.database import get_session
from app.models import User

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

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


def hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")


def verify_password(password: str, hashed: str) -> bool:
    return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))


def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_session),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await session.get(User, user_id)
    if user is None or not user.is_active:
        raise credentials_exception
    return user


def require_role(role: str):
    def role_checker(current_user: User = Depends(get_current_user)):
        if current_user.role != role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions",
            )
        return current_user
    return Depends(role_checker)

Here is the complete app/schemas.py:

from pydantic import BaseModel, EmailStr, field_validator


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

    @field_validator("password")
    @classmethod
    def password_min_length(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        return v


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


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


class TransferRequest(BaseModel):
    to_user_id: int
    amount: float

    @field_validator("amount")
    @classmethod
    def amount_positive(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("Amount must be positive")
        return v


class BalanceResponse(BaseModel):
    user_id: int
    balance: float


class TransferResponse(BaseModel):
    message: str
    new_balance: float

Here is the complete app/routes/auth_routes.py:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select

from app.database import get_session
from app.models import User, Wallet
from app.schemas import UserRegister, UserLogin, TokenResponse
from app.auth import hash_password, verify_password, create_access_token

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post("/register", response_model=TokenResponse, status_code=201)
async def register(data: UserRegister, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.email == data.email))
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Email already registered")

    user = User(email=data.email, hashed_password=hash_password(data.password))
    session.add(user)
    await session.commit()
    await session.refresh(user)

    wallet = Wallet(user_id=user.id, balance=1000.0)
    session.add(wallet)
    await session.commit()

    token = create_access_token({"sub": user.id})
    return TokenResponse(access_token=token)


@router.post("/login", response_model=TokenResponse)
async def login(data: UserLogin, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.email == data.email))
    user = result.scalar_one_or_none()

    if not user or not verify_password(data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials",
        )

    token = create_access_token({"sub": user.id})
    return TokenResponse(access_token=token)

Your authentication layer is fully built. Next up, you will create the wallet and transfer endpoints that use get_current_user and require_role to protect every financial operation.

Build the Wallet and Transfer Endpoints with RBAC

Your authentication system is locked down. Users can register, log in, and receive JWT tokens that prove their identity.

Now it is time to build the endpoints that actually do something with that identity. In this step, you will create the wallet balance query, a fund transfer endpoint with atomic transactions, and an admin-only endpoint protected by role-based access control. You will also add rate limiting to prevent abuse.

In this step, get ready to:
  • Create the wallet balance and admin endpoints with role-based access control.
  • Build the fund transfer endpoint with atomic transactions and validation.
  • Wire up rate limiting and the FastAPI application entry point.
Create the balance and admin endpoints

Authenticated users need a way to check their wallet balance. Admins need a way to see all wallets in the system.

  • Open app/routes/wallet_routes.py and add the following code for the balance endpoint:
from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
from slowapi import Limiter
from slowapi.util import get_remote_address

from app.database import get_session
from app.models import User, Wallet, Transaction
from app.schemas import TransferRequest, BalanceResponse, TransferResponse
from app.auth import get_current_user, require_role

router = APIRouter(prefix="/wallet", tags=["wallet"])
limiter = Limiter(key_func=get_remote_address)


@router.get("/balance", response_model=BalanceResponse)
@limiter.limit("30/minute")
async def get_balance(
    request: Request,
    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    result = await session.execute(
        select(Wallet).where(Wallet.user_id == current_user.id)
    )
    wallet = result.scalar_one_or_none()
    if not wallet:
        raise HTTPException(status_code=404, detail="Wallet not found")
    return BalanceResponse(user_id=current_user.id, balance=wallet.balance)

What does this code do?

The get_balance function depends on get_current_user, which means FastAPI will automatically extract and validate the JWT token from the request header before the function runs.

The @limiter.limit("30/minute") decorator restricts each IP address to 30 balance checks per minute. The Request parameter is required by SlowAPI to identify the caller.

  • Save app/routes/wallet_routes.py.

Next, admins need visibility into all wallets. This is where role-based access control comes in.

  • Open app/routes/admin_routes.py and add the following code:
from fastapi import APIRouter, Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
from slowapi import Limiter
from slowapi.util import get_remote_address

from app.database import get_session
from app.models import User, Wallet
from app.auth import require_role

router = APIRouter(prefix="/admin", tags=["admin"])
limiter = Limiter(key_func=get_remote_address)


@router.get("/wallets")
@limiter.limit("10/minute")
async def list_all_wallets(
    request: Request,
    current_user: User = require_role("admin"),
    session: AsyncSession = Depends(get_session),
):
    result = await session.execute(select(Wallet))
    wallets = result.scalars().all()
    return [{"user_id": w.user_id, "balance": w.balance} for w in wallets]

How does require_role work?

The require_role("admin") dependency first validates the JWT (via get_current_user), then checks that the user's role field is "admin". If either check fails, the request is rejected before the function body ever runs.

This is RBAC in action. Different users get different levels of access based on their assigned role.

  • Save app/routes/admin_routes.py.

Seeing import errors in your editor?

Make sure your virtual environment is active. The slowapi package needs to be installed for the import to resolve. If your editor still shows red underlines after activating the venv, try reloading the editor window.

help me fix import errors for slowapi in my editor

Build the fund transfer endpoint

Transfers are the core of any fintech API. They need multiple safety checks to prevent invalid transactions.

  • In app/routes/wallet_routes.py, add the transfer endpoint below the get_balance function:
@router.post("/transfer", response_model=TransferResponse)
@limiter.limit("10/minute")
async def transfer(
    request: Request,
    data: TransferRequest,
    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    if data.to_user_id == current_user.id:
        raise HTTPException(status_code=400, detail="Cannot transfer to yourself")

    result = await session.execute(
        select(Wallet).where(Wallet.user_id == current_user.id)
    )
    sender_wallet = result.scalar_one_or_none()
    if not sender_wallet:
        raise HTTPException(status_code=404, detail="Sender wallet not found")

    if sender_wallet.balance < data.amount:
        raise HTTPException(status_code=400, detail="Insufficient balance")

    result = await session.execute(
        select(Wallet).where(Wallet.user_id == data.to_user_id)
    )
    receiver_wallet = result.scalar_one_or_none()
    if not receiver_wallet:
        raise HTTPException(status_code=404, detail="Recipient wallet not found")

    sender_wallet.balance -= data.amount
    receiver_wallet.balance += data.amount

    transaction = Transaction(
        from_wallet_id=sender_wallet.id,
        to_wallet_id=receiver_wallet.id,
        amount=data.amount,
    )
    session.add(transaction)
    await session.commit()

    return TransferResponse(
        message="Transfer successful", new_balance=sender_wallet.balance
    )

What makes this transfer atomic?

Both the debit (subtract from sender) and credit (add to receiver) happen within the same database session. The single await session.commit() at the end means either both changes are saved, or neither is. No money disappears between wallets.

The validation layers are ordered from cheapest to most expensive. Self-transfer check is a simple comparison. Balance sufficiency requires one database read. Recipient existence requires another. This ordering avoids unnecessary queries when early checks fail.

  • Save app/routes/wallet_routes.py.

Not sure where to place the transfer function?

The transfer function goes directly below the get_balance function, separated by two blank lines. Make sure it is at the module level (no indentation), not nested inside another function.

help me place the transfer endpoint correctly in wallet_routes.py

Wire up rate limiting and the app entry point

The rate limit decorators on your endpoints need an app-level handler to return proper 429 responses. Without it, exceeding a rate limit would cause an unhandled server error.

  • Open app/main.py and add the following code to create the application entry point:
from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

from app.database import lifespan
from app.routes.auth_routes import router as auth_router
from app.routes.wallet_routes import router as wallet_router
from app.routes.admin_routes import router as admin_router

limiter = Limiter(key_func=get_remote_address)

app = FastAPI(title="VaultPay", version="1.0.0", lifespan=lifespan)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

app.include_router(auth_router)
app.include_router(wallet_router)
app.include_router(admin_router)


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

What does this code do?

  • app.state.limiter attaches the limiter instance to the app so that route-level @limiter.limit() decorators can function.
  • add_exception_handler(RateLimitExceeded, ...) catches rate limit violations and returns a proper HTTP 429 response instead of a 500 error.
  • The lifespan parameter connects the database startup and shutdown logic you built in Step 2.
  • The /health endpoint gives you a quick way to verify the server is running.

Why rate limit a fintech API?

Without rate limiting, a single client could flood your transfer endpoint with thousands of requests per second. In fintech, this is especially dangerous because each request can move money.

The limits you set (10/minute on transfers, 30/minute on balance checks) allow normal usage while blocking automated abuse. Transfers get a tighter limit because they have higher risk.

  • Save app/main.py.
  • Test that the server starts by running:
uvicorn app.main:app --reload
  • You should see output showing the server running on http://127.0.0.1:8000.
  • Open http://127.0.0.1:8000/health in your browser. You should see {"status":"healthy"}.
  • Press Ctrl+C in the terminal to stop the server.

Server won't start?

Check that your virtual environment is active and all packages are installed. A common issue is a typo in the import paths. Make sure app/routes/__init__.py exists (even if empty) so Python treats the routes folder as a package.

help me debug why my FastAPI server won't start

✔️ Awesome, I've got everything!

Great. Double check that you have saved all three files before moving on.

ⓧ I'd like to double check the full code

Here is the full code for each file you created in this step.

from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
from slowapi import Limiter
from slowapi.util import get_remote_address

from app.database import get_session
from app.models import User, Wallet, Transaction
from app.schemas import TransferRequest, BalanceResponse, TransferResponse
from app.auth import get_current_user, require_role

router = APIRouter(prefix="/wallet", tags=["wallet"])
limiter = Limiter(key_func=get_remote_address)


@router.get("/balance", response_model=BalanceResponse)
@limiter.limit("30/minute")
async def get_balance(
    request: Request,
    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    result = await session.execute(
        select(Wallet).where(Wallet.user_id == current_user.id)
    )
    wallet = result.scalar_one_or_none()
    if not wallet:
        raise HTTPException(status_code=404, detail="Wallet not found")
    return BalanceResponse(user_id=current_user.id, balance=wallet.balance)


@router.post("/transfer", response_model=TransferResponse)
@limiter.limit("10/minute")
async def transfer(
    request: Request,
    data: TransferRequest,
    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    if data.to_user_id == current_user.id:
        raise HTTPException(status_code=400, detail="Cannot transfer to yourself")

    result = await session.execute(
        select(Wallet).where(Wallet.user_id == current_user.id)
    )
    sender_wallet = result.scalar_one_or_none()
    if not sender_wallet:
        raise HTTPException(status_code=404, detail="Sender wallet not found")

    if sender_wallet.balance < data.amount:
        raise HTTPException(status_code=400, detail="Insufficient balance")

    result = await session.execute(
        select(Wallet).where(Wallet.user_id == data.to_user_id)
    )
    receiver_wallet = result.scalar_one_or_none()
    if not receiver_wallet:
        raise HTTPException(status_code=404, detail="Recipient wallet not found")

    sender_wallet.balance -= data.amount
    receiver_wallet.balance += data.amount

    transaction = Transaction(
        from_wallet_id=sender_wallet.id,
        to_wallet_id=receiver_wallet.id,
        amount=data.amount,
    )
    session.add(transaction)
    await session.commit()

    return TransferResponse(
        message="Transfer successful", new_balance=sender_wallet.balance
    )
from fastapi import APIRouter, Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
from slowapi import Limiter
from slowapi.util import get_remote_address

from app.database import get_session
from app.models import User, Wallet
from app.auth import require_role

router = APIRouter(prefix="/admin", tags=["admin"])
limiter = Limiter(key_func=get_remote_address)


@router.get("/wallets")
@limiter.limit("10/minute")
async def list_all_wallets(
    request: Request,
    current_user: User = require_role("admin"),
    session: AsyncSession = Depends(get_session),
):
    result = await session.execute(select(Wallet))
    wallets = result.scalars().all()
    return [{"user_id": w.user_id, "balance": w.balance} for w in wallets]
from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

from app.database import lifespan
from app.routes.auth_routes import router as auth_router
from app.routes.wallet_routes import router as wallet_router
from app.routes.admin_routes import router as admin_router

limiter = Limiter(key_func=get_remote_address)

app = FastAPI(title="VaultPay", version="1.0.0", lifespan=lifespan)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

app.include_router(auth_router)
app.include_router(wallet_router)
app.include_router(admin_router)


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

Your API now has wallet operations, fund transfers with validation, admin access control, and rate limiting. Next up, you will deploy with multiple workers and load test your API to prove it handles high concurrency.

Deploy with Multi-Worker Uvicorn and Load Test

Your VaultPay API has authentication, wallet operations, rate limiting, and connection pooling. The final test is proving it can handle real concurrent traffic.

In this step, you'll deploy VaultPay with multiple Uvicorn workers to serve requests in parallel. Then you'll hammer the API with Locust, a load testing tool that simulates hundreds of simultaneous users hitting your endpoints.

In this step, get ready to:
  • Run VaultPay with multi-worker Uvicorn for parallel request handling.
  • Write a Locust load test that simulates concurrent API users.
  • Execute the load test and analyze performance metrics.
Run the API with multi-worker Uvicorn

A single Uvicorn process handles requests on one CPU core. By spawning multiple workers, each request can be served by a separate process in parallel.

  • Start VaultPay with 4 worker processes by running the command for your operating system:

macOS

uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop --http httptools

The --loop uvloop and --http httptools flags use faster C-based implementations for the event loop and HTTP parser. These are only available on macOS and Linux.

Windows

uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

On Windows, the uvloop and httptools packages are not available. The multi-worker setup still works with Python's default asyncio loop.

What do these flags do?

  • --workers 4 spawns 4 separate processes, each running your FastAPI app independently.
  • --host 0.0.0.0 listens on all network interfaces so Locust can connect.
  • --port 8000 sets the port your API listens on.
  • --loop uvloop replaces Python's default event loop with a faster C-based implementation (macOS/Linux only).
  • --http httptools uses a faster HTTP parser written in C (macOS/Linux only).
  • You should see output showing multiple workers starting, including lines that say Started server process for each worker.
  • Leave this terminal running. You'll open a second terminal for the load test.

Seeing a ModuleNotFoundError?

If you see an error about uvloop or httptools not being found on macOS, install them by running pip install uvloop httptools in your virtual environment. On Windows, use the command without the --loop uvloop --http httptools flags.

help me fix this Uvicorn startup error

Write the Locust load test

Locust simulates real user behavior by spawning virtual users that each run tasks against your API. Each simulated user registers, authenticates, then repeatedly hits your endpoints.

  • Open locustfile.py (in the root of your project folder) and paste the following code:
from locust import HttpUser, task, between
import random
import string


class VaultPayUser(HttpUser):
    wait_time = between(0.1, 0.5)

    def on_start(self):
        # Generate a unique email for each simulated user
        random_suffix = "".join(random.choices(string.ascii_lowercase, k=8))
        self.email = f"user_{random_suffix}@test.com"
        self.password = "testpass123"

        # Register a new account and store the JWT token
        response = self.client.post(
            "/auth/register",
            json={"email": self.email, "password": self.password},
        )
        if response.status_code == 201:
            self.token = response.json()["access_token"]
        else:
            # Fall back to login if registration fails (e.g. email collision)
            response = self.client.post(
                "/auth/login",
                json={"email": self.email, "password": self.password},
            )
            self.token = response.json()["access_token"]

        # Set auth header for all subsequent requests
        self.headers = {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def check_balance(self):
        # Balance checks happen 3x more often than transfers
        self.client.get("/wallet/balance", headers=self.headers)

    @task(1)
    def transfer_funds(self):
        # Transfers are less frequent, mimicking real usage
        self.client.post(
            "/wallet/transfer",
            json={"to_user_id": 1, "amount": 0.01},
            headers=self.headers,
        )

What does this code do?

  • VaultPayUser defines one simulated user. Locust spawns hundreds of these simultaneously.
  • on_start runs once per user at startup. It registers a unique account and stores the JWT token for authenticated requests.
  • @task(3) and @task(1) control how often each action runs. Balance checks happen 3x more often than transfers, mimicking real usage patterns.
  • wait_time = between(0.1, 0.5) adds a random pause between tasks to simulate realistic human behavior.
  • Save locustfile.py.

Seeing a syntax error?

Make sure the f-strings (lines with f"...") use regular double quotes, not curly or smart quotes. Also confirm the indentation uses consistent spaces throughout.

help me fix a syntax error in my locustfile.py

✔️ Awesome, I've got everything!

Great. Double check you've saved locustfile.py before running the load test.

ⓧ I'd like to double check the full code

from locust import HttpUser, task, between
import random
import string


class VaultPayUser(HttpUser):
    wait_time = between(0.1, 0.5)

    def on_start(self):
        random_suffix = "".join(random.choices(string.ascii_lowercase, k=8))
        self.email = f"user_{random_suffix}@test.com"
        self.password = "testpass123"

        response = self.client.post(
            "/auth/register",
            json={"email": self.email, "password": self.password},
        )
        if response.status_code == 201:
            self.token = response.json()["access_token"]
        else:
            response = self.client.post(
                "/auth/login",
                json={"email": self.email, "password": self.password},
            )
            self.token = response.json()["access_token"]

        self.headers = {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def check_balance(self):
        self.client.get("/wallet/balance", headers=self.headers)

    @task(1)
    def transfer_funds(self):
        self.client.post(
            "/wallet/transfer",
            json={"to_user_id": 1, "amount": 0.01},
            headers=self.headers,
        )
Run the load test and analyze results

With Uvicorn running in your first terminal, you'll now open a second terminal and fire 500 simulated users at the API over 30 seconds.

  • Open a new terminal window. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
  • Navigate to your project folder and activate your virtual environment in this new terminal.
  • Run the load test by running this command:
locust -f locustfile.py --headless -u 500 -r 50 --run-time 30s --host http://localhost:8000

What do these flags mean?

  • --headless runs Locust without a web UI, printing results directly to the terminal.
  • -u 500 spawns 500 concurrent simulated users.
  • -r 50 ramps up at 50 new users per second until reaching 500.
  • --run-time 30s stops the test after 30 seconds.
  • --host http://localhost:8000 tells Locust where your running API is located.

The test takes about 30 seconds. You'll see a statistics table printed every few seconds showing request counts, response times, and throughput.

  • When the test finishes, look at the final summary table. The key metrics to find are:
  • req/s (requests per second): your throughput. Higher is better.
  • Avg and Max columns: average and maximum response times in milliseconds.
  • # fails column: how many requests failed. A low percentage means your API is stable under load.

What should I see?

A healthy result shows the Aggregated row with hundreds of requests per second, average response times under 200ms, and a fail rate below 5%. Some transfer failures are expected because of rate limiting.

Locust also prints a response time percentiles table after the summary. The 95th percentile (p95) value tells you the worst-case latency for 95% of your users.

Seeing connection refused errors?

Make sure Uvicorn is still running in your first terminal. If it crashed, restart it with the multi-worker command from the previous substep. Also confirm the --host flag in the Locust command points to http://localhost:8000 where Uvicorn is listening.

help me troubleshoot my Locust load test connection errors

  • After reviewing your results, stop the Uvicorn server in your first terminal by pressing Ctrl+C.

Why 500 users instead of 10,000?

Running 10,000 users from a single machine would saturate your local CPU and network before stressing the API itself. The 500-user test proves your architecture (async handlers, connection pooling, multi-worker deployment) works under concurrent load.

In production, you'd use distributed Locust across multiple machines to reach 10k. The patterns you've built here scale directly to that setup.

Secret mission

Refresh Token Rotation with Reuse Detection

Your VaultPay API uses short-lived access tokens, but what happens when they expire? Users would need to log in again every 30 minutes. In production fintech apps, refresh tokens solve this by letting clients silently obtain new access tokens. But refresh tokens themselves can be stolen. Your mission: implement refresh token rotation with reuse detection, an enterprise-grade pattern that immediately revokes an entire token family if a stolen token is ever replayed.

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

Resources you used:

  • VaultPay project folder on your Desktop (contains all Python source files, the app/ directory, and locustfile.py).
  • Python virtual environment (inside your project folder).
  • SQLite database file vaultpay.db (created when the API first ran).

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with VaultPay.

  • Your project files, virtual environment, and database stay on your Desktop.
  • Next time you want to work on the project, activate your virtual environment and run Uvicorn again.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files stay on disk for next time.

  • Stop the Uvicorn server by pressing Ctrl+C in the terminal where it's running.
  • Deactivate your virtual environment by running:
deactivate

Your project folder and all files remain on your Desktop. To resume later, navigate back into the folder, activate the virtual environment, and start Uvicorn.

Delete - I don't want to use this again

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

  • Stop the Uvicorn server by pressing Ctrl+C in the terminal where it's running.
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the entire project folder by running:

macOS

cd ~/Desktop
rm -rf vaultpay

Windows

cd ~\Desktop
rmdir /s /q vaultpay

This removes all source files, the virtual environment, installed packages, and the vaultpay.db database in one step.

Nice Work!

Nice Work!

You just built VaultPay, a production-grade fintech wallet API that handles hundreds of concurrent users with security hardening at every layer.

Here's what you learned:

  • Built an async FastAPI application with tuned connection pooling and SQLModel data models for high-throughput database access.
  • Implemented secure JWT authentication with bcrypt password hashing, role-based access control, and rate limiting to protect every endpoint.
  • Deployed with multi-worker Uvicorn and proved performance under concurrent load using Locust load testing.
  • Added refresh token rotation with reuse detection to prevent stolen token attacks and revoke compromised session families.

Ready to quiz yourself?