A URL shortener with layers, SQLite, and bearer tokens

~45 min · prerequisite: A small notes REST service with FastAPI

What you will be able to do

• Split a small API into routes, service, and storage layers, and say which layer owns which rule.
• Store users and links in SQLite, and protect routes with the signed tokens from Hashing passwords and signing tokens.
• Paginate with limit and offset, return one error shape, and test the flow with TestClient.

The idea

Lesson 5 put every rule inside the route functions, which gets hard to change as a service grows. This lesson splits the code into three layers:

• The routes read HTTP input and pick status codes.
• The service holds the rules, such as "an alias must contain a hyphen".
• The storage runs SQL and nothing else.

Each layer calls only the one below it, so the rules can be tested without HTTP.

Data lives in SQLite, as in lesson 2. A transaction groups changes: the with conn: block commits when it ends cleanly and rolls back on an error.

Protected routes expect Authorization: Bearer <token>. The server keeps no session, so read_token checks the signature and exp on every request.

Short codes use base62: the symbols 0-9, a-z, and A-Z. The id 12345 becomes 3d7, and two ids never share a code.

A user may choose an alias such as spring-sale. Generated codes never contain -, and aliases always do, so the two can never be equal. A UNIQUE column blocks two identical aliases, and the service answers 409 Conflict.

Pagination returns one page. limit sets the page size, offset skips rows, and ordering by the unique id keeps pages stable.

Service errors and validation errors share one shape, {"error": {"code": ..., "message": ...}}, because one handler builds each.

Worked example

Build this folder layout. The commands in this lesson run from shortener-project/:

shortener-project/
    shortener/
        __init__.py     (empty file)
        base62.py
        security.py
        storage.py
        service.py
        app.py
    test_app.py

shortener/base62.py holds the functions from the runnable block in Try it, without its checks. It is pure logic.

shortener/security.py is lesson 3's code. Copy N, R, P, hash_password, verify_password, b64_encode, b64_decode, make_token, and read_token into this file with their imports, and add import os. Then replace lesson 3's SECRET line with this one, so the secret comes from the environment:

SECRET = os.environ.get("SHORTENER_SECRET", "dev-only-secret").encode("utf-8")

The default value is for local tests only.

shortener/storage.py is the only file with SQL. write commits each change or rolls it back.

