Build a URL Shortener API with FastAPI

Build a secure URL shortener with JWT auth, background tasks, and WebSockets.

Introduction

30 Second Summary

You know those compact links you see shared everywhere online? Behind each one is a system that stores the original URL, generates a short code, tracks who clicked it, and often requires you to log in before creating one. That tiny redirect hides real engineering.

In this project, you will build a production-style URL shortener API with FastAPI that authenticates users with JWT tokens, stores data in a cloud PostgreSQL database, tracks clicks using background tasks, and pushes live notifications over WebSockets.

What You'll Build

You will run a single FastAPI server locally, shorten a URL, visit the short link in your browser and watch it redirect instantly, then flip to a live dashboard and see the click notification appear in real time.

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

  • A secure API where you register, log in, and use a JWT token to prove your identity before creating or viewing short links.
  • A working URL shortener that turns any long URL into a short code and redirects visitors instantly while a background task silently records the click.
  • A real-time dashboard powered by WebSockets where you watch click notifications stream in the moment someone visits one of your links.
  • Secret Mission: add configurable link expiration so short URLs automatically stop working after a time limit you set.

Are there any prerequisites?

You should be comfortable with basic Python and have completed a basic CRUD project with FastAPI (or be familiar with HTTP methods and REST concepts). You also need an existing Neon account. The free tier is all you need for this project.

Before We Start

You are about to build a URL shortener API that combines five advanced FastAPI patterns into one cohesive project: JWT authentication, database ORM, dependency injection, background tasks, and WebSockets.

Before any of that can work, you need an isolated Python environment with all dependencies installed and a real cloud PostgreSQL database ready to accept connections. This step builds that foundation so every future step starts from a clean, reproducible baseline.

In this step, get ready to:
  • Create a Python project directory with a virtual environment.
  • Install FastAPI, SQLAlchemy, PyJWT, pwdlib, and other dependencies.
  • Connect to a Neon PostgreSQL database and create your project file structure.
Create the project directory and virtual environment

A virtual environment gives your project its own isolated set of Python packages. This prevents conflicts with other projects on your machine.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), typing Terminal, and pressing Enter.
  • 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
  • Create and activate a virtual environment by running the commands for your operating system:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

Your terminal prompt now shows (venv) at the beginning. This confirms the virtual environment is active and all packages you install next will stay inside this project.

Don't see (venv) in your prompt?

Make sure you ran the activate command (not just the venv creation command). On macOS, use source venv/bin/activate. On Windows, use venv\Scripts\activate.

If you see a permissions error on Windows, try running PowerShell as Administrator and entering Set-ExecutionPolicy RemoteSigned first.

help me activate my Python virtual environment

Install all dependencies

Your project needs six packages. Each one handles a specific job in the API you are about to build.

  • Install the project dependencies by running:
pip install "fastapi[standard]" sqlalchemy psycopg2-binary PyJWT "pwdlib[argon2]" python-dotenv

What does each package do?

  • fastapi[standard] is the web framework with its recommended server (Uvicorn) and extras included.
  • sqlalchemy is the ORM that lets you interact with your database using Python classes instead of raw SQL.
  • psycopg2-binary is the PostgreSQL driver that SQLAlchemy uses under the hood to talk to Neon.
  • PyJWT encodes and decodes JSON Web Tokens for authentication.
  • pwdlib[argon2] hashes passwords securely using the Argon2 algorithm.
  • python-dotenv loads environment variables from a .env file so you can keep secrets out of your code.

You will see download and install progress in your terminal. This may take 30-60 seconds depending on your connection speed.

Before you run the next command, what do you think will happen if one of the packages failed to install?

  • Verify all packages installed correctly by running:
python3 -c "import fastapi; import sqlalchemy; import jwt; import pwdlib; import psycopg2; print('All imports OK')"

You should see All imports OK printed. If any package failed, the import would raise a ModuleNotFoundError naming the missing one.

Seeing ModuleNotFoundError?

Check that your virtual environment is still active (you see (venv) in your prompt). If not, reactivate it and re-run the pip install command.

If the error names psycopg2 specifically, make sure you installed psycopg2-binary (with the -binary suffix). The non-binary version requires C compilation tools.

help me fix a ModuleNotFoundError after installing my Python dependencies

Set up the Neon database and create project files

Neon is a serverless PostgreSQL provider with a generous free tier. It gives your project a real cloud database that scales to zero when idle, so there is no ongoing cost.

  • Open your browser and go to console.neon.tech.
  • Log in to your Neon account.
  • Click New Project (or use an existing project if you prefer).
  • Name the project url-shortener and click Create Project.
  • On your project dashboard, click the Connect button.
  • Copy the connection string. It looks like postgresql://user:password@ep-xxx.region.aws.neon.tech/neondb?sslmode=require.

What is this connection string?

The connection string contains your database username, password, server address, and database name all in one URL. The sslmode=require at the end ensures all traffic between your app and Neon is encrypted.

Keep this string private. Anyone with it can read and write your database.

  • Open VS Code by pressing Cmd+Space (macOS) or the Windows key (Windows), typing VS Code, and pressing Enter.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the url-shortener folder, and click Open.
  • Create a new file called .env by right-clicking in the Explorer panel and selecting New File.
  • Paste the following line into .env, replacing the placeholder with the connection string you just copied from Neon:
DATABASE_URL=postgresql://your-user:your-password@ep-your-endpoint.us-east-2.aws.neon.tech/neondb?sslmode=require
  • Save .env.

Paste your real connection string

Replace everything after DATABASE_URL= with the actual connection string you copied from Neon. The placeholder above shows the format only.

  • Now create the empty project files. Switch to your terminal (still inside url-shortener) and run:

macOS

touch main.py database.py models.py schemas.py auth.py websocket_manager.py

Windows

New-Item -ItemType File -Path "main.py", "database.py", "models.py", "schemas.py", "auth.py", "websocket_manager.py"
  • Confirm the files exist by running:
ls

You should see .env, auth.py, database.py, main.py, models.py, schemas.py, venv, and websocket_manager.py listed.

Missing files in the output?

Make sure your terminal is inside the url-shortener directory. Run pwd (macOS) or Get-Location (Windows) to check your current directory.

If .env is missing from the ls output, it may be hidden. Run ls -a (macOS) or Get-ChildItem -Force (Windows) to show hidden files.

help me find my project files after creating them

Your foundation is set. You have an isolated environment, all your packages installed, a live Neon database connection, and a clean file structure ready for code. Next up, you will build the database models and get your first URL shortened.

Set Up the Database and URL Shortener

You have your Python environment ready and a Neon PostgreSQL database waiting in the cloud. Before layering on authentication or real-time features, you need the core URL shortener working.

