Production URL Shortener in Python

Build a FastAPI URL shortener with PostgreSQL, Redis caching, and click analytics.

Introduction

30 Second Summary

Every time you share a link on social media or in a message, you've probably used a URL shortener. Behind that simple redirect is a real backend system handling database writes, caching layers, and analytics pipelines.

In this project, you will build a production-grade URL shortening API with a real-time click analytics engine. You'll wire together FastAPI, PostgreSQL, Redis, and Docker Compose into a system you can run with one command.

What You'll Build

A complete REST API that generates unique short codes for URLs, redirects visitors, records click events with metadata, and serves aggregated analytics through cached endpoints.

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

  • A FastAPI REST API with endpoints for creating, resolving, and analyzing short URLs.
  • An async PostgreSQL database layer with SQLAlchemy ORM and Alembic migrations, plus a Redis caching layer for fast redirects.
  • A comprehensive pytest test suite and a Docker Compose stack that runs the entire system with one command.
  • Secret Mission: add rate limiting and API key authentication to protect your endpoints.

Are there any prerequisites?

You'll need basic Python knowledge (variables, functions, classes, loops, imports). All other tools are installed during the setup steps.

Before We Start

Every URL shortener needs a solid foundation: a package manager, a project structure, and pinned dependencies that guarantee the same behavior on every machine.

In this step, you will install uv (a modern Python package manager), initialize a new project called snip, and lock every dependency to exact versions.

In this step, get ready to:
  • Install uv, the modern Python package manager.
  • Initialize a new Python project called snip and configure its pyproject.toml.
  • Install all project dependencies with pinned versions.
Install uv

uv replaces pip, virtualenv, and poetry with a single tool that is 10-100x faster.

  • Open a terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Check whether uv is already installed by running this command:
uv --version

✔️ I see a version number

You already have uv installed. Update it to the latest version by running:

uv self update

You are ready to move on to the next substep.

ⓧ Command not found

uv is not installed yet. Install it using the official standalone installer for your operating system.

macOS

  • Install uv by running this command:
curl -LsSf https://astral.sh/uv/install.sh | sh
  • Close and reopen your terminal so the PATH changes take effect.
  • Verify the installation by running:
uv --version

You should see a version number printed in the terminal.

Windows

  • Install uv by running this command in PowerShell:
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
  • Close and reopen PowerShell so the PATH changes take effect.
  • Verify the installation by running:
uv --version

You should see a version number printed in the terminal.

Still seeing command not found?

Make sure you fully closed and reopened your terminal after the install script completed. The installer adds uv to your PATH, but existing terminal sessions do not pick up the change automatically.

help me troubleshoot why uv is not found after installing it

Create the project with uv init

uv can scaffold a new Python project with a single command. It generates a pyproject.toml file, a .python-version file, and a virtual environment.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create a new project called snip and move into it by running these commands:
uv init snip
cd snip

What does uv init do?

uv init creates a new directory with a pyproject.toml (project metadata and dependencies), a .python-version file (pins the Python version), and a starter main.py.

You will replace the generated pyproject.toml with a production-ready configuration in the next step.

  • Confirm the project folder was created by running:
ls

You should see pyproject.toml, main.py, and .python-version listed in the output.

Seeing a Python version error?

If uv init reports that it cannot find Python 3.12, run uv python install 3.12 to let uv download and manage it for you. Then retry uv init snip.

help me fix a Python version error when running uv init

Now you will replace the default pyproject.toml with the full project configuration that pins every dependency.

  • Open the snip folder in your code editor. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Cursor and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the snip folder, then click Open.
  • In the Explorer sidebar, click on pyproject.toml to open it.
  • Select all the existing content and delete it.
  • Replace it with the following configuration:
[project]
name = "snip"
version = "0.1.0"
description = "Production URL Shortener & Click Analytics Engine"
requires-python = ">=3.12"
dependencies = [
    "fastapi[standard]==0.136.1",
    "sqlalchemy[asyncio]==2.0.49",
    "asyncpg==0.31.0",
    "alembic==1.18.4",
    "redis==7.4.0",
    "pydantic-settings==2.13.4",
    "python-dotenv==1.1.0",
]

