Step 1: in-memory version
Get all five endpoints answering before you think about structure. One file, one dict[str, str] mapping code to URL.
- Set up the project as in Project setup with uv, add
fastapi[standard], and start the server as in Your first app.
- Write the five routes from the spec. Use path and body parameters as in Path, query and body.
- For now generate codes with
secrets (see Design hints) and return the short URL built from the request's base URL.
- Open
/docs and call every endpoint by hand, including the failure cases.
The whole app at this point:
# main.py
import secrets
import string
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse
from pydantic import BaseModel, HttpUrl
app = FastAPI(title="Shortener")
store: dict[str, str] = {} # code -> target URL
clicks: dict[str, int] = {} # code -> click count
ALPHABET = string.ascii_letters + string.digits
class LinkCreate(BaseModel):
url: HttpUrl
code: str | None = None
def make_code(length: int = 6) -> str:
return "".join(secrets.choice(ALPHABET) for _ in range(length))
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/links", status_code=201)
async def create_link(payload: LinkCreate, request: Request) -> dict:
code = payload.code or make_code()
if code in store:
raise HTTPException(status_code=409, detail="code already taken")
store[code] = str(payload.url)
clicks[code] = 0
return {"code": code, "short_url": f"{request.base_url}{code}"}
@app.get("/links/{code}")
async def get_link(code: str) -> dict:
if code not in store:
raise HTTPException(status_code=404, detail="link not found")
return {"code": code, "target_url": store[code], "clicks": clicks[code]}
@app.delete("/links/{code}", status_code=204)
async def delete_link(code: str) -> None:
if code not in store:
raise HTTPException(status_code=404, detail="link not found")
del store[code]
del clicks[code]
@app.get("/{code}") # catch-all: keep it last
async def follow(code: str) -> RedirectResponse:
if code not in store:
raise HTTPException(status_code=404, detail="link not found")
clicks[code] += 1
return RedirectResponse(store[code], status_code=302)
Done when: every row of the spec table returns the right status code from /docs, even though the data vanishes on restart.
Step 2: validation and errors
Replace loose inputs and ad hoc HTTPException calls with schemas and domain exceptions.
- Write
LinkCreate and LinkOut with Pydantic, as in Pydantic v2: HttpUrl for the target, an optional custom code with a length and pattern constraint.
- Add a validator that rejects reserved codes (see Design hints).
- Define exceptions such as
CodeTaken and LinkNotFound, following Exceptions, and map them to 409 and 404 with exception handlers, as in Responses and errors.
- Set
response_model and status_code on each route so the docs match the spec.
Still one file. The new parts are the schemas, the two exceptions with their handlers, and get_target, which every route shares:
# main.py
import secrets
import string
from typing import Annotated
from fastapi import FastAPI, Path, Request
from fastapi.responses import JSONResponse, RedirectResponse
from pydantic import BaseModel, Field, HttpUrl, field_validator
app = FastAPI(title="Shortener")
store: dict[str, str] = {} # code -> target URL
clicks: dict[str, int] = {} # code -> click count
ALPHABET = string.ascii_letters + string.digits
RESERVED = {"docs", "redoc", "openapi.json", "links", "health"}
Code = Annotated[str, Path(min_length=4, max_length=16)]
# --- errors ---------------------------------------------------------------
class LinkNotFound(Exception): ...
class CodeTaken(Exception): ...
@app.exception_handler(LinkNotFound)
async def on_not_found(request: Request, exc: LinkNotFound) -> JSONResponse:
return JSONResponse(status_code=404, content={"detail": f"no link '{exc}'"})
@app.exception_handler(CodeTaken)
async def on_taken(request: Request, exc: CodeTaken) -> JSONResponse:
return JSONResponse(status_code=409, content={"detail": f"code '{exc}' is taken"})
# --- schemas --------------------------------------------------------------
class LinkCreate(BaseModel):
url: HttpUrl
code: str | None = Field(default=None, min_length=4, max_length=16,
pattern=r"^[A-Za-z0-9_-]+$")
@field_validator("code")
@classmethod
def not_reserved(cls, v: str | None) -> str | None:
if v in RESERVED:
raise ValueError("this code is reserved")
return v
class LinkOut(BaseModel):
code: str
short_url: str
target_url: str
clicks: int = 0
# --- logic ----------------------------------------------------------------
def make_code(length: int = 6) -> str:
return "".join(secrets.choice(ALPHABET) for _ in range(length))
def get_target(code: str) -> str:
if code not in store:
raise LinkNotFound(code)
return store[code]
# --- routes ---------------------------------------------------------------
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/links", status_code=201)
async def create_link(payload: LinkCreate, request: Request) -> LinkOut:
code = payload.code or make_code()
if code in store:
raise CodeTaken(code)
store[code] = str(payload.url)
clicks[code] = 0
return LinkOut(code=code, short_url=f"{request.base_url}{code}",
target_url=store[code])
@app.get("/links/{code}")
async def read_link(code: Code, request: Request) -> LinkOut:
target = get_target(code)
return LinkOut(code=code, short_url=f"{request.base_url}{code}",
target_url=target, clicks=clicks[code])
@app.delete("/links/{code}", status_code=204)
async def delete_link(code: Code) -> None:
get_target(code)
del store[code]
del clicks[code]
@app.get("/{code}") # catch-all: keep it last
async def follow(code: Code) -> RedirectResponse:
target = get_target(code)
clicks[code] += 1
return RedirectResponse(target, status_code=302)
Done when: a bad URL gives 422, a duplicate custom code gives 409, and routes contain no raise HTTPException.
Step 3: structure and tests
Split the file by responsibility and lock the behavior in with tests before touching the database.
- Follow Project structure with APIRouter:
routers/ for HTTP, services/ for logic, schemas/, and a config.py using pydantic-settings (see Configuration with pydantic-settings).
- Keep routes thin. They parse input, call a service and return the result.
- Inject the store through
Depends (see Dependency injection with Depends) so tests can swap it.
- Write tests with
TestClient as in Testing endpoints and Testing with pytest: one happy path and one failure per endpoint, using fixtures and dependency overrides.
The layout:
app/
├── __init__.py # empty; one in every folder under app/
├── main.py
├── codes.py # make_code
├── errors.py # LinkNotFound, CodeTaken
├── schemas.py # LinkCreate, LinkOut
├── store.py # the in-memory store and its dependency
├── core/config.py
├── services/links.py
└── routers/
├── links.py
└── redirect.py
tests/
├── conftest.py
└── test_links.py
The code moves out of main.py mostly unchanged. The new idea is that services take the store as an argument and are async already, so the persistence step swaps the argument and keeps every signature.
# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env")
base_url: str = "http://localhost:8000"
settings = Settings()
# app/errors.py
class LinkNotFound(Exception): ...
class CodeTaken(Exception): ...
# app/codes.py
import secrets
import string
ALPHABET = string.ascii_letters + string.digits
def make_code(length: int = 6) -> str:
return "".join(secrets.choice(ALPHABET) for _ in range(length))
# app/schemas.py
from pydantic import BaseModel, Field, HttpUrl, field_validator
RESERVED = {"docs", "redoc", "openapi.json", "links", "health"}
class LinkCreate(BaseModel):
url: HttpUrl
code: str | None = Field(default=None, min_length=4, max_length=16,
pattern=r"^[A-Za-z0-9_-]+$")
@field_validator("code")
@classmethod
def not_reserved(cls, v: str | None) -> str | None:
if v in RESERVED:
raise ValueError("this code is reserved")
return v
class LinkOut(BaseModel):
code: str
short_url: str
target_url: str
clicks: int = 0
# app/store.py
from dataclasses import dataclass
from typing import Annotated
from fastapi import Depends
@dataclass
class LinkRecord:
code: str
target_url: str
clicks: int = 0
Store = dict[str, LinkRecord]
_store: Store = {}
def get_store() -> Store:
return _store
StoreDep = Annotated[Store, Depends(get_store)]
# app/services/links.py
# No FastAPI imports here.
from app.codes import make_code
from app.errors import CodeTaken, LinkNotFound
from app.store import LinkRecord, Store
async def create_link(store: Store, target_url: str, code: str | None = None) -> LinkRecord:
code = code or make_code()
if code in store:
raise CodeTaken(code)
store[code] = LinkRecord(code=code, target_url=target_url)
return store[code]
async def get_link(store: Store, code: str) -> LinkRecord:
if code not in store:
raise LinkNotFound(code)
return store[code]
async def resolve(store: Store, code: str) -> str:
return (await get_link(store, code)).target_url
async def record_click(store: Store, code: str) -> None:
store[code].clicks += 1
async def delete_link(store: Store, code: str) -> None:
await get_link(store, code)
del store[code]
# app/routers/links.py
from typing import Annotated
from fastapi import APIRouter, Path
from app.core.config import settings
from app.schemas import LinkCreate, LinkOut
from app.services import links as svc
from app.store import StoreDep
router = APIRouter(prefix="/links", tags=["links"])
Code = Annotated[str, Path(min_length=4, max_length=16)]
def to_out(link) -> LinkOut:
return LinkOut(code=link.code, short_url=f"{settings.base_url}/{link.code}",
target_url=link.target_url, clicks=link.clicks)
@router.post("", status_code=201)
async def create(payload: LinkCreate, store: StoreDep) -> LinkOut:
return to_out(await svc.create_link(store, str(payload.url), payload.code))
@router.get("/{code}")
async def read(code: Code, store: StoreDep) -> LinkOut:
return to_out(await svc.get_link(store, code))
@router.delete("/{code}", status_code=204)
async def remove(code: Code, store: StoreDep) -> None:
await svc.delete_link(store, code)
# app/routers/redirect.py
from fastapi import APIRouter
from fastapi.responses import RedirectResponse
from app.routers.links import Code
from app.services import links as svc
from app.store import StoreDep
router = APIRouter()
@router.get("/{code}")
async def follow(code: Code, store: StoreDep) -> RedirectResponse:
target = await svc.resolve(store, code)
await svc.record_click(store, code)
return RedirectResponse(target, status_code=302)
# app/main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.errors import CodeTaken, LinkNotFound
from app.routers import links, redirect
app = FastAPI(title="Shortener")
@app.exception_handler(LinkNotFound)
async def on_not_found(request: Request, exc: LinkNotFound) -> JSONResponse:
return JSONResponse(status_code=404, content={"detail": f"no link '{exc}'"})
@app.exception_handler(CodeTaken)
async def on_taken(request: Request, exc: CodeTaken) -> JSONResponse:
return JSONResponse(status_code=409, content={"detail": f"code '{exc}' is taken"})
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
app.include_router(links.router)
app.include_router(redirect.router) # catch-all /{code} goes last
# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.store import get_store
@pytest.fixture
def client():
fresh: dict = {}
app.dependency_overrides[get_store] = lambda: fresh # same dict for the whole test
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
# tests/test_links.py
BODY = {"url": "https://example.com", "code": "abcd"}
def test_health(client):
assert client.get("/health").json() == {"status": "ok"}
def test_create_and_read(client):
r = client.post("/links", json=BODY)
assert r.status_code == 201
assert r.json()["short_url"].endswith("/abcd")
assert client.get("/links/abcd").json()["clicks"] == 0
def test_duplicate_code_is_409(client):
client.post("/links", json=BODY)
assert client.post("/links", json=BODY).status_code == 409
def test_bad_input_is_422(client):
assert client.post("/links", json={"url": "not a url"}).status_code == 422
assert client.post("/links", json={**BODY, "code": "docs"}).status_code == 422
def test_redirect_counts_clicks(client):
client.post("/links", json=BODY)
r = client.get("/abcd", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"] == "https://example.com/"
assert client.get("/links/abcd").json()["clicks"] == 1
def test_unknown_code_is_404(client):
assert client.get("/nope1").status_code == 404
assert client.get("/links/nope1").status_code == 404
def test_delete(client):
client.post("/links", json=BODY)
assert client.delete("/links/abcd").status_code == 204
assert client.delete("/links/abcd").status_code == 404
Run them with uv run pytest.
Done when: pytest passes, and you can change the internals of a service without editing a test.
Step 4: persistence
Swap the dictionary for a database without changing the routes or the HTTP-level tests.
- Start Postgres and Redis as in Run Postgres and Redis locally.
- Define a
Link model with a unique indexed code, url, clicks, created_at and a nullable expires_at (see SQLAlchemy 2.0: engine, session, models).
- Rewrite the service functions following Services own the transaction: the service commits, routes never touch the session. Turn a unique-constraint failure into
CodeTaken.
- Create the table with an Alembic migration (see Migrations with Alembic) instead of
create_all.
- Point the tests at a separate test database and make each test start clean.
docker compose up -d
docker compose exec db createdb -U app shortener_test # a separate database for tests
uv run alembic init -t async migrations
# edit migrations/env.py as in the Alembic stop, then:
uv run alembic revision --autogenerate -m "create links"
uv run alembic upgrade head
# .env
DATABASE_URL=postgresql+asyncpg://app:app@localhost:5432/shortener
Delete app/store.py. schemas.py, errors.py, codes.py and tests/test_links.py do not change, which is the point of the previous step. These files change:
# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env")
database_url: str
test_database_url: str = "postgresql+asyncpg://app:app@localhost:5432/shortener_test"
base_url: str = "http://localhost:8000"
settings = Settings()
# app/db.py
from collections.abc import AsyncIterator
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
from app.core.config import settings
engine = create_async_engine(settings.database_url)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase):
pass
async def get_session() -> AsyncIterator[AsyncSession]:
async with SessionLocal() as session:
yield session
SessionDep = Annotated[AsyncSession, Depends(get_session)]
# app/models.py
from datetime import datetime
from sqlalchemy import DateTime, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db import Base
class Link(Base):
__tablename__ = "links"
id: Mapped[int] = mapped_column(primary_key=True)
code: Mapped[str] = mapped_column(String(16), unique=True, index=True)
target_url: Mapped[str] = mapped_column(String(2048))
clicks: Mapped[int] = mapped_column(default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# app/services/links.py
# No FastAPI imports here. Services own the transaction.
from sqlalchemy import select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.codes import make_code
from app.errors import CodeTaken, LinkNotFound
from app.models import Link
async def create_link(session: AsyncSession, target_url: str, code: str | None = None) -> Link:
link = Link(code=code or make_code(), target_url=target_url)
session.add(link)
try:
await session.commit()
except IntegrityError as e: # unique constraint on code
await session.rollback()
raise CodeTaken(link.code) from e
await session.refresh(link) # load server defaults (created_at)
return link
async def get_link(session: AsyncSession, code: str) -> Link:
link = await session.scalar(select(Link).where(Link.code == code))
if link is None:
raise LinkNotFound(code)
return link
async def resolve(session: AsyncSession, code: str) -> str:
return (await get_link(session, code)).target_url
async def record_click(session: AsyncSession, code: str) -> None:
# atomic increment in the database: no read-modify-write race
await session.execute(update(Link).where(Link.code == code).values(clicks=Link.clicks + 1))
await session.commit()
async def delete_link(session: AsyncSession, code: str) -> None:
link = await get_link(session, code)
await session.delete(link)
await session.commit()
# app/routers/links.py
from typing import Annotated
from fastapi import APIRouter, Path
from app.core.config import settings
from app.db import SessionDep
from app.models import Link
from app.schemas import LinkCreate, LinkOut
from app.services import links as svc
router = APIRouter(prefix="/links", tags=["links"])
Code = Annotated[str, Path(min_length=4, max_length=16)]
def to_out(link: Link) -> LinkOut:
return LinkOut(code=link.code, short_url=f"{settings.base_url}/{link.code}",
target_url=link.target_url, clicks=link.clicks)
@router.post("", status_code=201)
async def create(payload: LinkCreate, session: SessionDep) -> LinkOut:
return to_out(await svc.create_link(session, str(payload.url), payload.code))
@router.get("/{code}")
async def read(code: Code, session: SessionDep) -> LinkOut:
return to_out(await svc.get_link(session, code))
@router.delete("/{code}", status_code=204)
async def remove(code: Code, session: SessionDep) -> None:
await svc.delete_link(session, code)
# app/routers/redirect.py
from fastapi import APIRouter
from fastapi.responses import RedirectResponse
from app.db import SessionDep
from app.routers.links import Code
from app.services import links as svc
router = APIRouter()
@router.get("/{code}")
async def follow(code: Code, session: SessionDep) -> RedirectResponse:
target = await svc.resolve(session, code)
await svc.record_click(session, code)
return RedirectResponse(target, status_code=302)
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.db import engine
from app.errors import CodeTaken, LinkNotFound
from app.routers import links, redirect
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
await engine.dispose()
app = FastAPI(title="Shortener", lifespan=lifespan)
@app.exception_handler(LinkNotFound)
async def on_not_found(request: Request, exc: LinkNotFound) -> JSONResponse:
return JSONResponse(status_code=404, content={"detail": f"no link '{exc}'"})
@app.exception_handler(CodeTaken)
async def on_taken(request: Request, exc: CodeTaken) -> JSONResponse:
return JSONResponse(status_code=409, content={"detail": f"code '{exc}' is taken"})
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
app.include_router(links.router)
app.include_router(redirect.router) # catch-all /{code} goes last
# tests/conftest.py
import asyncio
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool
from app.core.config import settings
from app.db import Base, get_session
from app.main import app
# NullPool: no connection is reused across event loops (pytest's and TestClient's).
test_engine = create_async_engine(settings.test_database_url, poolclass=NullPool)
TestSession = async_sessionmaker(test_engine, expire_on_commit=False)
async def reset_tables() -> None:
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async def override_session():
async with TestSession() as session:
yield session
@pytest.fixture
def client():
asyncio.run(reset_tables()) # empty tables for every test
app.dependency_overrides[get_session] = override_session
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
Why NullPool in tests: TestClient runs the app on its own event loop, and a pooled async connection created on a different loop fails when reused. The table reset uses create_all because tests need a clean schema, not migration history.
Done when: the same tests from the previous step pass against Postgres, and links survive a restart.
Step 5: caching
Make the hot path, the redirect, avoid the database.
- Connect Redis in the lifespan handler (see Lifespan, middleware and background tasks) and inject the client.
- Apply Cache-aside to
GET /{code}: look in Redis, fall back to the database, store the result with a TTL.
- Delete the cache key in the delete service. Decide what a cache entry must never outlive, an expired link for example, and set its TTL accordingly.
- Still count the click on a cache hit. Decide where that write happens, then see the click-counting hint.
Redis was already listed in compose.yaml. schemas.py, errors.py, codes.py and db.py are unchanged; cache.py is new:
# app/cache.py
from typing import Annotated
import redis.asyncio as redis
from fastapi import Depends, Request
def get_redis(request: Request) -> redis.Redis:
return request.app.state.redis
RedisDep = Annotated[redis.Redis, Depends(get_redis)]
# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env")
database_url: str
test_database_url: str = "postgresql+asyncpg://app:app@localhost:5432/shortener_test"
redis_url: str = "redis://localhost:6379/0"
test_redis_url: str = "redis://localhost:6379/15"
base_url: str = "http://localhost:8000"
settings = Settings()
# app/main.py
from contextlib import asynccontextmanager
import redis.asyncio as redis
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.core.config import settings
from app.db import engine
from app.errors import CodeTaken, LinkNotFound
from app.routers import links, redirect
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = redis.from_url(settings.redis_url, decode_responses=True)
yield
await app.state.redis.aclose()
await engine.dispose()
app = FastAPI(title="Shortener", lifespan=lifespan)
@app.exception_handler(LinkNotFound)
async def on_not_found(request: Request, exc: LinkNotFound) -> JSONResponse:
return JSONResponse(status_code=404, content={"detail": f"no link '{exc}'"})
@app.exception_handler(CodeTaken)
async def on_taken(request: Request, exc: CodeTaken) -> JSONResponse:
return JSONResponse(status_code=409, content={"detail": f"code '{exc}' is taken"})
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
app.include_router(links.router)
app.include_router(redirect.router) # catch-all /{code} goes last
# app/services/links.py
# No FastAPI imports here. Services own the transaction.
import redis.asyncio as redis
from sqlalchemy import select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.codes import make_code
from app.errors import CodeTaken, LinkNotFound
from app.models import Link
async def create_link(session: AsyncSession, target_url: str, code: str | None = None) -> Link:
link = Link(code=code or make_code(), target_url=target_url)
session.add(link)
try:
await session.commit()
except IntegrityError as e: # unique constraint on code
await session.rollback()
raise CodeTaken(link.code) from e
await session.refresh(link) # load server defaults (created_at)
return link
async def get_link(session: AsyncSession, code: str) -> Link:
link = await session.scalar(select(Link).where(Link.code == code))
if link is None:
raise LinkNotFound(code)
return link
async def resolve(session: AsyncSession, cache: redis.Redis, code: str) -> str:
if cached := await cache.get(f"link:{code}"):
return cached # hit
link = await get_link(session, code) # miss: raises LinkNotFound
await cache.set(f"link:{code}", link.target_url, ex=86_400) # fill
return link.target_url
async def record_click(session: AsyncSession, code: str) -> None:
# atomic increment in the database: no read-modify-write race
await session.execute(update(Link).where(Link.code == code).values(clicks=Link.clicks + 1))
await session.commit()
async def delete_link(session: AsyncSession, cache: redis.Redis, code: str) -> None:
link = await get_link(session, code)
await session.delete(link)
await session.commit()
await cache.delete(f"link:{code}") # invalidate after the commit
# app/routers/links.py
from typing import Annotated
from fastapi import APIRouter, Path
from app.core.config import settings
from app.cache import RedisDep
from app.db import SessionDep
from app.models import Link
from app.schemas import LinkCreate, LinkOut
from app.services import links as svc
router = APIRouter(prefix="/links", tags=["links"])
Code = Annotated[str, Path(min_length=4, max_length=16)]
def to_out(link: Link) -> LinkOut:
return LinkOut(code=link.code, short_url=f"{settings.base_url}/{link.code}",
target_url=link.target_url, clicks=link.clicks)
@router.post("", status_code=201)
async def create(payload: LinkCreate, session: SessionDep) -> LinkOut:
return to_out(await svc.create_link(session, str(payload.url), payload.code))
@router.get("/{code}")
async def read(code: Code, session: SessionDep) -> LinkOut:
return to_out(await svc.get_link(session, code))
@router.delete("/{code}", status_code=204)
async def remove(code: Code, session: SessionDep, cache: RedisDep) -> None:
await svc.delete_link(session, cache, code)
# app/routers/redirect.py
from fastapi import APIRouter
from fastapi.responses import RedirectResponse
from app.cache import RedisDep
from app.db import SessionDep
from app.routers.links import Code
from app.services import links as svc
router = APIRouter()
@router.get("/{code}")
async def follow(code: Code, session: SessionDep, cache: RedisDep) -> RedirectResponse:
target = await svc.resolve(session, cache, code)
await svc.record_click(session, code)
return RedirectResponse(target, status_code=302)
# tests/conftest.py
import asyncio
import pytest
import redis as redis_sync
import redis.asyncio as redis
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool
from app.cache import get_redis
from app.core.config import settings
from app.db import Base, get_session
from app.main import app
# NullPool: no connection is reused across event loops (pytest's and TestClient's).
test_engine = create_async_engine(settings.test_database_url, poolclass=NullPool)
TestSession = async_sessionmaker(test_engine, expire_on_commit=False)
async def reset_tables() -> None:
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async def override_session():
async with TestSession() as session:
yield session
async def override_redis():
r = redis.from_url(settings.test_redis_url, decode_responses=True)
try:
yield r
finally:
await r.aclose()
@pytest.fixture
def cache(): # sync client for assertions inside tests
r = redis_sync.Redis.from_url(settings.test_redis_url, decode_responses=True)
r.flushdb()
yield r
r.close()
@pytest.fixture
def client(cache):
asyncio.run(reset_tables()) # empty tables for every test
app.dependency_overrides[get_session] = override_session
app.dependency_overrides[get_redis] = override_redis
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
Add this test to tests/test_links.py:
def test_redirect_fills_cache_and_delete_clears_it(client, cache):
client.post("/links", json=BODY)
assert cache.get("link:abcd") is None
client.get("/abcd", follow_redirects=False)
assert cache.get("link:abcd") == "https://example.com/"
client.delete("/links/abcd")
assert cache.get("link:abcd") is None
assert client.get("/abcd").status_code == 404
What the cache saves: a hit skips the SELECT, but record_click still writes on every redirect. Moving that write to a Redis counter is the stretch goal in Design hints.
Done when: a second redirect to the same code does not query Postgres (check with SQL logging or a counter in a test), and a deleted link returns 404 immediately.
Step 6: hardening
Add the three protections a public service needs.
- Expiry: return
410 when expires_at has passed, using timezone-aware datetimes.
- API key: put the check in a dependency and attach it to
POST and DELETE only. Read the key from settings and compare with secrets.compare_digest.
- Rate limiting: a fixed-window counter in Redis as described in Design hints, written as a dependency that returns
429.
- Add a test for each protection.
New files are security.py and ratelimit.py. The rest add an exception, a field and two dependencies:
# app/security.py
import secrets
from typing import Annotated
from fastapi import Header, HTTPException
from app.core.config import settings
async def require_api_key(x_api_key: Annotated[str | None, Header()] = None) -> None:
expected = settings.api_key.encode()
if x_api_key is None or not secrets.compare_digest(x_api_key.encode(), expected):
raise HTTPException(status_code=401, detail="invalid API key")
# app/ratelimit.py
import time
from fastapi import HTTPException, Request
from app.cache import RedisDep
from app.core.config import settings
async def rate_limit(request: Request, cache: RedisDep) -> None:
client = request.client.host if request.client else "unknown"
key = f"rl:{client}:{int(time.time() // 60)}" # one counter per client per minute
async with cache.pipeline(transaction=True) as pipe: # INCR and EXPIRE run together
pipe.incr(key)
pipe.expire(key, 60)
count, _ = await pipe.execute()
if count > settings.rate_limit_per_minute:
raise HTTPException(status_code=429, detail="too many requests")
# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env")
database_url: str
test_database_url: str = "postgresql+asyncpg://app:app@localhost:5432/shortener_test"
redis_url: str = "redis://localhost:6379/0"
test_redis_url: str = "redis://localhost:6379/15"
api_key: str = "change-me"
rate_limit_per_minute: int = 30
base_url: str = "http://localhost:8000"
settings = Settings()
# app/errors.py
class LinkNotFound(Exception): ...
class CodeTaken(Exception): ...
class LinkExpired(Exception): ...
# app/schemas.py
from datetime import datetime, timezone
from pydantic import AwareDatetime, BaseModel, Field, HttpUrl, field_validator
RESERVED = {"docs", "redoc", "openapi.json", "links", "health"}
class LinkCreate(BaseModel):
url: HttpUrl
code: str | None = Field(default=None, min_length=4, max_length=16,
pattern=r"^[A-Za-z0-9_-]+$")
expires_at: AwareDatetime | None = None # must carry a timezone
@field_validator("code")
@classmethod
def not_reserved(cls, v: str | None) -> str | None:
if v in RESERVED:
raise ValueError("this code is reserved")
return v
@field_validator("expires_at")
@classmethod
def in_the_future(cls, v: datetime | None) -> datetime | None:
if v is not None and v <= datetime.now(timezone.utc):
raise ValueError("expires_at must be in the future")
return v
class LinkOut(BaseModel):
code: str
short_url: str
target_url: str
clicks: int = 0
expires_at: datetime | None = None
# app/services/links.py
# No FastAPI imports here. Services own the transaction.
from datetime import datetime, timezone
import redis.asyncio as redis
from sqlalchemy import select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.codes import make_code
from app.errors import CodeTaken, LinkExpired, LinkNotFound
from app.models import Link
async def create_link(session: AsyncSession, target_url: str, code: str | None = None,
expires_at: datetime | None = None) -> Link:
link = Link(code=code or make_code(), target_url=target_url, expires_at=expires_at)
session.add(link)
try:
await session.commit()
except IntegrityError as e: # unique constraint on code
await session.rollback()
raise CodeTaken(link.code) from e
await session.refresh(link) # load server defaults (created_at)
return link
async def get_link(session: AsyncSession, code: str) -> Link:
link = await session.scalar(select(Link).where(Link.code == code))
if link is None:
raise LinkNotFound(code)
return link
async def resolve(session: AsyncSession, cache: redis.Redis, code: str) -> str:
if cached := await cache.get(f"link:{code}"):
return cached # hit
link = await get_link(session, code) # miss: raises LinkNotFound
ttl = 86_400
if link.expires_at is not None:
remaining = int((link.expires_at - datetime.now(timezone.utc)).total_seconds())
if remaining <= 0:
raise LinkExpired(code)
ttl = min(ttl, remaining) # never outlive the link
await cache.set(f"link:{code}", link.target_url, ex=ttl) # fill
return link.target_url
async def record_click(session: AsyncSession, code: str) -> None:
# atomic increment in the database: no read-modify-write race
await session.execute(update(Link).where(Link.code == code).values(clicks=Link.clicks + 1))
await session.commit()
async def delete_link(session: AsyncSession, cache: redis.Redis, code: str) -> None:
link = await get_link(session, code)
await session.delete(link)
await session.commit()
await cache.delete(f"link:{code}") # invalidate after the commit
# app/routers/links.py
from typing import Annotated
from fastapi import APIRouter, Depends, Path
from app.core.config import settings
from app.cache import RedisDep
from app.db import SessionDep
from app.models import Link
from app.ratelimit import rate_limit
from app.schemas import LinkCreate, LinkOut
from app.security import require_api_key
from app.services import links as svc
router = APIRouter(prefix="/links", tags=["links"])
Code = Annotated[str, Path(min_length=4, max_length=16)]
def to_out(link: Link) -> LinkOut:
return LinkOut(code=link.code, short_url=f"{settings.base_url}/{link.code}",
target_url=link.target_url, clicks=link.clicks, expires_at=link.expires_at)
@router.post("", status_code=201,
dependencies=[Depends(require_api_key), Depends(rate_limit)])
async def create(payload: LinkCreate, session: SessionDep) -> LinkOut:
return to_out(await svc.create_link(session, str(payload.url), payload.code,
payload.expires_at))
@router.get("/{code}")
async def read(code: Code, session: SessionDep) -> LinkOut:
return to_out(await svc.get_link(session, code))
@router.delete("/{code}", status_code=204, dependencies=[Depends(require_api_key)])
async def remove(code: Code, session: SessionDep, cache: RedisDep) -> None:
await svc.delete_link(session, cache, code)
In app/main.py, import LinkExpired and add one more handler:
# app/main.py (addition)
@app.exception_handler(LinkExpired)
async def on_expired(request: Request, exc: LinkExpired) -> JSONResponse:
return JSONResponse(status_code=410, content={"detail": f"link '{exc}' has expired"})
In tests/conftest.py, send the key by default: TestClient(app, headers={"x-api-key": settings.api_key}). Then extend tests/test_links.py:
# tests/test_links.py (new imports at the top)
import asyncio
from datetime import datetime, timedelta, timezone
from sqlalchemy import update
from app.core.config import settings
from app.models import Link
from tests.conftest import TestSession
# ...existing tests...
def test_write_endpoints_need_the_api_key(client):
bad = {"x-api-key": "wrong"}
assert client.post("/links", json=BODY, headers=bad).status_code == 401
assert client.delete("/links/abcd", headers=bad).status_code == 401
def test_rate_limit(client, monkeypatch):
monkeypatch.setattr(settings, "rate_limit_per_minute", 2)
codes = ["aaaa", "bbbb", "cccc"]
statuses = [client.post("/links", json={**BODY, "code": c}).status_code for c in codes]
assert statuses == [201, 201, 429]
def test_expiry(client):
assert client.post("/links", json={**BODY, "expires_at": "2020-01-01T00:00:00Z"}).status_code == 422
assert client.post("/links", json={**BODY, "expires_at": "2099-01-01T00:00:00"}).status_code == 422
client.post("/links", json={**BODY, "expires_at": "2099-01-01T00:00:00Z"})
asyncio.run(expire("abcd"))
assert client.get("/abcd", follow_redirects=False).status_code == 410
async def expire(code: str) -> None:
async with TestSession() as session:
past = datetime.now(timezone.utc) - timedelta(hours=1)
await session.execute(update(Link).where(Link.code == code).values(expires_at=past))
await session.commit()
Why expires_at is capped by the cache TTL: a cached entry is served without looking at the database, so it must disappear no later than the link does. Every dependency of POST /links runs in list order, so the key check comes before the rate limit and unauthenticated callers don't use up the quota.
Done when: missing or wrong key gives 401, the limit trips with 429, and an expired link gives 410.
Step 7: ship it
Run the whole system with one command.
- Write a Dockerfile: a Python base image, install dependencies with uv, copy the code, run
fastapi run.
- Add the app to
compose.yaml next to Postgres and Redis. Pass settings through environment variables, and use the service names as hosts.
- Run migrations before the app starts, either as a separate compose service or a startup command.
- Walk the full flow against the containers: create, redirect, details, delete.
Inside the compose network the hosts are the service names, so the URLs differ from your local .env. Environment variables beat .env in pydantic-settings, so nothing in the image needs editing.
# Dockerfile
FROM python:3.13-slim
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
WORKDIR /srv
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project
COPY . .
ENV PATH="/srv/.venv/bin:$PATH"
EXPOSE 8000
CMD ["sh", "-c", "alembic upgrade head && fastapi run app/main.py --port 8000"]
# .dockerignore
.venv
.env
__pycache__
.pytest_cache
.git
# compose.yaml
services:
api:
build: .
ports: ["8000:8000"]
environment:
DATABASE_URL: postgresql+asyncpg://app:app@db:5432/shortener
REDIS_URL: redis://cache:6379/0
BASE_URL: http://localhost:8000
API_KEY: ${API_KEY:?set API_KEY in your shell or a .env file}
depends_on:
db:
condition: service_healthy
cache:
condition: service_started
db:
image: postgres:17
environment:
POSTGRES_USER: app
POSTGRES_PASSWORD: app
POSTGRES_DB: shortener
ports: ["5432:5432"]
volumes: [pgdata:/var/lib/postgresql/data]
healthcheck:
test: ["CMD-SHELL", "pg_isready -U app -d shortener"]
interval: 2s
retries: 15
cache:
image: redis:7
ports: ["6379:6379"]
volumes:
pgdata:
API_KEY=pick-a-secret docker compose up --build
curl -s -X POST localhost:8000/links -H "x-api-key: pick-a-secret" \
-H "content-type: application/json" -d '{"url": "https://example.com"}'
curl -si localhost:8000/health
Watch out: the CLI needs every folder under app/ to be a package (an __init__.py in each), or from app... imports fail inside the container. Running migrations in the start command is fine for one container; with several replicas, run them as a separate one-off step instead.
Done when: docker compose up on a clean checkout gives a working API at /docs with nothing installed but Docker.