URL Shortener API with JWT Auth
Build a FastAPI URL shortener with JWT auth, background tasks, and middleware.
Introduction
30 Second Summary
Every time you click a shortened link, there's an API behind the scenes that looks up the original URL, redirects you there, and quietly logs that you clicked. Building one yourself teaches you authentication, background processing, and professional API architecture all in one project.
In this project, you will build a URL shortener REST API where users register, log in with JWT tokens, create short URLs, and get click analytics. You'll organize everything into a multi-file project structure with custom middleware.
What You'll Build
A production-patterned FastAPI application that handles user authentication, URL shortening with redirect, and background click tracking, organized across multiple files using APIRouter.
By the end of this project, you'll have:
- A JWT authentication system with user registration, login, and protected routes using OAuth2 password flow and Argon2 password hashing.
- URL shortening with redirect and background tasks that silently record click analytics (timestamp and IP address) without slowing down the response.
- A professionally structured multi-file project with custom middleware that logs every request with processing time, plus a SQLite database storing users, URLs, and clicks.
- Secret Mission: add a rate-limiting middleware that restricts each IP to 10 requests per minute.
Are there any prerequisites?
This project assumes you have basic FastAPI knowledge (creating routes, Pydantic models, path and query parameters) and Python fundamentals. You do not need prior experience with authentication or middleware.
Before We Start
A URL shortener takes a long web address and gives back a short code that redirects visitors to the original destination. Behind the scenes, it needs user accounts, protected routes, and analytics tracking.
In this step, you will create the project folder structure, install the dependencies, and confirm everything runs correctly before writing any real application code.
In this step, get ready to:
- Create the project folder with a multi-file package structure.
- Install FastAPI and authentication dependencies in a virtual environment.
- Verify the setup by running a minimal FastAPI development server.
Create the project folder and file structure
Professional FastAPI projects split code across multiple files inside a Python package. You will create that structure now so every file has a home from the start.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop
- Create your project folder and move into it by running these commands:
mkdir url-shortener
cd url-shortener
Now create the package directories and the empty __init__.py files that tell Python these folders are importable packages.
macOS
- Create the directory structure and package files by running these commands:
mkdir -p app/routers
touch app/__init__.py app/routers/__init__.py
Windows
- Create the directory structure and package files by running these commands:
mkdir app/routers
New-Item -ItemType File -Path "app/__init__.py", "app/routers/__init__.py"
- Confirm the files exist by running:
ls app
ls app/routers
You should see __init__.py listed in both outputs.
What are __init__.py files?
An __init__.py file marks a folder as a Python package. Without it, Python cannot import modules from that folder using from app.routers import urls style imports.
These files can be completely empty and still serve their purpose.
Don't see __init__.py in the output?
Make sure you ran the touch or New-Item command from inside the url-shortener folder. Run pwd to check your current location.
Help me find where my __init__.py files went.
Set up the virtual environment and install dependencies
A virtual environment keeps this project's packages separate from your system Python. This prevents version conflicts between projects.
- Create a virtual environment by running this command:
python3 -m venv venv
- Activate the virtual environment:
macOS
source venv/bin/activate
Windows
.\venv\Scripts\activate
You should see (venv) appear at the beginning of your terminal prompt.
What is a virtual environment?
A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay local to this project and do not affect other Python projects on your machine.
- Open VS Code in the current folder by running:
code .
- In VS Code, create a new file called requirements.txt in the root of your url-shortener folder.
- Paste the following contents into requirements.txt:
fastapi[standard]==0.136.3
pyjwt==2.13.0
pwdlib[argon2]==0.3.0
What are these packages?
- fastapi[standard] installs FastAPI along with uvicorn (the server), python-multipart (for form data), and the FastAPI CLI.
- pyjwt is the library for creating and verifying JWT tokens.
- pwdlib[argon2] provides password hashing using the Argon2 algorithm, the current recommendation for securely storing passwords.
- Save requirements.txt.
- Return to your terminal and install all dependencies by running:
pip install -r requirements.txt
You should see each package download and install with a final message confirming success.
Seeing installation errors?
Make sure (venv) appears in your terminal prompt. If it does not, the virtual environment is not active. Run the activation command again for your platform.
Help me fix my pip install errors.
✔️ Awesome, I've got everything!
Great. Double check you have saved requirements.txt.
ⓧ I'd like to double check the full code
Here is the complete requirements.txt file:
fastapi[standard]==0.136.3
pyjwt==2.13.0
pwdlib[argon2]==0.3.0
Verify the setup by running a minimal server
Before building the full application, confirm that FastAPI starts correctly with a single endpoint.
- In VS Code, create a new file at app/main.py inside the app folder.
- Paste the following code into app/main.py:
from fastapi import FastAPI
app = FastAPI(title="URL Shortener API", version="1.0.0")
@app.get("/", tags=["Health"])
def health_check():
return {"status": "ok", "message": "URL Shortener API is running"}
What does this code do?
- FastAPI(title=..., version=...) creates the application instance with metadata that appears in the auto-generated docs.
- @app.get("/") registers a health check route at the root URL so you can quickly confirm the server is alive.
- Save app/main.py.
- Start the development server by running:
fastapi dev app/main.py
Your terminal should show that Uvicorn is running on http://127.0.0.1:8000.
- Open your browser and go to http://127.0.0.1:8000.
You should see a JSON response:
{"status": "ok", "message": "URL Shortener API is running"}
- Back in your terminal, press Ctrl+C to stop the server.
Server not starting?
If you see a "command not found" error for fastapi, make sure your virtual environment is still active. Look for (venv) in your prompt.
If you see an import error, confirm you ran pip install -r requirements.txt successfully inside the virtual environment.
Help me get my FastAPI server running.
Your project skeleton is ready and FastAPI runs without errors. Next up, you will build the database layer and define the data models that power the URL shortener.
Set Up the Database and Models
Your project structure is ready and FastAPI confirmed it can run. Now it's time to build the data layer that your API stores everything on.
Every URL shortener needs to remember things: which users exist, which URLs were shortened, and who clicked what. In this step, you'll create the configuration, database schema, and data models that power the entire application.
In this step, get ready to:
- Create the configuration file with security settings and constants.
- Set up the SQLite database with three tables for users, URLs, and clicks.
- Define Pydantic models for request and response validation.
Create the configuration file
Before the database module can work, it needs to know where the database file lives. A centralized configuration file keeps all your app's settings in one place so other modules can import what they need.
- Create a new file called app/config.py in your app/ folder and add the following code:
import secrets
SECRET_KEY = secrets.token_hex(32)
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
DATABASE_PATH = "url_shortener.db"
What does this code do?
This file centralizes your application's settings so every other module imports from one place.
- SECRET_KEY generates a random 64-character hex string every time the app starts. This key signs your JWT tokens so nobody can forge them.
- ALGORITHM and ACCESS_TOKEN_EXPIRE_MINUTES control how JWT tokens are created and how long they last.
- DATABASE_PATH tells SQLite where to store the database file on disk.
- Save app/config.py.
- You won't see visible output yet. The next file (database.py) imports from this one.
✔️ Awesome, I've got everything!
Great. Make sure app/config.py is saved before moving on.
ⓧ I'd like to double check the full code
import secrets
SECRET_KEY = secrets.token_hex(32)
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
DATABASE_PATH = "url_shortener.db"
Set up the database module
Your API needs a place to store users, shortened URLs, and click data. SQLite is built into Python, so there is nothing extra to install. You just need a module that connects to the database and creates the tables.
- Create a new file called app/database.py and add the connection helper by pasting this code:
import sqlite3
from app.config import DATABASE_PATH
def get_connection():
conn = sqlite3.connect(DATABASE_PATH)
conn.row_factory = sqlite3.Row
return conn
What does this code do?
get_connection() opens a connection to the SQLite database file defined in your config.
Setting conn.row_factory = sqlite3.Row makes query results behave like dictionaries. Instead of accessing columns by number index (row[0]), you can access them by name (row["username"]). This makes your API code much more readable.
- Still in app/database.py, add the create_tables function below get_connection by pasting this code:
def create_tables():
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
hashed_password TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS urls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
short_code TEXT UNIQUE NOT NULL,
original_url TEXT NOT NULL,
owner_id INTEGER NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (owner_id) REFERENCES users (id)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS clicks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
short_code TEXT NOT NULL,
clicked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
ip_address TEXT,
FOREIGN KEY (short_code) REFERENCES urls (short_code)
)
""")
conn.commit()
conn.close()
What do these three tables store?
- users stores registered accounts with a unique username and a hashed password. Passwords are never stored in plain text.
- urls stores each shortened URL with its unique short code and a foreign key linking it to the user who created it.
- clicks records every time someone visits a short URL. It logs the timestamp and the visitor's IP address for analytics.
CREATE TABLE IF NOT EXISTS means the tables are only created when they don't already exist. This makes the function safe to call multiple times.
- Save app/database.py.
- Verify the database works by running this command in your terminal:
python -c "from app.database import create_tables; create_tables(); print('Tables created!')"
- You should see Tables created! printed in the terminal.
- Confirm the database file now exists by running:
ls url_shortener.db
- You should see url_shortener.db in the output.
Seeing ModuleNotFoundError?
Make sure you are running the command from your project's root directory (the folder that contains the app/ folder). If you are inside the app/ folder itself, Python cannot resolve the from app.config import.
Also confirm your virtual environment is still active. You should see the environment name in your terminal prompt.
Help me debug my ModuleNotFoundError when importing from app.database
✔️ Awesome, I've got everything!
Great. Double check you've saved app/database.py and that url_shortener.db appeared in your project folder.
ⓧ I'd like to double check the full code
import sqlite3
from app.config import DATABASE_PATH
def get_connection():
conn = sqlite3.connect(DATABASE_PATH)
conn.row_factory = sqlite3.Row
return conn
def create_tables():
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
hashed_password TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS urls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
short_code TEXT UNIQUE NOT NULL,
original_url TEXT NOT NULL,
owner_id INTEGER NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (owner_id) REFERENCES users (id)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS clicks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
short_code TEXT NOT NULL,
clicked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
ip_address TEXT,
FOREIGN KEY (short_code) REFERENCES urls (short_code)
)
""")
conn.commit()
conn.close()
Define the Pydantic models
FastAPI uses Pydantic models to validate incoming request data and shape outgoing responses. Each model defines the exact fields and types that an endpoint accepts or returns.
- Create a new file called app/models.py and add the authentication-related models by pasting this code:
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
password: str
class UserResponse(BaseModel):
id: int
username: str
class Token(BaseModel):
access_token: str
token_type: str
What are these models for?
- UserCreate defines what the registration endpoint expects: a username and password.
- UserResponse defines what the API sends back after registration. Notice it does not include the password. You never send passwords back to the client.
- Token defines the shape of the login response: an access token string and its type (always bearer).
- Still in app/models.py, add the URL and analytics models below the Token class by pasting this code:
class URLCreate(BaseModel):
original_url: str
class URLResponse(BaseModel):
short_code: str
original_url: str
created_at: str
class ClickInfo(BaseModel):
clicked_at: str
ip_address: str | None
class URLStats(BaseModel):
short_code: str
original_url: str
total_clicks: int
clicks: list[ClickInfo]
What are these models for?
- URLCreate accepts the original long URL that needs shortening.
- URLResponse returns the generated short code, the original URL, and when it was created.
- ClickInfo represents a single click event with its timestamp and the visitor's IP address (which can be None).
- URLStats bundles a URL's information with its total click count and a list of individual click records.
- Save app/models.py.
- Verify the models load without errors by running:
python -c "from app.models import URLStats; print('Models loaded successfully!')"
- You should see Models loaded successfully! printed in the terminal.
Seeing a syntax error?
The str | None union syntax requires Python 3.10 or higher. Run python --version to confirm you're on 3.10+. If you see an older version, make sure your virtual environment is using the correct Python installation.
Help me fix a syntax error in my Pydantic models file
✔️ Awesome, I've got everything!
Great. Double check you've saved app/models.py and that the import ran without errors.
ⓧ I'd like to double check the full code
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
password: str
class UserResponse(BaseModel):
id: int
username: str
class Token(BaseModel):
access_token: str
token_type: str
class URLCreate(BaseModel):
original_url: str
class URLResponse(BaseModel):
short_code: str
original_url: str
created_at: str
class ClickInfo(BaseModel):
clicked_at: str
ip_address: str | None
class URLStats(BaseModel):
short_code: str
original_url: str
total_clicks: int
clicks: list[ClickInfo]
Your data layer is complete. You have a configuration module, a working database with three tables, and Pydantic models that validate every request and response. Next up, you'll build the authentication system that protects your API routes with JWT tokens.
Build the Authentication System
Your database and models are ready. Now you need a way to identify who is making requests to the API.
In this step, you will build a complete authentication system using JWT tokens and Argon2 password hashing. Users will register, log in to receive a token, and then include that token in future requests to prove who they are.
In this step, get ready to:
- Create the authentication module with password hashing and JWT token handling.
- Build user registration and login endpoints using OAuth2 password flow.
- Test the full authentication flow in FastAPI's interactive docs.
Create the auth module with password hashing and JWT tokens
The auth module is the core of your security system. It handles hashing passwords so they are never stored in plain text, creating JWT tokens for logged-in users, and verifying those tokens on protected routes.
- Create a new file called app/auth.py and add the imports, password hasher, OAuth2 scheme, and utility functions by pasting this code:
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pwdlib import PasswordHash
from app.config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
from app.database import get_connection
# Create a password hasher using the recommended Argon2 algorithm
password_hash = PasswordHash.recommended()
# Tell FastAPI where the login endpoint lives (used by the interactive docs)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
def hash_password(password: str) -> str:
# Convert plain text password into a secure hash
return password_hash.hash(password)
def verify_password(plain_password: str, hashed: str) -> bool:
# Check if a plain password matches the stored hash
return password_hash.verify(plain_password, hashed)
def create_access_token(data: dict) -> str:
# Build a JWT token with an expiration time
to_encode = data.copy()
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
What does this code do?
PasswordHash.recommended() creates an Argon2 hasher. Argon2 is the winner of the Password Hashing Competition and is the current recommended algorithm for storing passwords securely.
OAuth2PasswordBearer(tokenUrl="auth/token") tells FastAPI that the token endpoint lives at /auth/token. This powers the "Authorize" button in the interactive docs.
create_access_token() takes a dictionary (like {"sub": "alice"}), adds an expiration claim, and encodes it into a JWT string using your secret key.
- Now add the get_current_user dependency below create_access_token in the same file:
def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> dict:
# Prepare a reusable 401 error for invalid credentials
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
# Decode the JWT and extract the username from the "sub" claim
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str | None = payload.get("sub")
if username is None:
raise credentials_exception
except jwt.InvalidTokenError:
raise credentials_exception
# Look up the user in the database
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT id, username FROM users WHERE username = ?", (username,))
user = cursor.fetchone()
conn.close()
if user is None:
raise credentials_exception
return {"id": user["id"], "username": user["username"]}
What does get_current_user do?
This function is a FastAPI "dependency." Any route that includes it in its parameters automatically requires a valid JWT token. FastAPI extracts the token from the Authorization: Bearer header, then this function decodes it.
If the token is missing, expired, or the user does not exist in the database, the function raises a 401 error. Otherwise it returns a dictionary with the user's id and username.
- Save app/auth.py.
You will not see any visible change yet. The next substep builds the endpoints that use these functions.
Seeing an import error?
If your editor shows an error on import jwt, make sure your virtual environment is active and that you installed pyjwt in Step 1. The Python package is called pyjwt but the import is jwt.
help me fix an import error with jwt or pwdlib in my auth module
✔️ Awesome, I've got everything!
Great. Double check you have saved app/auth.py before moving on.
ⓧ I'd like to double check the full code
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pwdlib import PasswordHash
from app.config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
from app.database import get_connection
password_hash = PasswordHash.recommended()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
def hash_password(password: str) -> str:
return password_hash.hash(password)
def verify_password(plain_password: str, hashed: str) -> bool:
return password_hash.verify(plain_password, hashed)
def create_access_token(data: dict) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> dict:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str | None = payload.get("sub")
if username is None:
raise credentials_exception
except jwt.InvalidTokenError:
raise credentials_exception
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT id, username FROM users WHERE username = ?", (username,))
user = cursor.fetchone()
conn.close()
if user is None:
raise credentials_exception
return {"id": user["id"], "username": user["username"]}
Create user registration and login endpoints
With the auth utilities ready, you can now build the actual API endpoints that users interact with. These endpoints use OAuth2 password flow, which is the standard way to handle username/password login in REST APIs.
- Create a new file called app/routers/users.py and add the imports, router, and registration endpoint by pasting this code:
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from app.auth import hash_password, verify_password, create_access_token, get_current_user
from app.database import get_connection
from app.models import UserCreate, UserResponse, Token
# Group all auth-related routes under /auth with a descriptive tag
router = APIRouter(prefix="/auth", tags=["Authentication"])
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate):
conn = get_connection()
cursor = conn.cursor()
# Check if the username is already taken
cursor.execute("SELECT id FROM users WHERE username = ?", (user_data.username,))
if cursor.fetchone():
conn.close()
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already taken",
)
# Hash the password and insert the new user
hashed = hash_password(user_data.password)
cursor.execute(
"INSERT INTO users (username, hashed_password) VALUES (?, ?)",
(user_data.username, hashed),
)
conn.commit()
user_id = cursor.lastrowid
conn.close()
return UserResponse(id=user_id, username=user_data.username)
What does this code do?
APIRouter(prefix="/auth", tags=["Authentication"]) creates a router that groups endpoints under the /auth path. This keeps your code organized as the project grows.
The register endpoint first checks whether the username already exists. If not, it hashes the password with Argon2, inserts the user into the database, and returns the new user's ID and username (never the password).
- Add the login and "get current user" endpoints below the register function in the same file:
@router.post("/token", response_model=Token)
def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT id, username, hashed_password FROM users WHERE username = ?",
(form_data.username,),
)
user = cursor.fetchone()
conn.close()
# Reject if user not found or password does not match
if not user or not verify_password(form_data.password, user["hashed_password"]):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
# Issue a JWT with the username in the "sub" claim
access_token = create_access_token(data={"sub": user["username"]})
return Token(access_token=access_token, token_type="bearer")
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: Annotated[dict, Depends(get_current_user)]):
return UserResponse(id=current_user["id"], username=current_user["username"])
What does the login endpoint do?
OAuth2PasswordRequestForm is a FastAPI class that reads username and password from form data. This is required by the OAuth2 password flow spec.
If the credentials match, the endpoint creates a JWT token with the username stored in the sub (subject) claim and returns it. The /auth/me endpoint is a protected route that proves the token works. It uses get_current_user as a dependency to extract and validate the token automatically.
- Save app/routers/users.py.
You will not see output yet because the router has not been mounted on an app. You will test it in the next substep.
Seeing a circular import error?
Make sure your app/__init__.py and app/routers/__init__.py files exist (they can be empty). Without them, Python does not recognize the folders as packages.
help me fix a circular import or ModuleNotFoundError in my FastAPI project
✔️ Awesome, I've got everything!
Great. Double check you have saved app/routers/users.py before moving on.
ⓧ I'd like to double check the full code
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from app.auth import hash_password, verify_password, create_access_token, get_current_user
from app.database import get_connection
from app.models import UserCreate, UserResponse, Token
router = APIRouter(prefix="/auth", tags=["Authentication"])
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: UserCreate):
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT id FROM users WHERE username = ?", (user_data.username,))
if cursor.fetchone():
conn.close()
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already taken",
)
hashed = hash_password(user_data.password)
cursor.execute(
"INSERT INTO users (username, hashed_password) VALUES (?, ?)",
(user_data.username, hashed),
)
conn.commit()
user_id = cursor.lastrowid
conn.close()
return UserResponse(id=user_id, username=user_data.username)
@router.post("/token", response_model=Token)
def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT id, username, hashed_password FROM users WHERE username = ?",
(form_data.username,),
)
user = cursor.fetchone()
conn.close()
if not user or not verify_password(form_data.password, user["hashed_password"]):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(data={"sub": user["username"]})
return Token(access_token=access_token, token_type="bearer")
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: Annotated[dict, Depends(get_current_user)]):
return UserResponse(id=current_user["id"], username=current_user["username"])
Test the authentication flow in the interactive docs
To test these endpoints, you need a temporary FastAPI app that mounts the router and creates the database tables. You will build the final app/main.py in a later step, but for now a quick test file is enough.
- Create a temporary file called test_server.py in the project root (same level as requirements.txt) by pasting this code:
from fastapi import FastAPI
from app.database import create_tables
from app.routers import users
# Temporary app for testing auth endpoints
app = FastAPI()
app.include_router(users.router)
@app.on_event("startup")
def on_startup():
create_tables()
- Start the development server by running this command:
fastapi dev test_server.py
- Open your browser and go to http://127.0.0.1:8000/docs.
- You should see the Swagger UI with three endpoints listed under the **Authentication** section: POST /auth/register, POST /auth/token, and GET /auth/me.
What is the Swagger UI?
FastAPI automatically generates interactive API docs at /docs. You can expand each endpoint, fill in sample data, and click **Execute** to send real requests. It also shows an **Authorize** button for OAuth2 login.
- Expand **POST /auth/register** and click **Try it out**.
- Replace the example JSON in the request body with:
{
"username": "alice",
"password": "secret123"
}
- Click **Execute**.
- You should see a 201 Created response with the user's id and username.
- Now click the **Authorize** button at the top of the page.
- Enter alice in the **username** field and secret123 in the **password** field.
- Click **Authorize**, then **Close**.
- Expand **GET /auth/me** and click **Try it out**, then **Execute**.
- You should see a 200 response with your user's id and username, confirming the token is valid.
Getting a 401 Unauthorized on /auth/me?
Make sure you clicked the **Authorize** button and logged in before trying the /auth/me endpoint. The lock icon next to the endpoint should appear closed (locked) after authorizing.
If you see a 401 on the login itself, double-check that you registered the user first with the exact same username and password.
help me debug a 401 Unauthorized error when testing my JWT auth endpoints
- Stop the development server by pressing Ctrl+C in the terminal.
- Delete the temporary test_server.py file. You will build the real app/main.py in a later step.
Delete the test database too
Running the server created a url_shortener.db file. You can delete it now. A fresh one will be created automatically when you run the final app.
Your authentication system is fully working. Next up, you will build the URL shortening endpoints that use this auth to protect route access.
Build URL Shortening with Background Tasks
Your authentication system is in place. Users can register, log in, and receive JWT tokens. Now it's time to build the core feature: URL shortening.
In this step, you'll create endpoints that let authenticated users shorten URLs and redirect visitors to the original destination. The interesting part is what happens during a redirect. A background task silently records every click without making the visitor wait.
In this step, get ready to:
- Build a protected endpoint that creates short URLs for authenticated users.
- Build a public redirect that uses background tasks to track clicks.
- Build endpoints that list your URLs and display click analytics.
Build POST /urls/ to create a short URL
Every URL shortener needs a way to accept a long URL and return a short code. Your endpoint will require a valid JWT token, generate a random short code, and store the mapping in the database.
- Create a new file called app/routers/urls.py by pasting the following code:
import secrets
from typing import Annotated
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status
from fastapi.responses import RedirectResponse
from app.auth import get_current_user
from app.database import get_connection
from app.models import URLCreate, URLResponse, URLStats, ClickInfo
router = APIRouter(tags=["URLs"])
def generate_short_code() -> str:
# Generate a URL-safe random string (8 characters from 6 bytes)
return secrets.token_urlsafe(6)
def record_click(short_code: str, ip_address: str | None):
# Save click data to the database (called as a background task)
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO clicks (short_code, ip_address) VALUES (?, ?)",
(short_code, ip_address),
)
conn.commit()
conn.close()
@router.post("/urls/", response_model=URLResponse, status_code=status.HTTP_201_CREATED)
def create_short_url(
url_data: URLCreate,
current_user: Annotated[dict, Depends(get_current_user)],
):
short_code = generate_short_code()
# Insert the new short URL linked to the current user
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO urls (short_code, original_url, owner_id) VALUES (?, ?, ?)",
(short_code, url_data.original_url, current_user["id"]),
)
conn.commit()
# Fetch the created_at timestamp for the response
cursor.execute("SELECT created_at FROM urls WHERE short_code = ?", (short_code,))
row = cursor.fetchone()
conn.close()
return URLResponse(
short_code=short_code,
original_url=url_data.original_url,
created_at=row["created_at"],
)
What does this code do?
This file sets up the URL router with two helper functions and the first endpoint.
- generate_short_code() uses Python's secrets module to create a random URL-safe string. Six bytes produces an 8-character code.
- record_click() inserts a row into the clicks table. You'll call this as a background task later.
- create_short_url requires a valid JWT (via the get_current_user dependency), generates a short code, stores it in the database linked to the authenticated user, and returns the new URL.
- Save app/routers/urls.py.
- You won't see a change yet. The redirect endpoint you'll build next uses record_click as a background task.
Build GET /{short_code} to redirect visitors with background click tracking
When someone visits a short URL, they should be instantly redirected to the original destination. At the same time, you want to record that the click happened. BackgroundTasks lets you do both: send the redirect response immediately, then log the click after the response is already on its way.
- Add the following function to the bottom of app/routers/urls.py:
@router.get("/{short_code}")
def redirect_to_url(short_code: str, request: Request, background_tasks: BackgroundTasks):
# Look up the original URL from the database
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT original_url FROM urls WHERE short_code = ?", (short_code,))
row = cursor.fetchone()
conn.close()
if not row:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Short URL not found")
# Record the click in the background (runs after the response is sent)
client_ip = request.client.host if request.client else None
background_tasks.add_task(record_click, short_code, client_ip)
return RedirectResponse(url=row["original_url"], status_code=status.HTTP_307_TEMPORARY_REDIRECT)
What does this code do?
This endpoint is the public-facing redirect. No JWT is required because anyone with the short URL should be able to visit it.
- background_tasks.add_task(record_click, short_code, client_ip) schedules record_click to run after the response is sent. The visitor gets redirected instantly without waiting for the database write.
- RedirectResponse with a 307 status tells the browser to go to the original URL while preserving the HTTP method.
Why BackgroundTasks instead of Celery?
Click logging is a lightweight operation. It takes microseconds to insert one row into SQLite. For tasks this small, FastAPI's built-in BackgroundTasks is the perfect fit.
Celery is designed for heavy tasks that need a message queue and separate worker processes. That would be overkill here and would add infrastructure complexity.
- Save app/routers/urls.py.
- You won't see this in action yet. The stats endpoint in the next section reads the clicks this background task records.
Build endpoints to list URLs and view click stats
Authenticated users need a way to see all the URLs they've created and check how many clicks each one has received. These two endpoints complete the URL router.
- Add the following two endpoints to the bottom of app/routers/urls.py, above the redirect_to_url function:
@router.get("/urls/mine", response_model=list[URLResponse])
def list_my_urls(current_user: Annotated[dict, Depends(get_current_user)]):
# Fetch all URLs belonging to the authenticated user
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT short_code, original_url, created_at FROM urls WHERE owner_id = ?",
(current_user["id"],),
)
rows = cursor.fetchall()
conn.close()
return [
URLResponse(
short_code=row["short_code"],
original_url=row["original_url"],
created_at=row["created_at"],
)
for row in rows
]
@router.get("/urls/{short_code}/stats", response_model=URLStats)
def get_url_stats(
short_code: str,
current_user: Annotated[dict, Depends(get_current_user)],
):
conn = get_connection()
cursor = conn.cursor()
# Verify the URL exists
cursor.execute(
"SELECT short_code, original_url, owner_id FROM urls WHERE short_code = ?",
(short_code,),
)
url_row = cursor.fetchone()
if not url_row:
conn.close()
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="URL not found")
# Only the owner can view stats
if url_row["owner_id"] != current_user["id"]:
conn.close()
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not your URL")
# Fetch all click records for this short code
cursor.execute(
"SELECT clicked_at, ip_address FROM clicks WHERE short_code = ? ORDER BY clicked_at DESC",
(short_code,),
)
click_rows = cursor.fetchall()
conn.close()
clicks = [
ClickInfo(clicked_at=row["clicked_at"], ip_address=row["ip_address"])
for row in click_rows
]
return URLStats(
short_code=url_row["short_code"],
original_url=url_row["original_url"],
total_clicks=len(clicks),
clicks=clicks,
)
What does this code do?
- list_my_urls queries the urls table filtered by the current user's ID. Only your own URLs are returned.
- get_url_stats first checks that the URL exists and belongs to the authenticated user (returning 403 if not). Then it fetches all click records and returns them with a total count.
- Both endpoints require a valid JWT. The get_current_user dependency rejects unauthenticated requests automatically.
Why does route order matter?
The /{short_code} route matches any path segment. If it were defined before /urls/mine, FastAPI would treat mine as a short code and try to redirect. Placing specific routes (/urls/mine, /urls/{short_code}/stats) before the catch-all /{short_code} route prevents this conflict.
- Save app/routers/urls.py.
- Verify your file has no syntax errors by running this command:
python -c "from app.routers.urls import router; print('URL router loaded successfully')"
You should see URL router loaded successfully printed in the terminal.
Seeing an ImportError or SyntaxError?
- Check that your virtual environment is active. The fastapi and pyjwt packages must be importable.
- Make sure app/routers/__init__.py exists as an empty file. Without it, Python cannot import from app.routers.
- Confirm the endpoints are in the correct order: create_short_url, then list_my_urls, then get_url_stats, then redirect_to_url last.
Can't figure out the error? Help me debug this ImportError in my urls.py router
✔️ Awesome, I've got everything!
Great. Double check you've saved app/routers/urls.py and move on.
ⓧ I'd like to double check the full code
import secrets
from typing import Annotated
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status
from fastapi.responses import RedirectResponse
from app.auth import get_current_user
from app.database import get_connection
from app.models import URLCreate, URLResponse, URLStats, ClickInfo
router = APIRouter(tags=["URLs"])
def generate_short_code() -> str:
return secrets.token_urlsafe(6)
def record_click(short_code: str, ip_address: str | None):
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO clicks (short_code, ip_address) VALUES (?, ?)",
(short_code, ip_address),
)
conn.commit()
conn.close()
@router.post("/urls/", response_model=URLResponse, status_code=status.HTTP_201_CREATED)
def create_short_url(
url_data: URLCreate,
current_user: Annotated[dict, Depends(get_current_user)],
):
short_code = generate_short_code()
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO urls (short_code, original_url, owner_id) VALUES (?, ?, ?)",
(short_code, url_data.original_url, current_user["id"]),
)
conn.commit()
cursor.execute("SELECT created_at FROM urls WHERE short_code = ?", (short_code,))
row = cursor.fetchone()
conn.close()
return URLResponse(
short_code=short_code,
original_url=url_data.original_url,
created_at=row["created_at"],
)
@router.get("/urls/mine", response_model=list[URLResponse])
def list_my_urls(current_user: Annotated[dict, Depends(get_current_user)]):
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT short_code, original_url, created_at FROM urls WHERE owner_id = ?",
(current_user["id"],),
)
rows = cursor.fetchall()
conn.close()
return [
URLResponse(
short_code=row["short_code"],
original_url=row["original_url"],
created_at=row["created_at"],
)
for row in rows
]
@router.get("/urls/{short_code}/stats", response_model=URLStats)
def get_url_stats(
short_code: str,
current_user: Annotated[dict, Depends(get_current_user)],
):
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT short_code, original_url, owner_id FROM urls WHERE short_code = ?",
(short_code,),
)
url_row = cursor.fetchone()
if not url_row:
conn.close()
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="URL not found")
if url_row["owner_id"] != current_user["id"]:
conn.close()
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not your URL")
cursor.execute(
"SELECT clicked_at, ip_address FROM clicks WHERE short_code = ? ORDER BY clicked_at DESC",
(short_code,),
)
click_rows = cursor.fetchall()
conn.close()
clicks = [
ClickInfo(clicked_at=row["clicked_at"], ip_address=row["ip_address"])
for row in click_rows
]
return URLStats(
short_code=url_row["short_code"],
original_url=url_row["original_url"],
total_clicks=len(clicks),
clicks=clicks,
)
@router.get("/{short_code}")
def redirect_to_url(short_code: str, request: Request, background_tasks: BackgroundTasks):
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT original_url FROM urls WHERE short_code = ?", (short_code,))
row = cursor.fetchone()
conn.close()
if not row:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Short URL not found")
client_ip = request.client.host if request.client else None
background_tasks.add_task(record_click, short_code, client_ip)
return RedirectResponse(url=row["original_url"], status_code=status.HTTP_307_TEMPORARY_REDIRECT)
Your URL router is complete. Next up, you'll build request logging middleware and wire everything together into the final app.
Add Middleware and Assemble the App
Your routers, authentication, and database modules are all built. Now you need a single entry point that wires everything together into a running application.
In this step, you'll create a custom middleware that logs every request with its processing time. Then you'll build the main application file that assembles all your modules and run the complete API end-to-end.
In this step, get ready to:
- Create a request logging middleware that measures processing time.
- Assemble the full FastAPI application with middleware, CORS, and routers.
- Run the complete API and test the full flow end-to-end.
Create the request logging middleware
Middleware sits between every incoming request and your route handlers. It can inspect, modify, or log requests before they reach your endpoints and after the response is generated.
- Create a new file called middleware.py inside your app/ folder.
- Add the following code to app/middleware.py:
import time
from fastapi import Request
async def log_requests(request: Request, call_next):
# Record the time before processing
start_time = time.perf_counter()
# Pass the request to the next handler
response = await call_next(request)
# Calculate how long it took
process_time = time.perf_counter() - start_time
# Print a log line with method, path, status, and time
print(
f"{request.method} {request.url.path} - "
f"Status: {response.status_code} - "
f"Time: {process_time:.4f}s"
)
# Add processing time to the response headers
response.headers["X-Process-Time"] = f"{process_time:.4f}"
return response
What does this code do?
This middleware function wraps every HTTP request your API receives.
- time.perf_counter() captures a high-precision timestamp before and after the request is processed.
- call_next(request) passes the request along to the actual route handler (or the next middleware in the chain).
- The X-Process-Time header lets API consumers see how long each request took to process.
- Save app/middleware.py.
You won't see a change yet. The middleware needs to be registered in the main app file, which you'll create next.
Assemble the main application file
The main file is where all your modules come together. It creates the FastAPI app instance, registers middleware, attaches routers, and defines startup behaviour.
- Create a new file called main.py inside your app/ folder.
- Add the following code to app/main.py:
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import create_tables
from app.middleware import log_requests
from app.routers import users, urls
# Create the FastAPI app with a title and version
app = FastAPI(title="URL Shortener API", version="1.0.0")
# Register the request logging middleware
app.middleware("http")(log_requests)
# Allow cross-origin requests from any domain
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Attach the authentication and URL routers
app.include_router(users.router)
app.include_router(urls.router)
# Create database tables when the app starts
@app.on_event("startup")
def on_startup():
create_tables()
# Health check endpoint to verify the API is running
@app.get("/", tags=["Health"])
def health_check():
return {"status": "ok", "message": "URL Shortener API is running"}
What does this code do?
- app.middleware("http")(log_requests) registers your custom logging middleware so every request gets timed and logged.
- CORSMiddleware allows requests from any origin, which is important if a frontend ever calls your API from a different domain.
- include_router() connects your authentication and URL routers so their endpoints are accessible.
- The @app.on_event("startup") decorator ensures your database tables are created before the first request arrives.
Why use CORSMiddleware?
CORS (Cross-Origin Resource Sharing) is a browser security feature that blocks requests from different domains by default. Without this middleware, a frontend app hosted on a different URL would be unable to call your API.
- Save app/main.py.
Seeing an import error?
Make sure your app/__init__.py and app/routers/__init__.py files exist (they can be empty). Python needs these files to treat the folders as packages.
Help me debug my import error in app/main.py
✔️ Awesome, I've got everything!
Great. Double check you've saved both app/middleware.py and app/main.py.
ⓧ I'd like to double check the full code
Here's the complete app/middleware.py:
import time
from fastapi import Request
async def log_requests(request: Request, call_next):
start_time = time.perf_counter()
response = await call_next(request)
process_time = time.perf_counter() - start_time
print(
f"{request.method} {request.url.path} - "
f"Status: {response.status_code} - "
f"Time: {process_time:.4f}s"
)
response.headers["X-Process-Time"] = f"{process_time:.4f}"
return response
And the complete app/main.py:
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import create_tables
from app.middleware import log_requests
from app.routers import users, urls
app = FastAPI(title="URL Shortener API", version="1.0.0")
app.middleware("http")(log_requests)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(users.router)
app.include_router(urls.router)
@app.on_event("startup")
def on_startup():
create_tables()
@app.get("/", tags=["Health"])
def health_check():
return {"status": "ok", "message": "URL Shortener API is running"}
Run the complete API and test the full flow
Your entire API is assembled. Time to start the server and test the full authentication-to-analytics flow using FastAPI's built-in interactive docs.
- Start the development server by running this command in your terminal:
fastapi dev app/main.py
- You should see output indicating the server is running at http://127.0.0.1:8000.
- Open your browser and navigate to http://127.0.0.1:8000/docs to open the interactive API documentation.
What should I see?
The Swagger UI shows all your endpoints grouped by tags: Authentication, URLs, and Health. You'll also see an Authorize button at the top for JWT authentication.
Now test the full flow from registration to click tracking.
- Expand the POST /auth/register endpoint.
- Click Try it out.
- Enter a username and password in the request body, then click Execute. You should see a 201 response with your new user ID.
- Click the Authorize button at the top of the page.
- Enter the same username and password you just registered, then click Authorize. This logs you in and stores your JWT token for subsequent requests.
- Expand the POST /urls/ endpoint, click Try it out, and enter a URL like https://www.example.com as the original_url. Click Execute.
- Copy the short_code from the 201 response.
- Open a new browser tab and visit http://127.0.0.1:8000/ followed by the short code you copied. You should be redirected to the original URL.
- Back in the docs, expand GET /urls/{short_code}/stats and enter your short code. You should see a click recorded with a timestamp and IP address.
Check your terminal. You should see log lines from your middleware showing each request with its processing time.
Server won't start?
- Make sure your virtual environment is still active. You should see the environment name in your terminal prompt.
- Check that you're running the command from your project's root folder (the one containing the app/ directory), not from inside app/ itself.
- Verify that both app/__init__.py and app/routers/__init__.py exist as empty files.
Help me debug why fastapi dev app/main.py won't start
Your URL shortener API is fully operational. You've built a production-patterned application with authentication, background tasks, middleware, and a multi-file structure. Next up, you'll take on a bonus challenge to protect your API from abuse.
Secret mission
Add Rate-Limiting Middleware
Your API is open to the world, but what stops a single client from hammering it with thousands of requests? In this secret mission, you'll build a rate-limiting middleware that restricts each IP to 10 requests per minute and returns a proper 429 response when the limit is exceeded.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Project folder on your Desktop (contains app/, requirements.txt, and all source files).
- Python virtual environment (inside the project folder).
- SQLite database file (url_shortener.db in the project root).
- FastAPI development server (running process).
Keep everything running
No action needed. Choose this if you're still actively building or want to extend the project further.
- Your API only runs when you explicitly start it with fastapi dev. There are no background processes or cloud resources consuming money.
- The SQLite database (url_shortener.db) stays on disk but uses negligible space.
- Your virtual environment stays ready for the next time you activate it.
Pause - I'll come back to this later
Shut down running processes to free up memory. Your files stay on disk so you can pick up later.
- Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
- Deactivate your virtual environment by running:
deactivate
When you come back, navigate to your project folder, reactivate the virtual environment, and start the server again.
Delete - I don't want to use this again
Remove all project resources and start fresh. This deletes your code, database, and virtual environment permanently.
- Stop the FastAPI dev server by pressing Ctrl+C in the terminal.
- Deactivate your virtual environment by running:
deactivate
- Navigate to your Desktop and delete the project folder. On macOS/Linux, run:
cd ~/Desktop
rm -rf url-shortener
- On Windows, run:
cd ~/Desktop
rmdir /s /q url-shortener
This removes all source files, the virtual environment, the url_shortener.db database, and installed packages in one go.
Nice Work!
Nice Work!
You just built a production-patterned URL shortener API from scratch using FastAPI. Your API handles user registration, login, URL shortening with redirect tracking, and request logging across a professionally structured multi-file project.
You've learned how to:
- Implement JWT authentication with OAuth2 password flow and Argon2 password hashing to secure your API endpoints.
- Use FastAPI's BackgroundTasks to track click analytics without slowing down URL redirects.
- Build custom middleware for request logging and organize a multi-file project using APIRouter.
- Secret Mission: Added rate-limiting middleware to protect your API from abuse.
Ready to quiz yourself?