[project.optional-dependencies]
dev = [
    "pytest==8.3.5",
    "pytest-asyncio==0.26.0",
    "httpx==0.28.1",
    "coverage==7.8.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/snip"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

What does this configuration do?

  • [project] defines metadata and pins every runtime dependency to an exact version (==). This guarantees reproducible builds.
  • [project.optional-dependencies] lists dev-only packages (testing tools) that are not needed in production.
  • [tool.hatch.build.targets.wheel] tells the build system where your source code lives (src/snip).
  • [tool.pytest.ini_options] configures pytest to run in async mode and look for tests in the tests directory.
  • Save pyproject.toml by pressing Cmd+S (macOS) or Ctrl+S (Windows).

✔️ Awesome, I've got everything!

Great. Make sure you have saved pyproject.toml before moving on.

ⓧ I'd like to double check the full code

Here is the complete pyproject.toml you should have at this point:

[project]
name = "snip"
version = "0.1.0"
description = "Production URL Shortener & Click Analytics Engine"
requires-python = ">=3.12"
dependencies = [
    "fastapi[standard]==0.136.1",
    "sqlalchemy[asyncio]==2.0.49",
    "asyncpg==0.31.0",
    "alembic==1.18.4",
    "redis==7.4.0",
    "pydantic-settings==2.13.4",
    "python-dotenv==1.1.0",
]

[project.optional-dependencies]
dev = [
    "pytest==8.3.5",
    "pytest-asyncio==0.26.0",
    "httpx==0.28.1",
    "coverage==7.8.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/snip"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
Install all dependencies

With the pyproject.toml in place, one command resolves and installs every dependency into an isolated virtual environment.

  • Switch back to your terminal (make sure you are inside the snip folder).
  • Install all dependencies (including dev) by running this command:
uv sync --all-extras

What does uv sync do?

uv sync reads your pyproject.toml, resolves all dependency versions, creates a lockfile (uv.lock), and installs everything into a .venv virtual environment inside your project folder.

The --all-extras flag tells uv to also install the packages listed under [project.optional-dependencies] (your testing tools).

You should see uv resolving packages and printing a summary of what it installed. Look for a line showing that packages were installed successfully.

Seeing a resolution error?

Double-check that every version number in your pyproject.toml matches the configuration above exactly. A single typo in a version pin will cause the resolver to fail.

Also confirm that you are running the command from inside the snip folder (run pwd to check).

help me fix a uv sync resolution error in my snip project

  • Verify the virtual environment was created by running:
ls .venv

You should see folders like bin (macOS/Linux) or Scripts (Windows), lib, and pyvenv.cfg inside .venv.

Your project skeleton is ready. Next up, you will build the full directory structure that separates your code into clean, professional layers.

Set Up Project Architecture and File Structure

Your dependencies are locked and your virtual environment is ready. Before writing any application code, you need a project structure that keeps modules organized and imports predictable.

Professional Python projects use a src layout. This means all importable code lives inside a src/ directory, separated from tests, configuration, and tooling. In this step, you'll create the full directory tree and placeholder files that form the skeleton of your URL shortener.

In this step, get ready to:
  • Create the directory structure following Python's src layout pattern.
  • Configure the package so Python can import your modules.
  • Create placeholder files for every module you'll build.
Create the src layout directory structure

Python adds the current working directory to its import path by default. If your package lives at the project root, tests can accidentally import the local source files instead of the installed package. The src layout prevents this by nesting your code one level deeper.

  • Create all the directories your project needs by running this command:
mkdir -p src/snip/services tests

What does this command do?

  • mkdir -p creates directories and any missing parent directories in one go.
  • src/snip/services creates three nested directories: src/, src/snip/, and src/snip/services/.
  • tests creates a top-level directory for your test files.
  • Confirm the directories exist by running:
ls src/snip

You should see services listed in the output.

Don't see the services directory?

  • Make sure you ran the command from inside the snip project folder (the one containing pyproject.toml).
  • Check for typos in the path. The -p flag creates all intermediate directories, so a misspelled folder name creates the wrong tree.

Help me debug my directory structure.

Why src layout instead of flat layout?

In a flat layout, Python's import path includes the project root. Your tests could import raw source files instead of the installed package, masking packaging bugs that only appear in production.

The src layout forces the package to be properly installed before anything can import it. Your pyproject.toml already declares packages = ["src/snip"] so Hatchling knows where to find your code.

Set up the package configuration for imports

For Python to treat a directory as a package, it needs an __init__.py file inside it. Your pyproject.toml already tells the build system where the package lives. Now you need the __init__.py files to complete the connection.

  • Create the __init__.py files for all three package directories by running:
touch src/snip/__init__.py src/snip/services/__init__.py tests/__init__.py

What do __init__.py files do?

These files mark directories as Python packages. Without them, import snip would raise a ModuleNotFoundError. The tests/__init__.py helps pytest discover and import test modules without name collisions.

  • Open src/snip/__init__.py in Cursor and add the package docstring:
"""Snip - Production URL Shortener & Click Analytics Engine."""

Why add a docstring here?

The docstring identifies your package when tools inspect it. The services/__init__.py and tests/__init__.py files stay empty. They only need to exist so Python recognizes those directories as packages.

  • Save src/snip/__init__.py.
  • Verify the package is importable by running:
uv run python -c "import snip; print(snip.__doc__)"

You should see Snip - Production URL Shortener & Click Analytics Engine. printed in the terminal.

Seeing ModuleNotFoundError?

  • Make sure src/snip/__init__.py exists and is saved with the docstring content.
  • Check that your pyproject.toml contains the [tool.hatch.build.targets.wheel] section with packages = ["src/snip"]. This tells the build system where to find the package.
  • Run uv sync to ensure the package is installed in editable mode.

Help me fix this ModuleNotFoundError for my snip package.

Create placeholder files for each module

Each file in your project has a single responsibility. Creating them all now gives you a map of the system before you write any logic.

  • Create the application module files by running:
touch src/snip/config.py src/snip/database.py src/snip/models.py src/snip/schemas.py src/snip/cache.py src/snip/routes.py src/snip/app.py
  • Create the service layer files by running:
touch src/snip/services/url_service.py src/snip/services/click_service.py
  • Create the test files by running:
touch tests/test_url_service.py tests/test_api.py

What is each module for?

  • config.py loads environment variables into typed settings.
  • database.py creates the async database engine and session factory.
  • models.py defines your SQLAlchemy ORM classes (URL, Click).
  • schemas.py defines Pydantic request/response models for API validation.
  • cache.py handles Redis caching for fast URL lookups.
  • routes.py contains FastAPI endpoint definitions.
  • app.py creates the FastAPI application instance and wires everything together.
  • services/url_service.py holds URL shortening business logic.
  • services/click_service.py holds click tracking and analytics logic.
  • Verify the complete file structure by running:
find src tests -type f | sort

You should see all 14 files listed in alphabetical order across the src/ and tests/ directories.

Missing files in the output?

  • If a file is missing, re-run the touch command for that specific file. The touch command is safe to re-run on existing files.
  • Make sure you didn't accidentally create files in the wrong directory. Check that url_service.py is inside src/snip/services/, not src/snip/.

Help me verify my snip project file structure.

✔️ Awesome, I've got everything!

Great. Double check you've saved src/snip/__init__.py with its docstring.

ⓧ I'd like to double check the full code

Here's the only file that needs content at this point. All other files are empty placeholders.

"""Snip - Production URL Shortener & Click Analytics Engine."""

Your project skeleton is complete. Every module has a home, and your package is already importable. Next up, you'll fill in the database models and define the schema for URLs and clicks.

Define Database Models and Configuration

Your project skeleton is ready with all the directories and placeholder files in place. Now it's time to fill in the most important layer of any backend system: the data layer.

The data layer defines what your application stores and how it connects to the database. You'll define typed Python classes that map directly to database tables, plus the configuration and connection logic that ties everything together.

In this step, get ready to:
  • Define SQLAlchemy ORM models with full type annotations for URLs and clicks.
  • Configure application settings with pydantic-settings.
  • Set up the async database engine and session factory.
Define the SQLAlchemy ORM models

An ORM (Object-Relational Mapper) lets you define database tables as Python classes. Instead of writing raw SQL, you work with typed attributes that SQLAlchemy translates into table columns automatically.

Your URL shortener needs two tables: one for shortened URLs and one for tracking every click.

  • In Cursor, open src/snip/models.py from the Explorer sidebar.
  • Replace the placeholder content with the following code for the imports, base class, and URL model:
from datetime import datetime
from typing import Optional

from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class URL(Base):
    __tablename__ = "urls"

    id: Mapped[int] = mapped_column(primary_key=True)
    short_code: Mapped[str] = mapped_column(String(10), unique=True, index=True)
    original_url: Mapped[str] = mapped_column(String(2048))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    click_count: Mapped[int] = mapped_column(default=0)
    is_active: Mapped[bool] = mapped_column(default=True)

    clicks: Mapped[list["Click"]] = relationship(back_populates="url")

What does this code do?

The Base class is the foundation all your models inherit from. SQLAlchemy uses it to track every model you define and generate the correct SQL.

The URL model maps to a urls table in your database. Each field uses Mapped[type] to declare both the Python type and the database column type in one place.

  • short_code is indexed for fast lookups when someone visits a short URL.
  • is_active enables soft-deleting URLs without removing the data permanently.
  • clicks sets up a relationship to the Click model you'll define next.

You won't see a change yet. The next chunk defines the Click model that completes the relationship.

  • Add the Click model below the URL class in the same file:
class Click(Base):
    __tablename__ = "clicks"

    id: Mapped[int] = mapped_column(primary_key=True)
    url_id: Mapped[int] = mapped_column(ForeignKey("urls.id"))
    clicked_at: Mapped[datetime] = mapped_column(server_default=func.now())
    referrer: Mapped[Optional[str]] = mapped_column(String(2048), nullable=True)
    user_agent: Mapped[Optional[str]] = mapped_column(String(512), nullable=True)

    url: Mapped["URL"] = relationship(back_populates="clicks")

What does this code do?

The Click model maps to a clicks table that records every visit to a shortened URL.

  • url_id is a foreign key linking each click back to the URL it belongs to.
  • referrer and user_agent capture where the visitor came from and what browser they used.
  • The back_populates argument links Click.url and URL.clicks so you can navigate the relationship in both directions.
  • Save src/snip/models.py.
  • Verify the models load correctly by running this command in your terminal:
uv run python -c "from snip.models import Base, URL, Click; print('Models loaded successfully')"

You should see Models loaded successfully printed in your terminal.

Seeing an ImportError or SyntaxError?

  • Check that src/snip/__init__.py exists. Without it, Python can't treat snip as an importable package.
  • Make sure you have a blank line between the URL class and the Click class. Python doesn't require it, but a missing colon or indent error nearby can cause confusing messages.
  • Confirm your pyproject.toml has packages = ["src/snip"] under [tool.hatch.build.targets.wheel].

Help me debug my models import error

✔️ Awesome, I've got everything!

Great. Make sure you've saved src/snip/models.py before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime
from typing import Optional

from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class URL(Base):
    __tablename__ = "urls"

    id: Mapped[int] = mapped_column(primary_key=True)
    short_code: Mapped[str] = mapped_column(String(10), unique=True, index=True)
    original_url: Mapped[str] = mapped_column(String(2048))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    click_count: Mapped[int] = mapped_column(default=0)
    is_active: Mapped[bool] = mapped_column(default=True)

    clicks: Mapped[list["Click"]] = relationship(back_populates="url")


class Click(Base):
    __tablename__ = "clicks"

    id: Mapped[int] = mapped_column(primary_key=True)
    url_id: Mapped[int] = mapped_column(ForeignKey("urls.id"))
    clicked_at: Mapped[datetime] = mapped_column(server_default=func.now())
    referrer: Mapped[Optional[str]] = mapped_column(String(2048), nullable=True)
    user_agent: Mapped[Optional[str]] = mapped_column(String(512), nullable=True)

    url: Mapped["URL"] = relationship(back_populates="clicks")
Configure application settings

Your database connection needs credentials, and your app needs to know where Redis lives. Rather than scattering these values across files, you'll centralize them in a single settings module that reads from environment variables.

  • Open src/snip/config.py and replace the placeholder content with the following:
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str
    redis_url: str
    base_url: str = "http://localhost:8000"
    cache_ttl_seconds: int = 3600

    model_config = {"env_file": ".env"}


settings = Settings()

What does this code do?

The Settings class inherits from pydantic-settings BaseSettings, which automatically reads values from environment variables or a .env file.

  • database_url and redis_url are required fields with no defaults. The app won't start without them.
  • base_url and cache_ttl_seconds have sensible defaults for local development.
  • model_config = {"env_file": ".env"} tells pydantic-settings to look for a .env file in the project root.

The module-level settings = Settings() creates a single shared instance that other modules import.

  • Save src/snip/config.py.

You won't see a change yet. This module needs a .env file to load values from, which you'll create in the next step when you set up Docker.

✔️ Awesome, I've got everything!

Great. Make sure you've saved src/snip/config.py before moving on.

ⓧ I'd like to double check the full code

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str
    redis_url: str
    base_url: str = "http://localhost:8000"
    cache_ttl_seconds: int = 3600

    model_config = {"env_file": ".env"}


settings = Settings()
Set up the async database connection

With settings in place, you can create the database engine that your app uses to talk to PostgreSQL. This module creates an async engine (for non-blocking I/O) and a session factory that your API endpoints will use to run queries.

  • Open src/snip/database.py and replace the placeholder content with the following:
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine, AsyncSession

from snip.config import settings

engine = create_async_engine(settings.database_url, echo=False)
async_session = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

What does this code do?

  • create_async_engine builds an async connection pool using the database_url from your settings. The echo=False flag suppresses SQL logging in production.
  • async_sessionmaker is a factory that creates new database sessions on demand. Setting expire_on_commit=False means you can still read attributes from objects after a commit without hitting the database again.
  • get_session is a dependency injection function. FastAPI will call it automatically to provide a fresh session to each request, then close it when the request finishes.
  • Save src/snip/database.py.

You won't see a change yet. This module imports settings from config.py, which needs the .env file you'll create in the next step.

✔️ Awesome, I've got everything!

Great. Make sure you've saved src/snip/database.py before moving on.

ⓧ I'd like to double check the full code

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine, AsyncSession

from snip.config import settings

engine = create_async_engine(settings.database_url, echo=False)
async_session = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

Your data layer is fully defined. Next up, you'll spin up PostgreSQL and Redis with Docker Compose so these models have a real database to connect to.

Launch Docker Compose Infrastructure and Run Migrations

You have your database models and async engine defined, but there is no actual database to connect to yet. Your code describes the shape of the data. Now you need infrastructure to store it.

In this step, you will spin up PostgreSQL and Redis using Docker Compose, then use Alembic to turn your ORM models into real database tables.

In this step, get ready to:
  • Write a Docker Compose file that launches PostgreSQL and Redis containers.
  • Create an environment file that stores your database connection strings.
  • Initialize Alembic, generate a migration from your models, and apply it.
Write the Docker Compose file

Docker Compose lets you define multiple containers in a single file and start them all with one command. You need two services: PostgreSQL for persistent storage and Redis for caching.

  • Create a new file called docker-compose.yml in the root of your snip project folder, then paste this configuration:
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: snip
      POSTGRES_PASSWORD: snip_secret
      POSTGRES_DB: snip
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U snip"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis:
    image: redis:7
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  pgdata:

What does this configuration do?

  • postgres:16 pulls the official PostgreSQL 16 image and creates a database called snip with the user snip and password snip_secret.
  • The pgdata volume persists database data between container restarts so you never lose your tables.
  • redis:7 pulls the official Redis 7 image for caching.
  • Both services include a healthcheck so Docker knows when the container is actually ready to accept connections.
  • Save docker-compose.yml.

YAML indentation errors?

YAML is sensitive to indentation. Make sure you use spaces (not tabs) and that each level is indented by exactly 2 spaces. If your editor shows a red squiggle, check that the environment keys are indented under postgres, not at the same level.

Help me debug a YAML indentation error in my docker-compose.yml

Create the environment file

Your config.py already reads from a .env file. Now you need to create that file with the connection strings that match your Docker services.

  • Create a new file called .env in the root of your snip project folder, then paste these values:
DATABASE_URL=postgresql+asyncpg://snip:snip_secret@localhost:5432/snip
REDIS_URL=redis://localhost:6379/0
BASE_URL=http://localhost:8000

What do these values mean?

  • DATABASE_URL uses the postgresql+asyncpg:// scheme to tell SQLAlchemy to use the async asyncpg driver. The credentials match what you set in docker-compose.yml.
  • REDIS_URL points to the Redis container on port 6379, database index 0.
  • BASE_URL is the public-facing URL your short links will use.
  • Save .env.
  • Confirm both new files exist by running:
ls docker-compose.yml .env

You should see both filenames listed in the output.

Start the infrastructure and run migrations

With your configuration files in place, you can now start the containers and create the database tables.

  • Start both containers in the background by running:
docker compose up -d
  • Wait for both containers to report healthy by running:
docker compose ps

You should see both postgres and redis listed with a status of healthy. If they show starting, wait a few seconds and run the command again.

Containers not starting?

Make sure Docker Desktop is running. If you see a port conflict on 5432 or 6379, another service is already using that port. Stop the conflicting service or change the host port in docker-compose.yml.

Help me troubleshoot why my Docker Compose containers won't start

Now you need Alembic to manage your database schema. Alembic tracks every change to your models as a versioned migration file, so you can upgrade or roll back your database at any time.

  • Initialize Alembic with the async template by running:
uv run alembic init -t async alembic

What does this command do?

This creates an alembic/ directory and an alembic.ini file. The -t async flag selects the async template, which generates an env.py configured to use asyncio and asyncpg instead of synchronous connections.

Next you need to configure Alembic to find your models and database URL.

  • Open alembic.ini and find this line:
sqlalchemy.url = driver://user:pass@localhost/dbname
  • Replace it with your async database URL so the line reads:
sqlalchemy.url = postgresql+asyncpg://snip:snip_secret@localhost:5432/snip
  • Save alembic.ini.
  • Now open alembic/env.py and find these commented lines near the top:
# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = None
  • Replace those lines with this import that points Alembic at your models:
# Import the Base metadata so Alembic can detect model changes
from snip.models import Base
target_metadata = Base.metadata

Why does Alembic need your metadata?

When you run autogenerate, Alembic compares the tables defined in Base.metadata against what actually exists in the database. It generates migration scripts for any differences it finds.

  • Save alembic/env.py.
  • Generate your first migration by running:
uv run alembic revision --autogenerate -m "create urls and clicks tables"

You should see output indicating that Alembic detected two new tables: urls and clicks.

  • Apply the migration to create the tables in your database by running:
uv run alembic upgrade head

You should see output confirming the migration was applied successfully.

  • Verify the tables exist by connecting to PostgreSQL and listing them:
docker compose exec postgres psql -U snip -d snip -c "\dt"

You should see urls, clicks, and alembic_version listed as tables.

Migration command failing?

If you see a connection refused error, your PostgreSQL container may not be fully ready. Run docker compose ps and confirm the status is healthy before retrying.

If you see an import error mentioning snip, make sure your pyproject.toml has the [tool.hatch.build.targets.wheel] section with packages = ["src/snip"].

Help me fix my Alembic migration errors

✔️ Awesome, I've got everything!

Great. Double check you have saved both docker-compose.yml and .env, and that your Docker containers are running healthy.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml:

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: snip
      POSTGRES_PASSWORD: snip_secret
      POSTGRES_DB: snip
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U snip"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis:
    image: redis:7
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  pgdata:

Here is the complete .env:

DATABASE_URL=postgresql+asyncpg://snip:snip_secret@localhost:5432/snip
REDIS_URL=redis://localhost:6379/0
BASE_URL=http://localhost:8000

Your infrastructure is live and your database schema is deployed. Next up, you will build the core business logic that shortens URLs and generates unique short codes.

Build the Core Service Layer

Your database tables are ready and your infrastructure is running. Now you need the business logic that actually shortens URLs, records clicks, and aggregates analytics.

A service layer sits between your API routes and your database. It holds the core logic (hashing URLs, querying clicks) without knowing anything about HTTP requests or response formatting. This separation means you can test the logic independently and swap out the transport layer later.

In this step, get ready to:
  • Implement the URL shortening algorithm with base62 encoding.
  • Build the click tracking service for recording and querying analytics.
  • Define Pydantic schemas for request and response validation.
Implement the URL shortening algorithm

URL shorteners work by converting a long URL into a short, unique code. Your algorithm hashes the URL with SHA-256, converts the hash into a base62 string (letters and digits only), then truncates to 7 characters.

  • Create the file src/snip/services/url_service.py and add the encoding algorithm by pasting this code:
import hashlib
import string
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from snip.models import URL

# Base62 alphabet: a-z, A-Z, 0-9
ALPHABET = string.ascii_letters + string.digits


def base62_encode(num: int) -> str:
    """Convert an integer to a base62 string."""
    if num == 0:
        return ALPHABET[0]
    result = []
    while num > 0:
        result.append(ALPHABET[num % 62])
        num //= 62
    return "".join(reversed(result))


def generate_short_code(url: str, length: int = 7) -> str:
    """Hash a URL with SHA-256 and return a 7-character base62 code."""
    hash_bytes = hashlib.sha256(url.encode()).digest()
    num = int.from_bytes(hash_bytes[:8], "big")
    return base62_encode(num)[:length]

What does this code do?

ALPHABET defines the 62 characters (letters + digits) used in short codes. No special characters means short codes are safe in URLs without encoding.

base62_encode converts an integer into a compact alphanumeric string using repeated division. generate_short_code hashes the full URL with SHA-256, takes the first 8 bytes as an integer, and encodes it. The same URL always produces the same short code.

  • Save url_service.py.
  • Verify the algorithm works by running this in your terminal:
uv run python -c "from snip.services.url_service import generate_short_code; print(generate_short_code('https://example.com'))"

You should see a 7-character alphanumeric string printed in the terminal. Running it again with the same URL produces the same code.

Seeing a ModuleNotFoundError?

Make sure your terminal is in the project root (the folder containing pyproject.toml). The uv run command needs to find the project configuration to resolve imports.

Also check that your Docker Compose stack is still running. The config.py module reads .env on import and validates that the database URL resolves.

Help me fix my import error when running url_service.py

Now you need the async CRUD functions that use this algorithm to create, look up, and deactivate URLs in the database.

  • In src/snip/services/url_service.py, add the following three functions below generate_short_code:

async def create_short_url(session: AsyncSession, original_url: str) -> URL:
    """Generate a short code and persist the new URL record."""
    short_code = generate_short_code(original_url)

    # Check for hash collisions
    existing = await session.execute(
        select(URL).where(URL.short_code == short_code)
    )
    if existing.scalar_one_or_none():
        short_code = generate_short_code(original_url + str(hash(original_url)))

    url = URL(short_code=short_code, original_url=original_url)
    session.add(url)
    await session.commit()
    await session.refresh(url)
    return url


async def get_url_by_code(session: AsyncSession, short_code: str) -> URL | None:
    """Look up an active URL by its short code."""
    result = await session.execute(
        select(URL).where(URL.short_code == short_code, URL.is_active == True)
    )
    return result.scalar_one_or_none()


async def deactivate_url(session: AsyncSession, short_code: str) -> bool:
    """Soft-delete a URL by marking it inactive."""
    url = await get_url_by_code(session, short_code)
    if not url:
        return False
    url.is_active = False
    await session.commit()
    return True

What does this code do?

  • create_short_url generates a short code, checks for collisions (two different URLs hashing to the same code), then saves the record. If a collision occurs, it re-hashes with a salt.
  • get_url_by_code queries only active URLs. Deactivated URLs return None as if they don't exist.
  • deactivate_url performs a soft delete by setting is_active to False instead of removing the row. This preserves analytics history.

What is dependency injection?

Notice how every function receives session: AsyncSession as a parameter instead of creating its own database connection. This is dependency injection. The caller provides the dependency, which makes testing trivial because you can pass in a mock session.

  • Save url_service.py.
  • Confirm the full module imports cleanly by running:
uv run python -c "from snip.services.url_service import create_short_url, get_url_by_code, deactivate_url; print('URL service ready')"

You should see URL service ready printed with no errors.

Seeing an ImportError?

Check that the from snip.models import URL line at the top of your file matches your existing models.py. The URL class name is case-sensitive.

Help me debug why my url_service.py imports are failing

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

import hashlib
import string
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from snip.models import URL

ALPHABET = string.ascii_letters + string.digits


def base62_encode(num: int) -> str:
    if num == 0:
        return ALPHABET[0]
    result = []
    while num > 0:
        result.append(ALPHABET[num % 62])
        num //= 62
    return "".join(reversed(result))


def generate_short_code(url: str, length: int = 7) -> str:
    hash_bytes = hashlib.sha256(url.encode()).digest()
    num = int.from_bytes(hash_bytes[:8], "big")
    return base62_encode(num)[:length]


async def create_short_url(session: AsyncSession, original_url: str) -> URL:
    short_code = generate_short_code(original_url)

    existing = await session.execute(
        select(URL).where(URL.short_code == short_code)
    )
    if existing.scalar_one_or_none():
        short_code = generate_short_code(original_url + str(hash(original_url)))

    url = URL(short_code=short_code, original_url=original_url)
    session.add(url)
    await session.commit()
    await session.refresh(url)
    return url


async def get_url_by_code(session: AsyncSession, short_code: str) -> URL | None:
    result = await session.execute(
        select(URL).where(URL.short_code == short_code, URL.is_active == True)
    )
    return result.scalar_one_or_none()


async def deactivate_url(session: AsyncSession, short_code: str) -> bool:
    url = await get_url_by_code(session, short_code)
    if not url:
        return False
    url.is_active = False
    await session.commit()
    return True
Build the click tracking service

Every time someone visits a short URL, you want to record who clicked, when, and where they came from. The click service handles both writing individual click events and querying aggregate analytics.

  • Create src/snip/services/click_service.py and add the click recording function:
from datetime import datetime
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from snip.models import Click, URL


async def record_click(
    session: AsyncSession,
    url_id: int,
    referrer: str | None = None,
    user_agent: str | None = None,
) -> None:
    """Record a click event and increment the URL's click counter."""
    click = Click(url_id=url_id, referrer=referrer, user_agent=user_agent)
    session.add(click)
    # Lock the URL row to prevent race conditions on click_count
    await session.execute(
        select(URL).where(URL.id == url_id).with_for_update()
    )
    url = await session.get(URL, url_id)
    url.click_count += 1
    await session.commit()

Why use with_for_update()?

When multiple people click the same short link simultaneously, both requests read the same click_count value and both increment it by 1. You'd lose a click. with_for_update() locks the row so only one transaction can modify it at a time.

You won't see a change yet. The next chunk adds the analytics queries that read from the clicks table.

  • Below record_click in the same file, add the two analytics query functions:

async def get_clicks_per_day(
    session: AsyncSession, url_id: int
) -> list[dict[str, int | str]]:
    """Return click counts grouped by day for the last 30 days."""
    result = await session.execute(
        select(
            func.date(Click.clicked_at).label("date"),
            func.count().label("count"),
        )
        .where(Click.url_id == url_id)
        .group_by(func.date(Click.clicked_at))
        .order_by(func.date(Click.clicked_at).desc())
        .limit(30)
    )
    return [{"date": str(row.date), "count": row.count} for row in result]


async def get_top_referrers(
    session: AsyncSession, url_id: int, limit: int = 10
) -> list[dict[str, int | str]]:
    """Return the most common referrer domains, ranked by count."""
    result = await session.execute(
        select(
            Click.referrer,
            func.count().label("count"),
        )
        .where(Click.url_id == url_id, Click.referrer.isnot(None))
        .group_by(Click.referrer)
        .order_by(func.count().desc())
        .limit(limit)
    )
    return [{"referrer": row.referrer, "count": row.count} for row in result]

What do these queries do?

  • get_clicks_per_day groups clicks by their date and counts them. The .limit(30) caps the result to the last 30 days of data.
  • get_top_referrers counts which websites sent the most traffic to this short link. It filters out clicks with no referrer (direct visits).
  • Save click_service.py.
  • Confirm the module imports cleanly by running:
uv run python -c "from snip.services.click_service import record_click, get_clicks_per_day, get_top_referrers; print('Click service ready')"

You should see Click service ready printed with no errors.

Seeing a NameError or import failure?

Make sure your import line reads from snip.models import Click, URL. Both model classes must be imported since record_click uses both.

Help me fix my click_service.py import errors

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from snip.models import Click, URL


async def record_click(
    session: AsyncSession,
    url_id: int,
    referrer: str | None = None,
    user_agent: str | None = None,
) -> None:
    click = Click(url_id=url_id, referrer=referrer, user_agent=user_agent)
    session.add(click)
    await session.execute(
        select(URL).where(URL.id == url_id).with_for_update()
    )
    url = await session.get(URL, url_id)
    url.click_count += 1
    await session.commit()


async def get_clicks_per_day(
    session: AsyncSession, url_id: int
) -> list[dict[str, int | str]]:
    result = await session.execute(
        select(
            func.date(Click.clicked_at).label("date"),
            func.count().label("count"),
        )
        .where(Click.url_id == url_id)
        .group_by(func.date(Click.clicked_at))
        .order_by(func.date(Click.clicked_at).desc())
        .limit(30)
    )
    return [{"date": str(row.date), "count": row.count} for row in result]


async def get_top_referrers(
    session: AsyncSession, url_id: int, limit: int = 10
) -> list[dict[str, int | str]]:
    result = await session.execute(
        select(
            Click.referrer,
            func.count().label("count"),
        )
        .where(Click.url_id == url_id, Click.referrer.isnot(None))
        .group_by(Click.referrer)
        .order_by(func.count().desc())
        .limit(limit)
    )
    return [{"referrer": row.referrer, "count": row.count} for row in result]
Define Pydantic schemas for validation

Your service layer works with raw SQLAlchemy models. The API layer needs a contract for what data comes in and what goes out. Pydantic schemas define that contract and validate data automatically.

  • Create src/snip/schemas.py and paste the following schemas:
from datetime import datetime
from pydantic import BaseModel, HttpUrl


class URLCreate(BaseModel):
    """Request body for creating a new short URL."""
    original_url: HttpUrl


class URLResponse(BaseModel):
    """Response body returned after creating or looking up a URL."""
    short_code: str
    original_url: str
    short_url: str
    created_at: datetime
    click_count: int

    model_config = {"from_attributes": True}


class ClickStats(BaseModel):
    """A single day's click count."""
    date: str
    count: int


class AnalyticsResponse(BaseModel):
    """Full analytics payload for a short URL."""
    short_code: str
    total_clicks: int
    clicks_per_day: list[ClickStats]
    top_referrers: list[dict[str, int]]

What do these schemas do?

  • URLCreate validates that incoming requests contain a proper URL. HttpUrl rejects invalid formats before your code ever runs.
  • URLResponse defines exactly what the API returns. The from_attributes config lets Pydantic read data directly from SQLAlchemy model attributes.
  • AnalyticsResponse packages the aggregated click data (daily counts and top referrers) into a structured response.
  • Save schemas.py.
  • Confirm the schemas load correctly by running:
uv run python -c "from snip.schemas import URLCreate, URLResponse, AnalyticsResponse; print('Schemas ready')"

You should see Schemas ready printed with no errors.

Seeing a validation error on import?

If you see an error about HttpUrl, make sure the import line reads from pydantic import BaseModel, HttpUrl. The HttpUrl type is part of the core pydantic package, which is already installed as a dependency of FastAPI.

Help me fix my Pydantic schema import error

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime
from pydantic import BaseModel, HttpUrl


class URLCreate(BaseModel):
    original_url: HttpUrl


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

    model_config = {"from_attributes": True}


class ClickStats(BaseModel):
    date: str
    count: int


class AnalyticsResponse(BaseModel):
    short_code: str
    total_clicks: int
    clicks_per_day: list[ClickStats]
    top_referrers: list[dict[str, int]]

Your entire service layer is built. You have a URL shortening algorithm, click tracking with analytics queries, and validated schemas. Next up, you'll wire these services into FastAPI endpoints so the outside world can actually use them.

Build the REST API, Redis Cache, and FastAPI App

You have a database layer and service functions that can shorten URLs, resolve them, and track clicks. But none of that is accessible from the outside world yet.

In this step, you will build the three remaining pieces that bring your application to life: a Redis caching layer for fast redirects, a set of REST API endpoints that expose your services over HTTP, and a FastAPI application that ties everything together with proper startup and shutdown handling.

In this step, get ready to:
  • Build a Redis caching layer with get, set, and invalidate operations.
  • Create REST API endpoints for shortening, redirecting, deleting, and analyzing URLs.
  • Wire up the FastAPI application with a lifespan manager and run the server.
Build the Redis caching layer

Every time someone visits a short URL, your app needs the original URL to redirect them. Querying the database on every single click is wasteful when the same mapping rarely changes.

A cache stores frequently accessed data in memory for near-instant retrieval. Your caching layer will sit in front of the database and serve redirects without a round trip to PostgreSQL.

  • Create a new file src/snip/cache.py and add the following code:
import redis.asyncio as redis

from snip.config import settings

# Create a shared Redis client from the REDIS_URL in .env
redis_client = redis.from_url(settings.redis_url, decode_responses=True)


async def get_cached_url(short_code: str) -> str | None:
    # Look up the original URL by short code key
    return await redis_client.get(f"url:{short_code}")


async def set_cached_url(short_code: str, original_url: str) -> None:
    # Store the mapping with an expiry based on config
    await redis_client.set(
        f"url:{short_code}", original_url, ex=settings.cache_ttl_seconds
    )


async def invalidate_cached_url(short_code: str) -> None:
    # Remove the cached entry when a URL is deleted
    await redis_client.delete(f"url:{short_code}")

What does this code do?

This module creates a single shared Redis connection using redis.asyncio. Every function operates on keys prefixed with url: so they stay organized in Redis.

  • get_cached_url returns the original URL if it exists in cache, or None if it has expired or was never set.
  • set_cached_url stores the mapping with a TTL from cache_ttl_seconds (defaulting to 3600 seconds).
  • invalidate_cached_url removes the entry immediately when a URL is deleted, so stale redirects never happen.
  • Save src/snip/cache.py.

Why cache-aside instead of write-through?

The cache-aside pattern only populates the cache on a cache miss. This avoids caching URLs that never get visited. For a URL shortener, most short codes are created but only a fraction see heavy traffic.

Build the REST API endpoints

Your API router defines the HTTP contract for your service. Each endpoint maps an HTTP method and path to a service function, handles validation, and returns structured responses.

  • Create a new file src/snip/routes.py and add the imports, router, and the URL creation endpoint:
from fastapi import APIRouter, Depends, HTTPException, Request, BackgroundTasks
from fastapi.responses import RedirectResponse
from sqlalchemy.ext.asyncio import AsyncSession

from snip.cache import get_cached_url, invalidate_cached_url, set_cached_url
from snip.config import settings
from snip.database import get_session
from snip.schemas import AnalyticsResponse, URLCreate, URLResponse
from snip.services.click_service import get_clicks_per_day, get_top_referrers, record_click
from snip.services.url_service import create_short_url, deactivate_url, get_url_by_code

router = APIRouter()


@router.post("/urls", response_model=URLResponse, status_code=201)
async def shorten_url(
    payload: URLCreate, session: AsyncSession = Depends(get_session)
):
    # Call the service layer to generate a short code and persist it
    url = await create_short_url(session, str(payload.original_url))
    return URLResponse(
        short_code=url.short_code,
        original_url=url.original_url,
        short_url=f"{settings.base_url}/{url.short_code}",
        created_at=url.created_at,
        click_count=url.click_count,
    )

What does this code do?

  • The imports pull in everything the router needs: FastAPI utilities, your cache functions, database session, schemas, and service functions.
  • APIRouter() creates a modular router that gets mounted onto the main app later.
  • The POST /urls endpoint accepts a URLCreate body (validated by Pydantic), calls create_short_url, and returns a URLResponse with a 201 status.
  • Depends(get_session) injects a database session into the handler using FastAPI's dependency injection system.
  • Below the shorten_url function, add the redirect endpoint and its background task helper:
@router.get("/{short_code}")
async def redirect_to_url(
    short_code: str,
    request: Request,
    background_tasks: BackgroundTasks,
    session: AsyncSession = Depends(get_session),
):
    # Check the cache first for a fast redirect
    cached = await get_cached_url(short_code)
    if cached:
        background_tasks.add_task(
            _record_click_background, short_code, request, session
        )
        return RedirectResponse(url=cached, status_code=307)

    # Cache miss: query the database
    url = await get_url_by_code(session, short_code)
    if not url:
        raise HTTPException(status_code=404, detail="URL not found")

    # Populate the cache for next time
    await set_cached_url(short_code, url.original_url)
    background_tasks.add_task(
        _record_click_background, short_code, request, session
    )
    return RedirectResponse(url=url.original_url, status_code=307)


async def _record_click_background(
    short_code: str, request: Request, session: AsyncSession
) -> None:
    # Record click data after the response has already been sent
    url = await get_url_by_code(session, short_code)
    if url:
        await record_click(
            session,
            url.id,
            referrer=request.headers.get("referer"),
            user_agent=request.headers.get("user-agent"),
        )

What does this code do?

This is the most important endpoint in your service. When someone visits a short URL, this handler needs to redirect them as fast as possible.

  • It checks Redis first. If the URL is cached, it returns a 307 redirect immediately.
  • On a cache miss, it queries PostgreSQL, stores the result in Redis for next time, and then redirects.
  • BackgroundTasks schedules click recording AFTER the response is sent. The visitor does not wait for analytics to be written.
  • _record_click_background is a helper that runs asynchronously, capturing the referrer and user-agent from the original request.
  • Below _record_click_background, add the delete and analytics endpoints to finish the router:
@router.delete("/urls/{short_code}", status_code=204)
async def delete_url(
    short_code: str, session: AsyncSession = Depends(get_session)
):
    # Soft-delete the URL by marking it inactive
    success = await deactivate_url(session, short_code)
    if not success:
        raise HTTPException(status_code=404, detail="URL not found")
    # Remove from cache so stale redirects don't happen
    await invalidate_cached_url(short_code)


@router.get("/urls/{short_code}/analytics", response_model=AnalyticsResponse)
async def get_analytics(
    short_code: str, session: AsyncSession = Depends(get_session)
):
    # Look up the URL and gather its click data
    url = await get_url_by_code(session, short_code)
    if not url:
        raise HTTPException(status_code=404, detail="URL not found")

    clicks_per_day = await get_clicks_per_day(session, url.id)
    top_referrers = await get_top_referrers(session, url.id)

    return AnalyticsResponse(
        short_code=url.short_code,
        total_clicks=url.click_count,
        clicks_per_day=clicks_per_day,
        top_referrers=top_referrers,
    )

What does this code do?

  • DELETE /urls/{short_code} soft-deletes a URL (marks it inactive) and invalidates the cache entry. It returns 204 No Content on success.
  • GET /urls/{short_code}/analytics returns aggregated click data including clicks per day and top referrers.
  • Both endpoints use get_url_by_code which only returns active URLs. A deleted URL returns a 404.
  • Save src/snip/routes.py.

Seeing import errors in your editor?

If your editor underlines the snip imports with red squiggles, make sure your virtual environment is activated and that you ran uv pip install -e . previously. The editable install makes your src/snip package importable.

Help me fix import resolution errors for my snip package in Cursor.

✔️ Awesome, I've got everything!

Great. Double check you saved src/snip/routes.py before moving on.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, Depends, HTTPException, Request, BackgroundTasks
from fastapi.responses import RedirectResponse
from sqlalchemy.ext.asyncio import AsyncSession

from snip.cache import get_cached_url, invalidate_cached_url, set_cached_url
from snip.config import settings
from snip.database import get_session
from snip.schemas import AnalyticsResponse, URLCreate, URLResponse
from snip.services.click_service import get_clicks_per_day, get_top_referrers, record_click
from snip.services.url_service import create_short_url, deactivate_url, get_url_by_code

router = APIRouter()


@router.post("/urls", response_model=URLResponse, status_code=201)
async def shorten_url(
    payload: URLCreate, session: AsyncSession = Depends(get_session)
):
    url = await create_short_url(session, str(payload.original_url))
    return URLResponse(
        short_code=url.short_code,
        original_url=url.original_url,
        short_url=f"{settings.base_url}/{url.short_code}",
        created_at=url.created_at,
        click_count=url.click_count,
    )


@router.get("/{short_code}")
async def redirect_to_url(
    short_code: str,
    request: Request,
    background_tasks: BackgroundTasks,
    session: AsyncSession = Depends(get_session),
):
    cached = await get_cached_url(short_code)
    if cached:
        background_tasks.add_task(
            _record_click_background, short_code, request, session
        )
        return RedirectResponse(url=cached, status_code=307)

    url = await get_url_by_code(session, short_code)
    if not url:
        raise HTTPException(status_code=404, detail="URL not found")

    await set_cached_url(short_code, url.original_url)
    background_tasks.add_task(
        _record_click_background, short_code, request, session
    )
    return RedirectResponse(url=url.original_url, status_code=307)


async def _record_click_background(
    short_code: str, request: Request, session: AsyncSession
) -> None:
    url = await get_url_by_code(session, short_code)
    if url:
        await record_click(
            session,
            url.id,
            referrer=request.headers.get("referer"),
            user_agent=request.headers.get("user-agent"),
        )


@router.delete("/urls/{short_code}", status_code=204)
async def delete_url(
    short_code: str, session: AsyncSession = Depends(get_session)
):
    success = await deactivate_url(session, short_code)
    if not success:
        raise HTTPException(status_code=404, detail="URL not found")
    await invalidate_cached_url(short_code)


@router.get("/urls/{short_code}/analytics", response_model=AnalyticsResponse)
async def get_analytics(
    short_code: str, session: AsyncSession = Depends(get_session)
):
    url = await get_url_by_code(session, short_code)
    if not url:
        raise HTTPException(status_code=404, detail="URL not found")

    clicks_per_day = await get_clicks_per_day(session, url.id)
    top_referrers = await get_top_referrers(session, url.id)

    return AnalyticsResponse(
        short_code=url.short_code,
        total_clicks=url.click_count,
        clicks_per_day=clicks_per_day,
        top_referrers=top_referrers,
    )
Wire up the FastAPI application

The application entrypoint creates the FastAPI instance, attaches your router, and defines what happens when the server starts and stops. The lifespan context manager ensures database connections and the Redis client are properly closed on shutdown.

  • Create a new file src/snip/app.py and add the following code:
from contextlib import asynccontextmanager

from fastapi import FastAPI

from snip.cache import redis_client
from snip.database import engine
from snip.routes import router


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: nothing to do (connections are lazy)
    yield
    # Shutdown: close the database engine and Redis client
    await engine.dispose()
    await redis_client.aclose()


app = FastAPI(
    title="Snip",
    description="Production URL Shortener & Click Analytics Engine",
    version="0.1.0",
    lifespan=lifespan,
)

app.include_router(router)

What does this code do?

  • The lifespan async context manager runs code before and after the app serves requests. Everything before yield runs at startup. Everything after yield runs at shutdown.
  • On shutdown, engine.dispose() closes all database connection pool connections. redis_client.aclose() closes the Redis socket.
  • app.include_router(router) mounts all the routes you defined in routes.py onto the main application.
  • Save src/snip/app.py.
  • Make sure your Docker Compose stack is still running (docker compose up -d if not), then start the dev server by running:
uv run fastapi dev src/snip/app.py
  • Open a second terminal and test the API by running:
curl -X POST http://localhost:8000/urls \
  -H "Content-Type: application/json" \
  -d '{"original_url": "https://example.com"}'

You should see a JSON response with a short_code, original_url, short_url, created_at, and click_count of 0.

Server won't start?

  • If you see a connection refused error mentioning port 5432 or 6379, your Docker containers are not running. Run docker compose up -d from the project root.
  • If you see ModuleNotFoundError, confirm you installed the project in editable mode with uv pip install -e . previously.
  • If you see a relation "urls" does not exist error, your Alembic migration has not been applied. Run your migration command from the previous step.

Help me debug why my FastAPI server won't start with the snip app.

Your API is live. You can shorten URLs, resolve them with caching, and query analytics. Next up, you will add configuration management and health check endpoints to make the service production-ready.

Write Unit and Integration Tests

Your URL shortener API is fully functional. But how do you know it actually works correctly after every change you make?

Automated tests answer that question. In this step, you'll write unit tests that verify your shortening algorithm in isolation, then integration tests that hit your live API endpoints through httpx.

In this step, get ready to:
  • Install dev dependencies and configure pytest.
  • Write unit tests for the base62 encoding and short code generation.
  • Write integration tests that exercise the API endpoints.
Install dev dependencies

Your pyproject.toml already declares test libraries under [project.optional-dependencies]. You need to install them into your project environment before pytest can run.

  • Install the dev extras by running this command:
uv sync --extra dev

You should see uv resolve and install pytest, pytest-asyncio, httpx, and coverage into your environment.

Why sync with --extra dev?

Your test dependencies live under [project.optional-dependencies] dev in pyproject.toml. uv does not install optional dependencies by default.

The --extra dev flag tells uv to include that group when syncing your environment. After this sync, uv run pytest will work because uv uses inexact syncing (it does not remove previously-installed extras).

How is pytest already configured?

Your pyproject.toml already includes a [tool.pytest.ini_options] section that sets asyncio_mode = "auto" and testpaths = ["tests"].

This means pytest will automatically discover tests in the tests/ directory. The pytest-asyncio plugin will handle async test functions without extra decorators.

Command not found?

  • If uv is not recognized, make sure you installed it in a previous step and that your shell has been restarted.
  • If you see a resolution error, check that your pyproject.toml has the [project.optional-dependencies] section with a dev key.

Help me install my dev dependencies with uv

Write unit tests for the shortening algorithm

Unit tests verify individual functions in isolation. Your base62_encode and generate_short_code functions are pure (no database, no network), making them perfect candidates.

  • Create the file tests/test_url_service.py and add the following code:
from snip.services.url_service import base62_encode, generate_short_code


def test_base62_encode_zero():
    assert base62_encode(0) == "a"


def test_base62_encode_positive():
    result = base62_encode(12345)
    assert all(c.isalnum() for c in result)
    assert len(result) > 0


def test_generate_short_code_deterministic():
    code1 = generate_short_code("https://example.com")
    code2 = generate_short_code("https://example.com")
    assert code1 == code2


def test_generate_short_code_different_urls():
    code1 = generate_short_code("https://example.com")
    code2 = generate_short_code("https://other.com")
    assert code1 != code2


def test_generate_short_code_length():
    code = generate_short_code("https://example.com")
    assert len(code) == 7

What do these tests verify?

  • test_base62_encode_zero confirms the edge case: encoding 0 returns the first character in the alphabet ("a").
  • test_base62_encode_positive confirms the output is always alphanumeric and non-empty.
  • test_generate_short_code_deterministic confirms the same URL always produces the same short code. This is critical for idempotent URL creation.
  • test_generate_short_code_different_urls confirms different URLs produce different codes.
  • test_generate_short_code_length confirms codes are always 7 characters.
  • Save tests/test_url_service.py.
  • Run the unit tests by running this command:
uv run pytest tests/test_url_service.py -v

You should see all 5 tests pass with green output.

Tests failing with ModuleNotFoundError?

  • Make sure you ran uv sync --extra dev earlier. Without it, pytest is not installed.
  • Check that your src/snip/services/url_service.py file exists and contains the base62_encode and generate_short_code functions.
  • Verify that src/snip/services/__init__.py exists (even if empty). Python needs it to recognize the directory as a package.

Why are my unit tests failing?

Write integration tests with httpx

Integration tests exercise the full request/response cycle. Instead of calling functions directly, you send HTTP requests to your FastAPI app using httpx's ASGITransport. This skips the network layer but runs your entire application stack.

  • Create the file tests/test_api.py and add the following code:
import pytest
from httpx import ASGITransport, AsyncClient

from snip.app import app


@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_shorten_url(client):
    response = await client.post(
        "/urls", json={"original_url": "https://example.com"}
    )
    assert response.status_code == 201
    data = response.json()
    assert "short_code" in data
    assert data["original_url"] == "https://example.com/"


@pytest.mark.asyncio
async def test_redirect_not_found(client):
    response = await client.get("/nonexistent", follow_redirects=False)
    assert response.status_code == 404

What does this test file do?

  • The client fixture creates an async HTTP client that sends requests directly to your FastAPI app in-memory. No real HTTP server is started.
  • test_shorten_url posts a URL and asserts the response has a 201 status code with a short_code in the JSON body.
  • test_redirect_not_found requests a nonexistent short code and confirms a 404 is returned.

The trailing slash in "https://example.com/" is expected. Pydantic's HttpUrl type normalizes URLs by appending a trailing slash to bare domains.

  • Save tests/test_api.py.
  • Make sure your Docker Compose stack is still running (PostgreSQL and Redis must be up for integration tests).
  • Run the full test suite by running this command:
uv run pytest -v

You should see all 7 tests pass (5 unit tests + 2 integration tests).

Integration tests failing with connection errors?

  • The integration tests need PostgreSQL and Redis running. Check your Docker Compose stack is up with docker compose ps.
  • If you see a connection refused error for port 5432, restart the stack with docker compose up -d.
  • Verify your .env file uses localhost for both DATABASE_URL and REDIS_URL.

Why are my integration tests failing with connection errors?

✔️ Awesome, I've got everything!

Great. Double check you've saved both test files.

ⓧ I'd like to double check the full code

Here is the complete tests/test_url_service.py:

from snip.services.url_service import base62_encode, generate_short_code


def test_base62_encode_zero():
    assert base62_encode(0) == "a"


def test_base62_encode_positive():
    result = base62_encode(12345)
    assert all(c.isalnum() for c in result)
    assert len(result) > 0


def test_generate_short_code_deterministic():
    code1 = generate_short_code("https://example.com")
    code2 = generate_short_code("https://example.com")
    assert code1 == code2


def test_generate_short_code_different_urls():
    code1 = generate_short_code("https://example.com")
    code2 = generate_short_code("https://other.com")
    assert code1 != code2


def test_generate_short_code_length():
    code = generate_short_code("https://example.com")
    assert len(code) == 7

And the complete tests/test_api.py:

import pytest
from httpx import ASGITransport, AsyncClient

from snip.app import app


@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_shorten_url(client):
    response = await client.post(
        "/urls", json={"original_url": "https://example.com"}
    )
    assert response.status_code == 201
    data = response.json()
    assert "short_code" in data
    assert data["original_url"] == "https://example.com/"


@pytest.mark.asyncio
async def test_redirect_not_found(client):
    response = await client.get("/nonexistent", follow_redirects=False)
    assert response.status_code == 404

Your project now has automated tests that catch regressions the moment they happen. Next up, you'll bring the entire system together and run it end-to-end.

Run the Full Stack and Verify End-to-End

Your code is written, your tests are ready, and your infrastructure has been running in the background. The final step is to run everything together and confirm the full system works as a production-ready unit.

In this step, you will execute the test suite to prove correctness, then manually exercise every API endpoint with curl commands. This is the same verification a senior engineer does before shipping to production.

In this step, get ready to:
  • Confirm Docker Compose infrastructure is healthy.
  • Run the full test suite and verify all tests pass.
  • Test every API endpoint end-to-end with curl commands.
Confirm the Docker Compose stack is healthy

Before running anything, you need to confirm that PostgreSQL and Redis are both running and healthy. These services back your entire application.

  • Check your container status by running this command:
docker compose ps
  • You should see two containers listed with a status of running and healthy health checks.
  • Confirm both postgres and redis services are shown with ports 5432 and 6379 mapped respectively.

Containers not running?

If you see no containers or they show as exited, restart them from your project directory by running:

docker compose up -d

Wait a few seconds for the health checks to pass, then run docker compose ps again.

help me troubleshoot my Docker Compose containers not starting

Run the complete test suite

Your test suite covers both unit tests for the shortening algorithm and integration tests that hit the real API through httpx. Running them all at once proves the system is wired correctly from top to bottom.

  • Execute the full test suite by running:
uv run pytest

What does this command do?

The uv run prefix ensures pytest runs inside your project's virtual environment with all dependencies available. Pytest auto-discovers tests in the tests/ directory as configured in pyproject.toml.

The unit tests in test_url_service.py verify the base62 encoding and short code generation. The integration tests in test_api.py spin up the app in-process and make real HTTP requests against it.

  • You should see output showing all 7 tests passing with green checkmarks.

Seeing test failures?

If test_api.py tests fail with connection errors, confirm your Docker Compose stack is running. The integration tests connect to the real PostgreSQL database at localhost:5432.

If you see import errors, make sure you previously installed your project in editable mode with dev dependencies.

help me debug my failing pytest tests for the Snip URL shortener

Verify the API end-to-end with curl

Automated tests prove the code is correct. Manual curl testing proves the system works like a real client would use it. You will exercise the full lifecycle: create a short URL, follow the redirect, check analytics, and delete.

  • Start the FastAPI development server by running:
uv run fastapi dev src/snip/app.py

What does this command do?

This starts Uvicorn in development mode with auto-reload enabled. The server listens on http://127.0.0.1:8000 by default. You pass the file path src/snip/app.py so FastAPI knows where your app object lives.

  • You should see log output indicating the server is running on http://127.0.0.1:8000.
  • Open a second terminal window for the curl commands below. Keep the server running in the first terminal.

Now test the full URL shortening lifecycle. Start by creating a short URL.

  • Create a short URL by running:
curl -X POST http://localhost:8000/urls \
  -H "Content-Type: application/json" \
  -d '{"original_url": "https://example.com"}'
  • You should see a JSON response containing short_code, original_url, short_url, created_at, and click_count fields.
  • Copy the short_code value from the response. You will use it in the next commands.

What should I see?

The response looks something like this (your short_code and created_at values will differ):

{
  "short_code": "abc1234",
  "original_url": "https://example.com/",
  "short_url": "http://localhost:8000/abc1234",
  "created_at": "2026-05-23T10:30:00",
  "click_count": 0
}
  • Test the redirect by running this command (replace the short code with yours):
curl -I http://localhost:8000/[[SHORT_CODE="your short_code from the previous response"]]
  • You should see a 307 Temporary Redirect response with a location header pointing to https://example.com/.
  • Check the analytics for your short URL by running:
curl http://localhost:8000/urls/[[SHORT_CODE="your short_code from the previous response"]]/analytics
  • You should see a JSON response with total_clicks, clicks_per_day, and top_referrers fields. The total_clicks should be 1 from the redirect you just triggered.
  • Delete (deactivate) the short URL by running:
curl -X DELETE http://localhost:8000/urls/[[SHORT_CODE="your short_code from the previous response"]]
  • You should see an empty response with a 204 No Content status code. The URL is now deactivated.
  • Confirm deletion worked by trying the redirect again:
curl -I http://localhost:8000/[[SHORT_CODE="your short_code from the previous response"]]
  • You should now see a 404 Not Found response. The deactivated URL no longer resolves.

What did you just prove?

You exercised the full lifecycle of a URL in your system: creation, redirection with click tracking, analytics retrieval, and soft deletion with cache invalidation.

Every layer of your architecture participated. FastAPI routed the request, SQLAlchemy persisted the data, Redis cached the redirect, and background tasks recorded the click.

Getting connection refused?

Make sure the fastapi dev server is still running in your other terminal. The curl commands connect to localhost:8000 where the server is listening.

If you see database errors in the server logs, confirm Docker Compose is still running with docker compose ps.

help me troubleshoot curl errors against my local FastAPI server

  • Stop the development server by pressing Ctrl+C in the terminal where it is running.

Secret mission

Rate Limiting and API Key Authentication

Your URL shortener works, but anyone can hit it as hard as they want. In production, unprotected APIs get hammered by bots, scrapers, and bad actors. Can you add API key authentication and Redis-backed rate limiting to lock things down?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally via Docker, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker containers (PostgreSQL 16 and Redis 7) running via Docker Compose.
  • Docker volume pgdata storing your PostgreSQL database.
  • Local project folder snip on your Desktop.
  • uv package manager installed on your system.

Keep everything running

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

  • Your Docker containers will keep running until you stop them or restart your machine.
  • The snip project folder and all source code remain on your Desktop.
  • uv remains installed and available for other Python projects.

Pause - I'll come back to this later

Shut down running containers to free up system memory and CPU. Your data and files stay intact.

  • Stop the Docker Compose stack by running this command from your snip folder:
  • When you're ready to resume, start everything back up with docker compose up -d from the same folder.
docker compose down

This stops the containers but preserves the pgdata volume, so your database data is still there when you return.

Delete - I don't want to use this again

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

  • Stop the containers and remove the Docker volume by running this from your snip folder:
docker compose down -v
  • Delete the project folder:

macOS

cd ~/Desktop
rm -rf snip

Windows

cd ~\Desktop
rmdir /s /q snip
  • Remove uv and its cached data:

macOS

uv cache clean
rm -r "$(uv python dir)"
rm -r "$(uv tool dir)"
rm ~/.local/bin/uv ~/.local/bin/uvx

Windows

uv cache clean
rm -r "$(uv python dir)"
rm -r "$(uv tool dir)"
rm $HOME\.local\bin\uv.exe
rm $HOME\.local\bin\uvx.exe
rm $HOME\.local\bin\uvw.exe

Well Done!

Well Done!

Nice work! You just built a production-grade Python backend from scratch.

You've learned how to:

  • Build a complete URL shortener API with click analytics, running on FastAPI, PostgreSQL, and Redis with fully async I/O.
  • Design a clean layered architecture with src layout, dependency injection, typed ORM models, and database migrations using Alembic.
  • Write a comprehensive test suite with pytest and httpx, and containerize the entire system with Docker Compose.
  • Add rate limiting and API key authentication with a Redis-backed sliding window to protect your endpoints.

Ready to quiz yourself?