import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS users (username TEXT PRIMARY KEY, password_hash TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS links (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    owner TEXT NOT NULL,
    target TEXT NOT NULL,
    slug TEXT UNIQUE,
    clicks INTEGER NOT NULL DEFAULT 0
);
"""


class Storage:
    def __init__(self, path):
        self.conn = sqlite3.connect(path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self.conn.executescript(SCHEMA)

    def write(self, sql, params=()):
        with self.conn:
            return self.conn.execute(sql, params)

    def read(self, sql, params=()):
        return self.conn.execute(sql, params).fetchall()

    def add_user(self, username, password_hash):
        self.write("INSERT INTO users VALUES (?, ?)", (username, password_hash))

    def get_password_hash(self, username):
        rows = self.read("SELECT password_hash FROM users WHERE username = ?", (username,))
        return rows[0]["password_hash"] if rows else None

    def insert_link(self, owner, target, alias):
        return self.write("INSERT INTO links (owner, target, slug) VALUES (?, ?, ?)", (owner, target, alias)).lastrowid

    def set_slug(self, link_id, slug):
        self.write("UPDATE links SET slug = ? WHERE id = ?", (slug, link_id))

    def get_by_slug(self, slug):
        rows = self.read("SELECT * FROM links WHERE slug = ?", (slug,))
        return dict(rows[0]) if rows else None

    def record_click(self, link_id):
        self.write("UPDATE links SET clicks = clicks + 1 WHERE id = ?", (link_id,))

    def list_links(self, owner, limit, offset):
        rows = self.read("SELECT * FROM links WHERE owner = ? ORDER BY id LIMIT ? OFFSET ?", (owner, limit, offset))
        total = self.read("SELECT COUNT(*) FROM links WHERE owner = ?", (owner,))[0][0]
        return [dict(row) for row in rows], total

shortener/service.py holds the rules. It raises AppError, and it never builds a response.

import sqlite3
import time

from . import base62
from .security import hash_password, make_token, verify_password


class AppError(Exception):
    def __init__(self, status, code, message):
        super().__init__(message)
        self.status, self.code, self.message = status, code, message


class Service:
    def __init__(self, storage):
        self.storage = storage

    def register(self, username, password):
        try:
            self.storage.add_user(username, hash_password(password))
        except sqlite3.IntegrityError:
            raise AppError(409, "conflict", "username is taken") from None

    def login(self, username, password):
        stored = self.storage.get_password_hash(username)
        if stored is None or not verify_password(password, stored):
            raise AppError(401, "unauthorized", "wrong username or password")
        return make_token({"sub": username, "exp": int(time.time()) + 3600})

    def create_link(self, owner, target, alias=None):
        try:
            link_id = self.storage.insert_link(owner, target, alias)
        except sqlite3.IntegrityError:
            raise AppError(409, "conflict", "alias is already taken") from None
        slug = alias or base62.encode(link_id)
        if alias is None:
            self.storage.set_slug(link_id, slug)  # a second commit, because the slug depends on the id
        return {"id": link_id, "slug": slug, "target": target, "clicks": 0}

    def resolve(self, slug):
        link = self.storage.get_by_slug(slug)
        if link is None:
            raise AppError(404, "not_found", "no link with that slug")
        self.storage.record_click(link["id"])
        return link["target"]

    def list_links(self, owner, limit, offset):
        items, total = self.storage.list_links(owner, limit, offset)
        return {"items": items, "limit": limit, "offset": offset, "total": total}

shortener/app.py is the HTTP layer: Pydantic models check input, current_user is a dependency that reads the token, and exception handlers build the error shape.

import time

from fastapi import Depends, FastAPI, Header, Query, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, RedirectResponse
from pydantic import BaseModel, Field

from .security import read_token
from .service import AppError, Service
from .storage import Storage


class Credentials(BaseModel):
    username: str = Field(min_length=3, max_length=32, pattern=r"^[A-Za-z0-9_]+$")
    password: str = Field(min_length=8, max_length=200)


class LinkIn(BaseModel):
    target: str = Field(min_length=8, pattern=r"^https?://\S+$")
    # Aliases always contain a hyphen, so they never equal a generated code.
    alias: str | None = Field(default=None, min_length=3, max_length=32, pattern=r"^[a-z0-9]+-[a-z0-9-]*$")


def error(status_code, code, message, **extra):
    return JSONResponse(status_code=status_code, content={"error": {"code": code, "message": message, **extra}})


def create_app(db_path="shortener.db"):
    app = FastAPI(title="URL shortener")
    service = Service(Storage(db_path))

    @app.exception_handler(AppError)
    def handle_app_error(request, exc):
        return error(exc.status, exc.code, exc.message)

    @app.exception_handler(RequestValidationError)
    def handle_validation_error(request, exc):
        return error(422, "validation_error", "request is invalid")

    def current_user(authorization: str | None = Header(default=None)):
        if authorization is None or not authorization.startswith("Bearer "):
            raise AppError(401, "unauthorized", "send the header Authorization: Bearer <token>")
        try:
            return read_token(authorization.removeprefix("Bearer "), now=int(time.time()))["sub"]
        except ValueError:
            raise AppError(401, "unauthorized", "token is invalid or expired") from None

    @app.post("/register", status_code=status.HTTP_201_CREATED)
    def register(creds: Credentials):
        service.register(creds.username, creds.password)
        return {"username": creds.username}

    @app.post("/login")
    def login(creds: Credentials):
        return {"token": service.login(creds.username, creds.password), "token_type": "bearer"}

    @app.post("/links", status_code=status.HTTP_201_CREATED)
    def create_link(link: LinkIn, owner: str = Depends(current_user)):
        return service.create_link(owner, link.target, link.alias)

    # Declared before /{slug}, so that GET /links is not read as a slug.
    @app.get("/links")
    def list_links(owner: str = Depends(current_user), limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0)):
        return service.list_links(owner, limit, offset)

    @app.get("/{slug}")
    def follow(slug: str):
        return RedirectResponse(url=service.resolve(slug), status_code=302)

    return app

The redirect uses status 302. Browsers do not cache a 302, so every visit reaches the server and is counted. A 301 would let browsers skip the server after the first visit.

test_app.py runs the whole flow through TestClient, with no server.

import os
import tempfile

from fastapi.testclient import TestClient

from shortener.app import create_app


def main():
    client = TestClient(create_app(os.path.join(tempfile.mkdtemp(), "test.db")), follow_redirects=False)
    account = {"username": "ada_lovelace", "password": "analytical-engine"}
    assert client.post("/register", json=account).status_code == 201
    duplicate = client.post("/register", json=account)
    assert duplicate.status_code == 409 and duplicate.json()["error"]["code"] == "conflict"
    token = client.post("/login", json=account).json()["token"]
    auth = {"Authorization": f"Bearer {token}"}

    made = client.post("/links", json={"target": "https://example.com/a"}, headers=auth)
    assert made.status_code == 201 and made.json()["slug"] == "1"
    hop = client.get("/1")
    assert hop.status_code == 302 and hop.headers["location"] == "https://example.com/a"

    sale = {"target": "https://example.com/sale", "alias": "spring-sale"}
    assert client.post("/links", json=sale, headers=auth).status_code == 201
    assert client.post("/links", json={**sale, "target": "https://example.com/b"}, headers=auth).status_code == 409
    assert client.post("/links", json={"target": "https://example.com/c", "alias": "links"}, headers=auth).status_code == 422
    assert client.get("/spring-sale").headers["location"] == "https://example.com/sale"

    for i in range(25):
        client.post("/links", json={"target": f"https://example.com/{i}"}, headers=auth)
    page = client.get("/links?limit=10&offset=20", headers=auth).json()
    assert page["total"] == 27 and [item["id"] for item in page["items"]] == list(range(21, 28))

    print("all checks passed")


if __name__ == "__main__":
    main()

Try it

This block holds the base62 functions for shortener/base62.py, with checks.

ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"


def encode(number):
    if number < 0:
        raise ValueError("only zero and positive numbers can be encoded")
    if number == 0:
        return ALPHABET[0]
    digits = []
    while number > 0:
        number, remainder = divmod(number, 62)
        digits.append(ALPHABET[remainder])
    return "".join(reversed(digits))


def decode(text):
    number = 0
    for char in text:
        number = number * 62 + ALPHABET.index(char)
    return number


assert encode(0) == "0"
assert encode(61) == "Z"
assert encode(62) == "10"
assert encode(12345) == "3d7"
assert decode("3d7") == 12345
print("base62 checks passed")

Run the server tests from shortener-project/, with the packages from lesson 5 installed. Do not add -I (see the first mistake below):

$ python3 test_app.py
all checks passed

You may see a StarletteDeprecationWarning about httpx, as in lesson 5. It does not change the result.

Start the server with a secret of your choice. The --factory flag tells uvicorn to call create_app():

$ SHORTENER_SECRET=change-me-locally uvicorn shortener.app:create_app --factory --port 8000
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

In a second terminal, register, log in, create a link, and follow it:

$ curl -s -X POST -H "Content-Type: application/json" -d '{"username":"ada_lovelace","password":"analytical-engine"}' http://127.0.0.1:8000/register
{"username":"ada_lovelace"}
$ curl -s -X POST -H "Content-Type: application/json" -d '{"username":"ada_lovelace","password":"analytical-engine"}' http://127.0.0.1:8000/login
{"token":"<long text>","token_type":"bearer"}
$ export TOKEN="<paste the long text here>"
$ curl -s -X POST -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" -d '{"target":"https://example.com/a"}' http://127.0.0.1:8000/links
{"id":1,"slug":"1","target":"https://example.com/a","clicks":0}
$ curl -i http://127.0.0.1:8000/1
HTTP/1.1 302 Found
...
location: https://example.com/a

Stop the server with Ctrl+C.

Common mistakes

1. Running the test with -I. Isolated mode hides the project folder from the import path:

   ModuleNotFoundError: No module named 'shortener'
   

   Run python3 test_app.py from the folder that holds shortener/.

2. Declaring /{slug} before /links. Routes are matched in the order they are written. If follow comes first, GET /links is read as the slug links, and the response is:

   {"error":{"code":"not_found","message":"no link with that slug"}}
   

   Keep the fixed path above the variable one, as lesson 5 showed.

3. Changing SHORTENER_SECRET between runs. A token signed with the old secret no longer matches, so requests that use it get:

   {"error":{"code":"unauthorized","message":"token is invalid or expired"}}
   

   Keep the secret fixed.

Practice

1. Assert that base62.decode("100") == 3844 and base62.encode(3844) == "100".
2. Add a check to test_app.py that POST /register with a 7-character password returns 422.
3. Write is_valid_alias(text) that checks the length (3 to 32 characters) and the alias rule. Assert it for spring-sale, links, and a 40-character text with a hyphen.
4. In two sentences, explain why storage.py should not raise AppError, and which layer should.
5. Stretch: in test_app.py, walk every page with limit=7 until a page is empty. Assert that the ids are 1 to 27, in order, with no duplicates.

Solution

1. python assert base62.decode("100") == 3844 assert base62.encode(3844) == "100"

2. python assert client.post("/register", json={"username": "bo_li", "password": "1234567"}).status_code == 422

3. ```python import re

def is_valid_alias(text): return 3 <= len(text) <= 32 and re.fullmatch(r"[a-z0-9]+-[a-z0-9-]*", text) is not None

assert is_valid_alias("spring-sale") assert not is_valid_alias("links") assert not is_valid_alias("a-" + "b" * 39) ```

1. storage.py only knows SQL, so it returns data or raises a database error. The service decides which rule failed, and the route layer turns that into a response.

2. python seen, offset = [], 0 while True: items = client.get(f"/links?limit=7&offset={offset}", headers=auth).json()["items"] if not items: break seen.extend(item["id"] for item in items) offset += 7 assert seen == list(range(1, 28))

Recap

• Keep routes for HTTP, the service for rules, and storage for SQL. Each layer calls only the one below it.
• Protect routes with a bearer token checked by read_token, and return one error shape for every failure.
• Generated codes are base62 of the row id, and aliases contain a hyphen, so the two kinds can never collide.

Sources

• Python standard library: sqlite3
• Python standard library: re
• FastAPI: Query parameters and string validation
• FastAPI: Dependencies
• FastAPI: Handling errors
• FastAPI: Custom responses, including RedirectResponse
• FastAPI: Testing with TestClient
• Pydantic documentation
• RFC 9110: HTTP Semantics