In this step, you'll wire up SQLAlchemy to your Neon database, define your data models, and build the two essential endpoints: one to create short URLs and one to redirect visitors. Getting this working first gives you an early visible win and proves the concept with real infrastructure.

In this step, get ready to:
  • Connect SQLAlchemy to your Neon PostgreSQL database with a session dependency.
  • Define database models for users, URLs, and clicks.
  • Build POST /shorten and GET /{short_code} endpoints that create and redirect short links.
Connect SQLAlchemy to Neon PostgreSQL

SQLAlchemy needs three things to connect your app to Neon: an engine (the connection itself), a session factory (creates per-request database sessions), and a dependency function (makes sessions available to your endpoints).

  • Open database.py in VS Code and paste the following code:
import os

from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# Load environment variables from .env file
load_dotenv()

SQLALCHEMY_DATABASE_URL = os.getenv("DATABASE_URL")

# pool_pre_ping=True handles Neon scale-to-zero reconnections
engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


# Base class for all database models
class Base(DeclarativeBase):
    pass


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

What does this code do?

load_dotenv() reads your .env file so os.getenv("DATABASE_URL") can find your Neon connection string.

pool_pre_ping=True tells SQLAlchemy to check if the connection is still alive before using it. Neon scales its compute to zero after inactivity, which drops idle connections. Without this setting, your app would crash with a stale connection error after any pause.

get_db() is a FastAPI dependency. Each request gets a fresh database session, and the finally block ensures it closes even if something goes wrong.

  • Save database.py.
  • Verify the connection is configured by running this command in your terminal:
python3 -c "from database import engine; print('Connected to:', engine.url)"

You should see your Neon connection URL printed with the password masked as asterisks.

Seeing a ModuleNotFoundError?

  • Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
  • Confirm you're running the command from inside the url-shortener folder where database.py lives.
  • Check that .env exists in the same folder and contains your DATABASE_URL value.

Why can't Python find my database module?

✔️ Awesome, I've got everything!

Great. Double check you've saved your file and move on to the next chunk.

ⓧ I'd like to double check the full code

import os

from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

load_dotenv()

SQLALCHEMY_DATABASE_URL = os.getenv("DATABASE_URL")

engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
Define the database models

Models tell SQLAlchemy what tables to create and how they relate to each other. You need three: one for users (used in the next step), one for shortened URLs, and one for tracking clicks.

  • Open models.py and paste the imports and the User model:
from datetime import datetime, timezone

from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from database import Base


# User model - stores registered accounts
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_admin = Column(Boolean, default=False)

    urls = relationship("URL", back_populates="owner")

What does this code do?

Each class that inherits from Base becomes a database table. The __tablename__ string sets the actual table name in PostgreSQL.

The User model stores account credentials. You won't use it until the authentication step, but defining it now lets SQLAlchemy set up the foreign key relationships correctly.

  • Below the User class, add the URL and Click models:
# URL model - stores each shortened link
class URL(Base):
    __tablename__ = "urls"

    id = Column(Integer, primary_key=True, index=True)
    original_url = Column(String, nullable=False)
    short_code = Column(String, unique=True, index=True, nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    owner_id = Column(Integer, ForeignKey("users.id"), nullable=True)

    owner = relationship("User", back_populates="urls")
    clicks = relationship("Click", back_populates="url")


# Click model - records each visit to a short URL
class Click(Base):
    __tablename__ = "clicks"

    id = Column(Integer, primary_key=True, index=True)
    url_id = Column(Integer, ForeignKey("urls.id"), nullable=False)
    clicked_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

    url = relationship("URL", back_populates="clicks")

What does this code do?

The URL model is the heart of the shortener. short_code is the 6-character slug that goes in the short link. owner_id is nullable for now because authentication comes later.

The Click model records every visit. Each click links back to its URL via url_id. The relationship() calls let you navigate between related records in Python code (e.g. url.clicks gives all clicks for a URL).

  • Save models.py.
Define the Pydantic schemas

Pydantic schemas define the shape of data going in and out of your API. They validate incoming requests and format outgoing responses.

  • Open schemas.py and paste the URL-related schemas:
from datetime import datetime

from pydantic import BaseModel, HttpUrl

# --- URL schemas ---

class URLCreate(BaseModel):
    url: HttpUrl


class URLResponse(BaseModel):
    short_code: str
    original_url: str
    created_at: datetime
    click_count: int

    model_config = {"from_attributes": True}


class ClickInfo(BaseModel):
    clicked_at: datetime

    model_config = {"from_attributes": True}


class URLStats(BaseModel):
    short_code: str
    original_url: str
    click_count: int
    clicks: list[ClickInfo]

    model_config = {"from_attributes": True}

What does this code do?

URLCreate validates incoming requests. The HttpUrl type automatically rejects anything that isn't a valid URL.

URLResponse shapes what the API sends back after creating a short link. model_config = {"from_attributes": True} tells Pydantic to read data directly from SQLAlchemy model attributes.

URLStats and ClickInfo are for the analytics endpoint you'll build in a later step.

  • Below the URLStats class, add the auth-related schemas:
# --- Auth schemas (used in the next step) ---

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


class UserResponse(BaseModel):
    id: int
    username: str
    is_admin: bool

    model_config = {"from_attributes": True}


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

What does this code do?

These three schemas support the authentication system you'll build next. UserCreate accepts registration input, UserResponse formats user data for the API response, and Token wraps the JWT token returned at login.

  • Save schemas.py.
  • Verify both files load correctly by running this command:
python3 -c "from models import User, URL, Click; from schemas import URLCreate, URLResponse; print('Models and schemas loaded OK')"

You should see Models and schemas loaded OK printed with no errors.

Seeing an ImportError?

  • Check that database.py is saved. The models import Base from it.
  • Look for typos in class names. Python is case-sensitive, so DeclarativeBase must match exactly.

Why are my model imports failing?

✔️ Awesome, I've got everything!

Great. Confirm both files are saved and move on to building the endpoints.

ⓧ I'd like to double check the full code

Your models.py should contain these sections in order:

  • Imports (datetime, sqlalchemy columns, relationship, Base).
  • User class with username, hashed_password, is_admin columns.
  • URL class with original_url, short_code, created_at, owner_id columns.
  • Click class with url_id and clicked_at columns.

Your schemas.py should contain these sections in order:

  • Imports (datetime, BaseModel, HttpUrl).
  • URLCreate, URLResponse, ClickInfo, URLStats classes.
  • UserCreate, UserResponse, Token classes.
Build the URL shortening endpoints

Now you'll bring everything together in main.py. The FastAPI app uses a lifespan context manager to auto-create your database tables in Neon the moment the server starts.

  • Open main.py and paste the app setup with its lifespan:
import secrets
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session

from database import Base, engine, get_db
from models import Click, URL
from schemas import URLCreate, URLResponse


# Create database tables on startup, then yield control to the app
@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)
    yield


