URL Shortener Service with SQLite Backend

URL Shortener Service with SQLite Backend

Problem Statement

Sharing long URLs is painful — they break in emails, get truncated in chat apps, and look ugly on printed materials. A URL shortener maps a long URL to a compact, memorable code (e.g., http://localhost:8000/aB3xK9) that redirects to the original. This is the same core functionality behind bit.ly, TinyURL, and t.co.

Building your own shortener is a common internal need: marketing teams want branded short links, ops teams want to track click counts without third-party dependencies, and developers want a self-hosted solution with no external services. This tutorial builds a complete, production-ready shortener with FastAPI for the HTTP layer and SQLite for persistence — no Redis, no external DB, just a single file you can back up with cp.

How It Works

The service exposes three endpoints. POST /shorten accepts a JSON body with a url, validates it, generates a short code (either a random base62 string or a user-supplied custom alias), stores the mapping in SQLite, and returns the short URL. GET /{code} looks up the code, increments a click counter atomically, and issues a 307 Temporary Redirect to the original URL. GET /stats/{code} returns metadata (original URL, click count, creation timestamp) without redirecting.

Design decisions: SQLite with WAL mode for concurrent reads under a single-writer model (perfect for low-to-medium traffic). Base62 encoding of random integers gives short, URL-safe codes with negligible collision probability for the first few million links. INSERT OR IGNORE handles race conditions on custom aliases cleanly. Pydantic validates incoming URLs before they ever touch the database. The click counter uses a single UPDATE ... SET clicks = clicks + 1 statement, which is atomic in SQLite — no read-modify-write race.

Full Script

#!/usr/bin/env python3
"""
URL Shortener Service with SQLite Backend

A minimal production-ready URL shortener exposing a FastAPI app with:
  - POST /shorten           -> create a short link
  - GET  /{code}            -> redirect to the original URL
  - GET  /stats/{code}      -> retrieve link metadata
  - GET  /healthz           -> liveness probe

Run:  python3 -m uvicorn shortener:app --host 0.0.0.0 --port 8000
"""

import os
import re
import secrets
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Iterator

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse
from pydantic import BaseModel, HttpUrl, Field

# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
DB_PATH: str = os.environ.get("SHORTENER_DB", "shortener.db")
BASE_URL: str = os.environ.get("SHORTENER_BASE_URL", "http://localhost:8000").rstrip("/")
CODE_LENGTH: int = int(os.environ.get("SHORTENER_CODE_LENGTH", "7"))
MAX_URL_LENGTH: int = int(os.environ.get("SHORTENER_MAX_URL_LENGTH", "2048"))
ALIAS_PATTERN = re.compile(r"^[A-Za-z0-9_-]{3,32}$")

# Base62 alphabet — digits + uppercase + lowercase, URL-safe.
ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"


# ---------------------------------------------------------------------------
# Database layer
# ---------------------------------------------------------------------------
@contextmanager
def get_db() -> Iterator[sqlite3.Connection]:
    """Yield a SQLite connection with row access by column name."""
    conn = sqlite3.connect(DB_PATH, timeout=10.0)
    conn.row_factory = sqlite3.Row
    # WAL allows concurrent readers while a writer holds the lock.
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def init_db() -> None:
    """Create the links table if it does not already exist."""
    with get_db() as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS links (
                code       TEXT PRIMARY KEY,
                url        TEXT NOT NULL,
                created_at TEXT NOT NULL,
                clicks     INTEGER NOT NULL DEFAULT 0
            )
            """
        )
        # Index on url helps "find existing short link for this URL" queries.
        conn.execute("CREATE INDEX IF NOT EXISTS idx_links_url ON links(url)")


def _encode_base62(num: int) -> str:
    """Encode a non-negative integer into a base62 string."""
    if num == 0:
        return ALPHABET[0]
    out: list[str] = []
    while num > 0:
        num, rem = divmod(num, 62)
        out.append(ALPHABET[rem])
    return "".join(reversed(out))


def generate_code(length: int = CODE_LENGTH) -> str:
    """Generate a cryptographically random base62 code of the given length."""
    # secrets.randbelow is CSPRNG-backed; avoids predictable codes.
    num = secrets.randbits(length * 6)  # 6 bits per base62 char ~= enough entropy
    code = _encode_base62(num)
    # Pad with leading '0' if the random number was small.
    return code.rjust(length, "0")[:length]


def insert_link(code: str, url: str) -> bool:
    """
    Insert a link row. Returns True on success, False if code already exists.
    INSERT OR IGNORE makes this safe against concurrent duplicate inserts.
    """
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with get_db() as conn:
        cur = conn.execute(
            "INSERT OR IGNORE INTO links (code, url, created_at) VALUES (?, ?, ?)",
            (code, url, now),
        )
        return cur.rowcount == 1


def lookup_and_increment(code: str) -> str | None:
    """Atomically increment the click counter and return the target URL."""
    with get_db() as conn:
        row = conn.execute("SELECT url FROM links WHERE code = ?", (code,)).fetchone()
        if row is None:
            return None
        # Single UPDATE is atomic in SQLite — no read-modify-write race.
        conn.execute("UPDATE links SET clicks = clicks + 1 WHERE code = ?", (code,))
        return row["url"]


def get_stats(code: str) -> dict | None:
    """Return metadata for a short code, or None if not found."""
    with get_db() as conn:
        row = conn.execute(
            "SELECT code, url, created_at, clicks FROM links WHERE code = ?", (code,)
        ).fetchone()
        return dict(row) if row else None


# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="URL Shortener", version="1.0.0")


class ShortenRequest(BaseModel):
    url: HttpUrl = Field(..., description="The URL to shorten")
    alias: str | None = Field(
        None,
        description="Optional custom alias (3-32 chars, alphanumeric, _ or -)",
    )


class ShortenResponse(BaseModel):
    code: str
    short_url: str
    original_url: str


@app.on_event("startup")
def _startup() -> None:
    init_db()


@app.get("/healthz")
def healthz() -> dict:
    """Liveness probe — verifies DB is reachable."""
    try:
        with get_db() as conn:
            conn.execute("SELECT 1").fetchone()
        return {"status": "ok"}
    except sqlite3.Error as exc:
        raise HTTPException(status_code=503, detail=f"db unavailable: {exc}")


@app.post("/shorten", response_model=ShortenResponse, status_code=201)
def shorten(payload: ShortenRequest) -> ShortenResponse:
    """Create a short link for the given URL."""
    original = str(payload.url)
    if len(original) > MAX_URL_LENGTH:
        raise HTTPException(
            status_code=413,
            detail=f"URL exceeds maximum length of {MAX_URL_LENGTH} characters",
        )

    if payload.alias:
        if not ALIAS_PATTERN.match(payload.alias):
            raise HTTPException(
                status_code=400,
                detail="alias must be 3-32 chars of [A-Za-z0-9_-]",
            )
        if not insert_link(payload.alias, original):
            raise HTTPException(status_code=409, detail="alias already in use")
        code = payload.alias
    else:
        # Try a few random codes; collisions are astronomically unlikely
        # at CODE_LENGTH >= 7 but we handle them defensively.
        for _ in range(5):
            candidate = generate_code()
            if insert_link(candidate, original):
                code = candidate
                break
        else:
            raise HTTPException(status_code=500, detail="could not allocate code")

    return ShortenResponse(
        code=code,
        short_url=f"{BASE_URL}/{code}",
        original_url=original,
    )


@app.get("/{code}")
def redirect(code: str, request: Request) -> RedirectResponse:
    """Look up a code, count the click, and redirect."""
    url = lookup_and_increment(code)
    if url is None:
        raise HTTPException(status_code=404, detail="short code not found")
    # 307 preserves the original request method (GET -> GET).
    return RedirectResponse(url=url, status_code=307)


@app.get("/stats/{code}")
def stats(code: str) -> dict:
    """Return metadata for a short code without incrementing clicks."""
    data = get_stats(code)
    if data is None:
        raise HTTPException(status_code=404, detail="short code not found")
    data["short_url"] = f"{BASE_URL}/{data['code']}"
    return data


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    # Convenience for `python3 shortener.py` — delegates to uvicorn.
    import uvicorn

    uvicorn.run(
        "shortener:app",
        host=os.environ.get("SHORTENER_HOST", "0.0.0.0"),
        port=int(os.environ.get("SHORTENER_PORT", "8000")),
        reload=False,
    )

Configuration

Usage Examples

Start the server:

$ python3 -m uvicorn shortener:app --port 8000
INFO:     Started server process [12345]
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Create a short link:

$ curl -s -X POST http://localhost:8000/shorten \
    -H 'Content-Type: application/json' \
    -d '{"url":"https://example.com/a/very/long/path?with=query&params=here"}'
{"code":"aB3xK9z","short_url":"http://localhost:8000/aB3xK9z",
 "original_url":"https://example.com/a/very/long/path?with=query&params=here"}

Create with a custom alias, then follow it:

$ curl -s -X POST http://localhost:8000/shorten \
    -H 'Content-Type: application/json' \
    -d '{"url":"https://python.org","alias":"py"}'
{"code":"py","short_url":"http://localhost:8000/py","original_url":"https://python.org/"}

$ curl -sI http://localhost:8000/py | head -3
HTTP/1.1 307 Temporary Redirect
location: https://python.org/

Check stats:

$ curl -s http://localhost:8000/stats/py
{"code":"py","url":"https://python.org/","created_at":"2025-01-15T14:22:03+00:00",
 "clicks":1,"short_url":"http://localhost:8000/py"}

Error Handling

• Invalid URL — Pydantic's HttpUrl rejects non-URL strings with a 422 Unprocessable Entity before any DB work.
• URL too long — Returns 413 Payload Too Large with a clear message.
• Bad alias — Regex check returns 400 Bad Request.
• Duplicate alias — INSERT OR IGNORE returns rowcount == 0, translated to 409 Conflict.
• Unknown code — 404 Not Found on both GET /{code} and /stats/{code}.
• DB failure — /healthz returns 503; other endpoints surface sqlite3.Error as a 500 via FastAPI's default handler. The get_db() context manager rolls back on exception.
• Code collision — Retried up to 5 times with fresh random codes; practically impossible at length 7.
• Concurrent writes — WAL mode + timeout=10.0 on connect() means brief lock contention retries instead of failing immediately.

Extending Ideas

• Expiring links — Add an expires_at column and a background task (or a check in lookup_and_increment) that deletes expired rows. Return 410 Gone for dead links.
• Analytics dashboard — Add a clicks table logging (code, timestamp, user_agent, referer) per hit; expose GET /stats/{code}/timeline returning daily counts.
• Rate limiting — Wrap POST /shorten with slowapi or a small token-bucket in SQLite keyed by client IP to prevent abuse.
• Deduplication — Before generating a new code, SELECT code FROM links WHERE url = ? and return the existing short link if the URL was already shortened (the idx_links_url index already supports this).
• Reverse-proxy deployment — Put nginx or Caddy in front for TLS, and set SHORTENER_BASE_URL=https://sho.rt so responses advertise the public hostname.