app = FastAPI(title="URL Shortener", lifespan=lifespan)

What does this code do?

The lifespan function runs once when the server starts. Base.metadata.create_all(bind=engine) inspects all your model classes and creates the matching tables in Neon if they don't already exist.

After yield, the app handles requests normally. When the server shuts down, execution resumes after yield for any cleanup (none needed here).

  • Save main.py and start the development server by running:
fastapi dev main.py

You should see output showing the server started at http://127.0.0.1:8000. The lifespan just created your tables in Neon.

Confirm tables in Neon

Switch to your Neon dashboard in the browser, navigate to your project's Tables view, and you should see users, urls, and clicks tables listed. Your cloud database is live.

Server won't start?

  • If you see an SSL or connection error, double-check your DATABASE_URL in .env includes ?sslmode=require at the end.
  • If you see an import error for models or schemas, make sure those files are saved.

Why won't my FastAPI server connect to Neon?

The server is running with auto-reload enabled. Every time you save main.py, it restarts automatically with your changes.

  • Add the POST endpoint for creating short URLs below the app = FastAPI(...) line:
@app.post("/shorten", response_model=URLResponse, status_code=201)
def create_short_url(
    url_data: URLCreate,
    db: Annotated[Session, Depends(get_db)],
):
    # Generate a random 6-character code
    short_code = secrets.token_urlsafe(4)[:6]
    url = URL(
        original_url=str(url_data.url),
        short_code=short_code,
    )
    db.add(url)
    db.commit()
    db.refresh(url)
    # Count existing clicks for the response
    click_count = db.query(Click).filter(Click.url_id == url.id).count()
    return URLResponse(
        short_code=url.short_code,
        original_url=url.original_url,
        created_at=url.created_at,
        click_count=click_count,
    )

What does this code do?

secrets.token_urlsafe(4)[:6] generates a random URL-safe string and trims it to 6 characters. This becomes the short code in your link.

Depends(get_db) is dependency injection. FastAPI calls your get_db() function, gives the session to your endpoint, and closes it automatically when the request finishes.

The endpoint stores the URL in PostgreSQL, then builds a response with the short code and current click count (zero for new links).

  • Save main.py. The server auto-reloads.
  • Open http://127.0.0.1:8000/docs in your browser to see the interactive API docs.
  • Expand the POST /shorten endpoint, click Try it out, enter https://www.example.com as the url value, and click Execute.

You should see a 201 response with a short_code value (a random 6-character string) and click_count: 0.

  • Now add the redirect endpoint below the create_short_url function:
@app.get("/{short_code}")
def redirect_to_url(
    short_code: str,
    db: Annotated[Session, Depends(get_db)],
):
    # Look up the short code in the database
    url = db.query(URL).filter(URL.short_code == short_code).first()
    if not url:
        raise HTTPException(status_code=404, detail="Short URL not found")
    # Redirect the visitor to the original URL
    return RedirectResponse(url=url.original_url, status_code=307)

What does this code do?

When someone visits /{short_code}, the endpoint looks up the code in your database. If found, it returns a RedirectResponse that sends the visitor's browser to the original URL. If not found, it returns a 404 error.

The 307 status code preserves the original HTTP method during the redirect. This is the standard choice for URL shorteners.

  • Save main.py.

Before you test the full flow, what do you expect to happen when you visit your short code URL in the browser?

  • Copy the short_code value from your earlier POST response.
  • Visit http://127.0.0.1:8000/ followed by your short code in your browser (for example, http://127.0.0.1:8000/abc123).

Your browser should redirect you to https://www.example.com. The URL shortener works.

Check your data in Neon

Switch to the Neon console Tables view and select the urls table. You should see a row with your original_url and short_code persisted in the cloud database.

Redirect not working?

  • Make sure you copied the exact short_code from the POST response. Codes are case-sensitive.
  • If you see a 404, the server may have restarted between creating the URL and visiting it. Try creating a new short URL first, then immediately visit it.
  • Check that the GET /{short_code} function is placed after the POST /shorten function in your file.

Why isn't my short URL redirecting?

✔️ Awesome, I've got everything!

Great. Make sure all four files are saved and your server is still running.

ⓧ I'd like to double check the full code

Your main.py should contain these sections in order:

  • Imports (secrets, asynccontextmanager, Annotated, FastAPI, Depends, HTTPException, RedirectResponse, Session, and your local modules).
  • lifespan context manager calling Base.metadata.create_all.
  • app = FastAPI(title="URL Shortener", lifespan=lifespan) line.
  • POST /shorten endpoint generating a random code and storing the URL.
  • GET /{short_code} endpoint looking up the code and returning a RedirectResponse.

Your URL shortener is live and storing data in a real cloud database. Next up, you'll add JWT authentication so users own their links and only they can see their analytics.

Add JWT Authentication

Right now, anyone can hit your POST /shorten endpoint and create URLs with no identity attached.

You need authentication so users own their links and you can enforce who sees whose analytics. Without auth, this is a toy. With it, it is a real API. In this step you will build a full JWT authentication system using Argon2 password hashing and OAuth2 password flow.

In this step, get ready to:
  • Build the auth utility module with password hashing, token creation, and user verification.
  • Add registration, login, and user info endpoints to your API.
  • Protect the URL shortening endpoint so only authenticated users can create links.
Build the auth utilities

All of the authentication logic lives in a single file: auth.py. This module handles password hashing, token creation, and user validation.

  • Open auth.py in VS Code and add the imports, configuration, and password helpers:
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy.orm import Session

from database import get_db
from models import User

# Secret key for signing JWTs - in production, use a real secret
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Set up Argon2 password hashing
password_hash = PasswordHash.recommended()
# Pre-computed hash to prevent timing attacks on login
DUMMY_HASH = password_hash.hash("dummypassword")

# Tells FastAPI where the login endpoint lives
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


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


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

What does this code do?

PasswordHash.recommended() creates a hasher using the Argon2 algorithm. Argon2 is the modern standard for password hashing because it is intentionally slow and memory-hard, making brute-force attacks impractical.

DUMMY_HASH is a pre-computed hash that gets checked when a login attempt uses a username that does not exist. This ensures the response time is the same whether the user exists or not, preventing attackers from discovering valid usernames by measuring timing differences.

OAuth2PasswordBearer tells FastAPI's auto-docs where the login form should POST credentials. It also extracts the Bearer token from the Authorization header on protected requests.

  • Below the password helpers, add the token creation and user authentication functions:
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    # Sign the token with your secret key
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


def authenticate_user(db: Session, username: str, password: str) -> User | None:
    user = db.query(User).filter(User.username == username).first()
    if not user:
        # Still run verify to prevent timing attacks
        verify_password(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

What does this code do?

create_access_token() builds a JWT containing whatever data you pass in (like the username) plus an expiration timestamp. It then signs it with your secret key using HS256.

authenticate_user() looks up the user in the database and verifies their password. If the user does not exist, it still calls verify_password against the dummy hash so the response takes the same amount of time.

  • Confirm the module loads correctly by running this in your terminal:
python3 -c "from auth import create_access_token, authenticate_user, get_password_hash; print('Auth utilities OK')"

You should see Auth utilities OK printed with no errors.

Seeing a ModuleNotFoundError?

  • Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
  • Check that auth.py is saved in the same folder as database.py and models.py.
  • If you see an error about pwdlib, re-run pip install "pwdlib[argon2]" inside your venv.

help me fix my auth module import error

  • Now add the final piece to auth.py below authenticate_user. This is the dependency that protects your endpoints:
async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[Session, Depends(get_db)],
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except InvalidTokenError:
        raise credentials_exception
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception
    return user

What does this code do?

get_current_user() is a FastAPI dependency. Any endpoint that includes it in its parameters will automatically require a valid JWT Bearer token.

It decodes the token, extracts the sub claim (the username), looks up that user in the database, and returns the user object. If anything fails, it raises a 401 Unauthorized error.

  • Save auth.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved auth.py and move on to the next substep.

ⓧ I'd like to double check the full code

Here is the complete auth.py file. Scroll through and confirm your file matches:

  • Chunk 1: imports, SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES, password_hash, DUMMY_HASH, oauth2_scheme, verify_password(), get_password_hash().
  • Chunk 2: create_access_token() and authenticate_user().
  • Chunk 3: get_current_user() dependency.
Build the auth endpoints

With the utility module ready, you can now wire up registration, login, and a protected user-info endpoint in main.py.

  • In main.py, find the import section at the top of the file.
  • Replace the entire import block (everything above the @asynccontextmanager line) with this updated version:
import secrets
from contextlib import asynccontextmanager
from datetime import timedelta
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.responses import RedirectResponse
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from auth import (
    ACCESS_TOKEN_EXPIRE_MINUTES,
    authenticate_user,
    create_access_token,
    get_current_user,
    get_password_hash,
)
from database import Base, engine, get_db
from models import Click, URL, User
from schemas import Token, URLCreate, URLResponse, UserCreate, UserResponse

What changed in the imports?

  • Added timedelta, jwt, status, and OAuth2PasswordRequestForm for auth handling.
  • Imported the auth utilities you just built from auth.py.
  • Added User, Token, UserCreate, and UserResponse to the model and schema imports.
  • Below the app = FastAPI(title="URL Shortener", lifespan=lifespan) line, add the registration endpoint:
# --- Auth Endpoints ---


@app.post("/auth/register", response_model=UserResponse, status_code=201)
def register(user_data: UserCreate, db: Annotated[Session, Depends(get_db)]):
    existing = db.query(User).filter(User.username == user_data.username).first()
    if existing:
        raise HTTPException(status_code=400, detail="Username already registered")
    hashed = get_password_hash(user_data.password)
    user = User(username=user_data.username, hashed_password=hashed)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

What does this code do?

The register endpoint checks if the username already exists, hashes the password using Argon2, creates the user in your Neon database, and returns the user info (without the password).

  • Save main.py and switch to your browser.
  • Open the interactive docs at http://127.0.0.1:8000/docs.
  • Expand POST /auth/register, click Try it out, enter a username and password in the JSON body, then click Execute.

You should see a 201 response with your new user's id, username, and is_admin: false.

Getting a 500 Internal Server Error?

  • Check your terminal for the actual error. If it mentions a connection error, your Neon database might have gone to sleep. The pool_pre_ping=True setting should handle reconnections, but try the request again.
  • If you see a table error, stop the server with Ctrl+C and restart it with fastapi dev main.py to trigger create_all again.

help me debug my register endpoint 500 error

  • Now add the login and user-info endpoints below the register function in main.py:
@app.post("/auth/token", response_model=Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Annotated[Session, Depends(get_db)],
):
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.get("/users/me", response_model=UserResponse)
def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

What does this code do?

The /auth/token endpoint accepts a username and password via the standard OAuth2 form format. If credentials are valid, it returns a signed JWT access token.

The /users/me endpoint uses get_current_user as a dependency. FastAPI automatically extracts the Bearer token from the request, validates it, and injects the user object. If the token is missing or invalid, the endpoint returns a 401 before your code even runs.

  • Save main.py and refresh the docs page at http://127.0.0.1:8000/docs.
  • Expand POST /auth/token, click Try it out, enter the same username and password you registered with, then click Execute.
  • Copy the access_token value from the response (a long encoded string).
  • Click the Authorize button at the top of the docs page (the padlock icon).
  • Paste your token into the Value field and click Authorize.
  • Now expand GET /users/me, click Try it out, then Execute.

You should see a 200 response with your user's id, username, and is_admin fields. Your JWT authentication is working.

Getting 401 Unauthorized on /users/me?

  • Make sure you clicked Authorize and pasted the token before testing /users/me.
  • Copy just the token string from the access_token field in the login response. Do not include the quotes or the token_type field.
  • Tokens expire after 30 minutes. If you have been testing for a while, log in again to get a fresh token.

help me debug my JWT authentication 401 error

Protect the URL endpoints

Your auth system works. Now you need to lock down the URL shortening endpoint so only logged-in users can create links. You will also add an endpoint for users to list their own URLs.

  • In main.py, find your existing create_short_url function.
  • Find this line in the function signature:
def create_short_url(
    url_data: URLCreate,
    db: Annotated[Session, Depends(get_db)],
):
  • Replace it with this version that adds the current_user dependency:
def create_short_url(
    url_data: URLCreate,
    db: Annotated[Session, Depends(get_db)],
    current_user: Annotated[User, Depends(get_current_user)],
):
  • Inside the same function, find the line where you create the URL object:
    url = URL(
        original_url=str(url_data.url),
        short_code=short_code,
    )
  • Replace it with this version that sets the owner:
    url = URL(
        original_url=str(url_data.url),
        short_code=short_code,
        owner_id=current_user.id,
    )
  • Now add this new endpoint below create_short_url and above the redirect_to_url function. This lets users list their own URLs:
@app.get("/urls/me", response_model=list[URLResponse])
def get_my_urls(
    db: Annotated[Session, Depends(get_db)],
    current_user: Annotated[User, Depends(get_current_user)],
):
    urls = db.query(URL).filter(URL.owner_id == current_user.id).all()
    results = []
    for url in urls:
        click_count = db.query(Click).filter(Click.url_id == url.id).count()
        results.append(
            URLResponse(
                short_code=url.short_code,
                original_url=url.original_url,
                created_at=url.created_at,
                click_count=click_count,
            )
        )
    return results

What does this code do?

GET /urls/me queries the database for all URLs where owner_id matches the authenticated user. It builds a list of URLResponse objects that include the click count for each link.

Because get_current_user is a parameter dependency, FastAPI enforces the Bearer token requirement automatically. No token means no access.

  • Save main.py.

Before you test, what do you think will happen if you try POST /shorten without clicking Authorize first?

  • Refresh the docs page at http://127.0.0.1:8000/docs.
  • Click the Logout button in the Authorize dialog to clear your token.
  • Expand POST /shorten, click Try it out, enter any URL in the body, and click Execute.

You get a 401 Unauthorized response. The endpoint now refuses unauthenticated requests.

  • Log in again: expand POST /auth/token, enter your credentials, copy the new access_token, and click Authorize to paste it.
  • Try POST /shorten again with a URL. This time you should get a 201 response with your new short code.
  • Expand GET /urls/me and execute it. You should see a list containing the URL you just created.

Still getting 401 after authorizing?

  • Confirm the tokenUrl in auth.py matches your login path exactly: /auth/token.
  • Check that get_current_user is imported in main.py from auth, not redefined locally.
  • Restart the server if your terminal shows import errors after saving.

help me debug why my protected endpoints still return 401

✔️ Awesome, I've got everything!

Great. Confirm main.py is saved and your auth flow is working end-to-end.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point. Scroll through and confirm your file matches:

import secrets
from contextlib import asynccontextmanager
from datetime import timedelta
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.responses import RedirectResponse
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from auth import (
    ACCESS_TOKEN_EXPIRE_MINUTES,
    authenticate_user,
    create_access_token,
    get_current_user,
    get_password_hash,
)
from database import Base, engine, get_db
from models import Click, URL, User
from schemas import Token, URLCreate, URLResponse, UserCreate, UserResponse
@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)
    yield


app = FastAPI(title="URL Shortener", lifespan=lifespan)


# --- Auth Endpoints ---


@app.post("/auth/register", response_model=UserResponse, status_code=201)
def register(user_data: UserCreate, db: Annotated[Session, Depends(get_db)]):
    existing = db.query(User).filter(User.username == user_data.username).first()
    if existing:
        raise HTTPException(status_code=400, detail="Username already registered")
    hashed = get_password_hash(user_data.password)
    user = User(username=user_data.username, hashed_password=hashed)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user
@app.post("/auth/token", response_model=Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Annotated[Session, Depends(get_db)],
):
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.get("/users/me", response_model=UserResponse)
def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user
# --- URL Endpoints ---


@app.post("/shorten", response_model=URLResponse, status_code=201)
def create_short_url(
    url_data: URLCreate,
    db: Annotated[Session, Depends(get_db)],
    current_user: Annotated[User, Depends(get_current_user)],
):
    short_code = secrets.token_urlsafe(4)[:6]
    url = URL(
        original_url=str(url_data.url),
        short_code=short_code,
        owner_id=current_user.id,
    )
    db.add(url)
    db.commit()
    db.refresh(url)
    click_count = db.query(Click).filter(Click.url_id == url.id).count()
    return URLResponse(
        short_code=url.short_code,
        original_url=url.original_url,
        created_at=url.created_at,
        click_count=click_count,
    )
@app.get("/urls/me", response_model=list[URLResponse])
def get_my_urls(
    db: Annotated[Session, Depends(get_db)],
    current_user: Annotated[User, Depends(get_current_user)],
):
    urls = db.query(URL).filter(URL.owner_id == current_user.id).all()
    results = []
    for url in urls:
        click_count = db.query(Click).filter(Click.url_id == url.id).count()
        results.append(
            URLResponse(
                short_code=url.short_code,
                original_url=url.original_url,
                created_at=url.created_at,
                click_count=click_count,
            )
        )
    return results


@app.get("/{short_code}")
def redirect_to_url(
    short_code: str,
    db: Annotated[Session, Depends(get_db)],
):
    url = db.query(URL).filter(URL.short_code == short_code).first()
    if not url:
        raise HTTPException(status_code=404, detail="Short URL not found")
    return RedirectResponse(url=url.original_url, status_code=307)

Your API now has full JWT authentication. Users register, log in to receive a token, and use that token to create and manage their own short URLs. Next up, you will add background task processing so click tracking does not slow down redirects.

Track Clicks with Background Tasks

Your API is now protected. Users register, authenticate, and create links that belong to them.

There is a gap, though. When someone visits a short URL, the redirect fires but nobody knows it happened. You need click analytics. The naive approach is writing a database row inside the redirect handler before responding. That slows down every single redirect. Background tasks solve this by deferring the database write until after the response is sent.

In this step, get ready to:
  • Create a background task function that records each click in the database.
  • Wire the background task into the redirect endpoint so redirects stay fast.
  • Add a protected stats endpoint that returns click counts and timestamps for a URL.
Create the click-tracking background task

A background task is a regular Python function that FastAPI runs after it has already sent the HTTP response back to the client. The redirect happens instantly, and the database write trails behind.

  • In main.py, find the imports at the top of the file. Update the datetime import line to include datetime and timezone:
from datetime import timedelta
  • Replace it with:
from datetime import datetime, timedelta, timezone
  • Next, find the fastapi import line:
from fastapi import Depends, FastAPI, HTTPException, status
  • Replace it with this expanded import that adds BackgroundTasks:
from fastapi import (
    BackgroundTasks,
    Depends,
    FastAPI,
    HTTPException,
    status,
)
  • Now add the record_click function. Place it in main.py above the existing @app.get("/{short_code}") handler by pasting the following code:
def record_click(url_id: int, owner_id: int | None, short_code: str):
    # Import SessionLocal inside the function to avoid circular imports
    from database import SessionLocal

    db = SessionLocal()
    try:
        # Create a new Click record tied to this URL
        click = Click(url_id=url_id)
        db.add(click)
        db.commit()
        print(f"[Background] Recorded click for /{short_code}")
    finally:
        db.close()

What does this code do?

This function runs outside the request lifecycle. It opens its own database session, inserts a Click row, commits, and prints a confirmation.

It imports SessionLocal inside the function body because background tasks run after the request's session is already closed. Each background task needs its own independent session.

  • Save main.py.
  • Check the terminal where your FastAPI dev server is running. It should reload without any import errors.

Seeing an import error on reload?

Make sure BackgroundTasks is inside the from fastapi import (...) block with a comma after it. Also confirm the datetime import line now reads from datetime import datetime, timedelta, timezone.

help me debug an import error in my FastAPI app after adding BackgroundTasks

Wire the background task into the redirect handler

The function exists but nothing calls it yet. You need to tell FastAPI to run it after each redirect response.

  • In main.py, find the existing redirect handler:
@app.get("/{short_code}")
def redirect_to_url(
    short_code: str,
    db: Annotated[Session, Depends(get_db)],
):
    url = db.query(URL).filter(URL.short_code == short_code).first()
    if not url:
        raise HTTPException(status_code=404, detail="Short URL not found")
    return RedirectResponse(url=url.original_url, status_code=307)
  • Replace the entire handler with this updated version that adds a background_tasks parameter and schedules the click recording:
@app.get("/{short_code}")
def redirect_to_url(
    short_code: str,
    background_tasks: BackgroundTasks,
    db: Annotated[Session, Depends(get_db)],
):
    url = db.query(URL).filter(URL.short_code == short_code).first()
    if not url:
        raise HTTPException(status_code=404, detail="Short URL not found")
    # Schedule the click recording to run AFTER the response is sent
    background_tasks.add_task(
        record_click, url_id=url.id, owner_id=url.owner_id, short_code=url.short_code
    )
    return RedirectResponse(url=url.original_url, status_code=307)

What does this code do?

Declaring background_tasks: BackgroundTasks as a parameter tells FastAPI to inject its background-task scheduler. You do not import or create it yourself.

Calling background_tasks.add_task(record_click, ...) queues the function to run after the response. The user gets their redirect immediately. The database write happens a fraction of a second later, invisible to them.

  • Save main.py.
  • Open your browser and visit one of your short URLs (e.g. http://127.0.0.1:8000/ followed by a short code you created earlier). You should be redirected to the original URL.
  • Switch back to the terminal running your FastAPI dev server. You should see a line that reads [Background] Recorded click for / followed by your short code.

Why does the log appear after the redirect?

FastAPI sends the 307 redirect response first, then executes any queued background tasks. The print() inside record_click fires after your browser has already followed the redirect. That is the entire point of background tasks.

Not seeing the background log message?

Make sure you are visiting an actual short code that exists in your database. If the URL returns a 404, the background task is never scheduled.

Also confirm that record_click is defined above the @app.get("/{short_code}") handler in the file. Python needs to see the function before it is referenced.

help me figure out why my FastAPI background task is not running after a redirect

Add an analytics endpoint

Clicks are being recorded in the background. Now you need an endpoint where authenticated users can check how many clicks a URL has received and when each click happened.

  • First, update the schemas import at the top of main.py. Find this line:
from schemas import Token, URLCreate, URLResponse, UserCreate, UserResponse
  • Replace it with this version that adds URLStats:
from schemas import Token, URLCreate, URLResponse, URLStats, UserCreate, UserResponse
  • Now add the stats endpoint at the bottom of main.py, after the redirect handler:
@app.get("/urls/{short_code}/stats", response_model=URLStats)
def get_url_stats(
    short_code: str,
    db: Annotated[Session, Depends(get_db)],
    current_user: Annotated[User, Depends(get_current_user)],
):
    url = db.query(URL).filter(URL.short_code == short_code).first()
    if not url:
        raise HTTPException(status_code=404, detail="Short URL not found")
    # Only the owner or an admin can view stats
    if url.owner_id != current_user.id and not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Not authorized to view these stats")
    clicks = db.query(Click).filter(Click.url_id == url.id).all()
    return URLStats(
        short_code=url.short_code,
        original_url=url.original_url,
        click_count=len(clicks),
        clicks=clicks,
    )

What does this code do?

This endpoint requires authentication (via get_current_user) and then checks ownership. Only the person who created the URL (or an admin) can see its click data.

It queries all Click records for that URL and returns both the total count and a list of individual click timestamps. This is the analytics payoff of the background task work.

  • Save main.py.

Before you test the full loop, what do you think will happen when you call the stats endpoint for one of your short codes? Will the click count match the number of times you visited the short URL?

  • Open the interactive docs at http://127.0.0.1:8000/docs.
  • Click **Authorize** and paste your JWT token (log in via POST /auth/token if your previous token expired).
  • Expand GET /urls/{short_code}/stats and enter one of your short codes.
  • Click **Execute**. You should see a JSON response with a click_count of at least 1, plus a clicks array containing the timestamp of each visit.

Seeing a click count of 0?

Make sure you visited the short URL in a separate browser tab (not through the Swagger docs) after adding the background task code. Only visits to http://127.0.0.1:8000/{short_code} trigger the redirect and recording logic.

If you get a 403, confirm you are checking the stats for a URL you own. The endpoint only allows the URL's creator to view its analytics.

help me debug why my click count is zero even though I visited my short URL

You can also confirm the full loop by calling GET /urls/me in the Swagger docs. The click_count field on your URL should match the number of times you visited it.

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Confirm your main.py contains the following sections in order:

  • Updated imports including BackgroundTasks, datetime, timezone, and URLStats.
  • The lifespan context manager and app = FastAPI(...) line.
  • Auth endpoints: POST /auth/register, POST /auth/token, GET /users/me.
  • URL endpoints: POST /shorten and GET /urls/me.
  • The record_click function (defined above the redirect handler).
  • The updated GET /{short_code} redirect handler with background_tasks parameter.
  • The new GET /urls/{short_code}/stats endpoint at the bottom.

Your redirects now track every click without making users wait. Next up, you will push those click events to connected users in real time using WebSockets.

Add WebSocket Real-time Notifications

Right now, the only way to check if someone clicked your short URL is to call the stats endpoint manually. That means polling repeatedly, which wastes bandwidth and still leaves you waiting between requests.

WebSockets solve this by opening a persistent connection between the server and your browser. Instead of you asking "any new clicks?" over and over, the server pushes a notification to you the instant a click happens.

In this step, get ready to:
  • Build a ConnectionManager that tracks which users have active WebSocket connections.
  • Create a WebSocket endpoint and HTML dashboard for receiving live notifications.
  • Connect the background click-tracking task to push notifications over WebSocket.
Build the ConnectionManager

You need a central registry that knows which users are connected and where to send their notifications. The ConnectionManager class handles this by mapping each user ID to a list of active WebSocket connections.

  • Open websocket_manager.py in VS Code and add the following code:
from fastapi import WebSocket


class ConnectionManager:
    def __init__(self):
        # Map user IDs to their active WebSocket connections
        self.active_connections: dict[int, list[WebSocket]] = {}

    async def connect(self, user_id: int, websocket: WebSocket):
        await websocket.accept()
        if user_id not in self.active_connections:
            self.active_connections[user_id] = []
        self.active_connections[user_id].append(websocket)

    def disconnect(self, user_id: int, websocket: WebSocket):
        if user_id in self.active_connections:
            self.active_connections[user_id].remove(websocket)
            if not self.active_connections[user_id]:
                del self.active_connections[user_id]

    async def notify_user(self, user_id: int, message: dict):
        if user_id in self.active_connections:
            for connection in self.active_connections[user_id]:
                await connection.send_json(message)


manager = ConnectionManager()

What does this code do?

  • active_connections is a dictionary keyed by user ID. Each user can have multiple open tabs, so the value is a list of WebSocket objects.
  • connect() accepts the WebSocket handshake and registers the connection under the user's ID.
  • disconnect() removes a specific connection and cleans up the entry if no connections remain.
  • notify_user() sends a JSON message to every active connection belonging to a specific user.
  • The module-level manager instance is a singleton that the rest of the app imports.
  • Save websocket_manager.py.
  • Check your terminal where the dev server is running. It should restart without errors.

Seeing an import error?

  • Make sure the file is named exactly websocket_manager.py (with an underscore) and saved in the same url-shortener folder as main.py.
  • Confirm the from fastapi import WebSocket line is at the top of the file.

Help me debug an import error in websocket_manager.py

Create the WebSocket endpoint and dashboard

Browsers cannot send custom HTTP headers (like Authorization: Bearer ...) when opening a WebSocket connection. The standard workaround is to pass the JWT token as a query parameter in the URL.

You need to update the imports at the top of main.py to bring in WebSocket support and the manager.

  • Replace the entire import section at the top of main.py (everything before the @asynccontextmanager line) with:
import secrets
from contextlib import asynccontextmanager
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import (
    BackgroundTasks,
    Depends,
    FastAPI,
    HTTPException,
    Query,
    WebSocket,
    WebSocketDisconnect,
    status,
)
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.security import OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from sqlalchemy.orm import Session

from auth import (
    ACCESS_TOKEN_EXPIRE_MINUTES,
    ALGORITHM,
    SECRET_KEY,
    authenticate_user,
    create_access_token,
    get_current_user,
    get_password_hash,
)
from database import Base, engine, get_db
from models import Click, URL, User
from schemas import Token, URLCreate, URLResponse, URLStats, UserCreate, UserResponse
from websocket_manager import manager

What changed in the imports?

  • Query, WebSocket, and WebSocketDisconnect are new FastAPI imports for handling WebSocket connections.
  • HTMLResponse lets you serve the dashboard page directly from Python.
  • InvalidTokenError from jwt.exceptions handles malformed JWT tokens in the WebSocket endpoint.
  • ALGORITHM and SECRET_KEY from auth are needed to decode the JWT token directly in the WebSocket handler.
  • manager from websocket_manager is the singleton ConnectionManager instance.
  • Save main.py and confirm the server restarts without errors in your terminal.

Now add the WebSocket endpoint that authenticates users via their JWT token and holds the connection open.

  • In main.py, add the following function after the get_url_stats endpoint:
@app.websocket("/ws/notifications")
async def websocket_notifications(
    websocket: WebSocket, token: Annotated[str | None, Query()] = None
):
    # Reject connections with no token
    if not token:
        await websocket.close(code=1008)
        return
    # Validate the JWT token from the query parameter
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            await websocket.close(code=1008)
            return
    except InvalidTokenError:
        await websocket.close(code=1008)
        return

    # Look up the user in the database
    db = next(get_db())
    user = db.query(User).filter(User.username == username).first()
    db.close()
    if not user:
        await websocket.close(code=1008)
        return

    # Register the connection and hold it open
    await manager.connect(user.id, websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        manager.disconnect(user.id, websocket)

What does this endpoint do?

The endpoint reads the JWT from the ?token= query parameter (since browsers cannot send headers on WebSocket connections). If the token is missing or invalid, it closes the connection with code 1008 (policy violation).

Once authenticated, the connection enters a receive loop. This keeps the WebSocket open until the client disconnects. When the client closes their tab, WebSocketDisconnect fires and the manager cleans up.

Next, add the HTML dashboard that gives you a browser-based interface to test the WebSocket connection.

  • Add the following dashboard template and endpoint at the bottom of main.py, after the WebSocket function:
dashboard_html = """
<!DOCTYPE html>
<html>
<head><title>Click Notifications</title></head>
<body>
    <h1>Live Click Notifications</h1>
    <p>Enter your JWT token to connect:</p>
    <input type="text" id="token" placeholder="Paste your access_token here" style="width:400px"/>
    <button onclick="connect()">Connect</button>
    <h2>Events:</h2>
    <ul id="events"></ul>
    <script>
        var ws = null;
        function connect() {
            var token = document.getElementById("token").value;
            ws = new WebSocket("ws://localhost:8000/ws/notifications?token=" + token);
            ws.onmessage = function(event) {
                var events = document.getElementById("events");
                var li = document.createElement("li");
                li.textContent = event.data;
                events.appendChild(li);
            };
            ws.onopen = function() {
                var events = document.getElementById("events");
                var li = document.createElement("li");
                li.textContent = "Connected! Waiting for clicks...";
                li.style.color = "green";
                events.appendChild(li);
            };
            ws.onclose = function() {
                var events = document.getElementById("events");
                var li = document.createElement("li");
                li.textContent = "Disconnected.";
                li.style.color = "red";
                events.appendChild(li);
            };
        }
    </script>
</body>
</html>
"""


@app.get("/dashboard", response_class=HTMLResponse)
def get_dashboard():
    return HTMLResponse(content=dashboard_html)

What does the dashboard do?

This is a minimal HTML page served directly from your FastAPI app. It has a text field for pasting your JWT token and a Connect button.

The JavaScript creates a WebSocket connection to /ws/notifications?token=... and listens for messages. When a message arrives, it appends a new list item to the page. Green text means connected, red means disconnected.

  • Save main.py.
  • Open your browser and visit http://127.0.0.1:8000/dashboard.
  • You should see a page titled "Live Click Notifications" with a text input and a Connect button.

Page not loading?

  • Make sure HTMLResponse is imported from fastapi.responses in your imports section.
  • Check that the dashboard_html variable and get_dashboard function are placed after the WebSocket endpoint, not inside it.
  • Confirm the dev server restarted successfully (look for "Application startup complete" in the terminal).

Help me debug why my /dashboard page isn't loading

Connect background tasks to WebSocket notifications

The dashboard page is ready but it will never receive notifications yet. The record_click background task only writes to the database. You need to also push a notification to the link owner's WebSocket connection after each click.

  • In main.py, find the record_click function. Replace the entire function with:
def record_click(url_id: int, owner_id: int | None, short_code: str):
    from database import SessionLocal

    db = SessionLocal()
    try:
        click = Click(url_id=url_id)
        db.add(click)
        db.commit()
        print(f"[Background] Recorded click for /{short_code}")

        import asyncio

        # Push a live notification to the link owner
        if owner_id:
            notification = {
                "event": "click",
                "short_code": short_code,
                "clicked_at": datetime.now(timezone.utc).isoformat(),
            }
            loop = asyncio.new_event_loop()
            loop.run_until_complete(manager.notify_user(owner_id, notification))
            loop.close()
    finally:
        db.close()

Why create a new event loop?

The record_click function runs as a synchronous background task (it is a regular def, not async def). But manager.notify_user() is an async method that sends WebSocket messages.

To call async code from synchronous code, you create a temporary event loop with asyncio.new_event_loop() and run the coroutine inside it. The loop is closed immediately after to prevent resource leaks.

  • Save main.py.

Before you test the full loop, what do you think will happen when you visit a short URL while the dashboard is connected?

  • Open a browser tab and go to http://127.0.0.1:8000/dashboard.
  • You need a fresh JWT token. In another tab, open http://127.0.0.1:8000/docs and call POST /auth/token with your username and password.
  • Copy the access_token value from the response.
  • Switch back to the dashboard tab, paste the token into the input field, and click Connect.
  • You should see a green "Connected! Waiting for clicks..." message.
  • Now in a third tab, visit one of your short URLs (e.g. http://127.0.0.1:8000/ followed by one of your short codes).
  • Switch back to the dashboard tab. A new list item should appear showing the click event with the short code and timestamp.

The full loop is now working. A click triggers a redirect, the background task records it in the database AND pushes a live notification to the dashboard over WebSocket.

Not seeing notifications in the dashboard?

  • Make sure you see the green "Connected!" message before clicking a short URL. If it says "Disconnected" immediately, your token may be expired. Generate a fresh one from POST /auth/token.
  • Confirm the short URL you visit belongs to the same user whose token you pasted. Notifications only go to the link owner.
  • Check the terminal for the [Background] Recorded click log line. If you see it, the background task ran. If no notification appeared, the owner_id might be None (the URL was created before auth was added).

Help me debug WebSocket notifications not appearing in my dashboard

✔️ Awesome, I've got everything!

Great. Double check you have saved both websocket_manager.py and main.py.

ⓧ I'd like to double check the full code

Here is the complete websocket_manager.py:

from fastapi import WebSocket


class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[int, list[WebSocket]] = {}

    async def connect(self, user_id: int, websocket: WebSocket):
        await websocket.accept()
        if user_id not in self.active_connections:
            self.active_connections[user_id] = []
        self.active_connections[user_id].append(websocket)

    def disconnect(self, user_id: int, websocket: WebSocket):
        if user_id in self.active_connections:
            self.active_connections[user_id].remove(websocket)
            if not self.active_connections[user_id]:
                del self.active_connections[user_id]

    async def notify_user(self, user_id: int, message: dict):
        if user_id in self.active_connections:
            for connection in self.active_connections[user_id]:
                await connection.send_json(message)


manager = ConnectionManager()

For main.py, confirm these sections are present in order by scrolling through your file:

  • Updated imports (including Query, WebSocket, WebSocketDisconnect, HTMLResponse, InvalidTokenError, ALGORITHM, SECRET_KEY, manager).
  • The lifespan, app creation, and all existing auth/URL endpoints (unchanged).
  • Updated record_click function with the asyncio notification block after the print statement.
  • The redirect_to_url and get_url_stats endpoints (unchanged).
  • New websocket_notifications WebSocket endpoint.
  • New dashboard_html string and get_dashboard endpoint at the bottom.

Your API now delivers real-time click notifications over WebSocket. Next up, you will tackle the Secret Mission and add link expiration to make your URL shortener even more production-ready.

Secret mission

Add Link Expiration

Real URL shorteners let users set expiration times on their links. Can you add a configurable time limit so short URLs automatically stop working and return a 410 Gone response after they expire?

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 on Neon's free tier, which scales to zero automatically when idle, so there are no ongoing costs.

Resources you used:

  • Neon PostgreSQL database project "url-shortener" (free tier, no ongoing cost).
  • Local url-shortener project folder (includes virtual environment, source files, and .env with your connection string).

Keep everything running

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

  • Your Neon database stays on the free tier and scales to zero automatically when idle.
  • Your local project files remain in the url-shortener folder on your Desktop.
  • To resume development later, activate your virtual environment and run fastapi dev main.py again.

Pause - I'll come back to this later

Shut down the running server and deactivate your virtual environment to free up system resources. Your files and database stay intact.

  • Press Ctrl+C in the terminal running your FastAPI dev server to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • Neon automatically suspends compute after inactivity. No action needed on the database side.
  • To resume later, navigate back to your url-shortener folder, run source venv/bin/activate, and start the server again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your data.

  • Press Ctrl+C in the terminal running your FastAPI dev server to stop it.
  • Deactivate your virtual environment by running:
deactivate

Delete your local project folder by running:

cd ~/Desktop
rm -rf url-shortener

On Windows, use this instead:

cd ~\Desktop
rmdir /s /q url-shortener

Delete the Neon database project:

  • Go to console.neon.tech and open your "url-shortener" project.
  • Click Project Settings in the sidebar.
  • Scroll down and click Delete project to permanently remove all data.

Nice Work!

Nice Work!

You just built a production-style URL shortener API that combines five advanced FastAPI patterns into one working system backed by a real cloud database.

You've learned how to:

  • Implement JWT authentication with OAuth2 password flow, Argon2 password hashing, and protected endpoints that reject unauthenticated requests with 401 responses.
  • Use FastAPI background tasks to record click analytics without blocking the redirect response, keeping short URL redirects instant for visitors.
  • Build a WebSocket notification system with a ConnectionManager pattern that pushes live click events to authenticated users the moment someone visits their link.
  • Secret Mission: Added configurable link expiration with proper 410 Gone handling so short URLs automatically stop working after a set time limit.

Ready to quiz yourself?