Reading Real Python Code: A FastAPI Service End-to-End

#The skim-then-dive method

The reason this works: real Python files have a predictable spine. Imports at top, config constants next, then domain types, then dependency wiring, then endpoint or entrypoint code. Once you have read one well-structured file, you know where to look in every other one.

#The file we are reading

Here is the entire file -- around 180 lines. Read it top-to-bottom once before continuing. Sixty seconds. Do not stop on anything confusing. Just note "imports, dataclass, schemas, endpoints" and keep scrolling.

Done with pass 1? You should be able to answer: how many endpoints? (Five.) What is the storage? (In-memory dict.) Where is the validation? (Pydantic models.) If you cannot answer those, scroll back up. That is the entire point of pass 1.

Now let's walk through each section.

#The imports: what every line buys you

from __future__ import annotations

import logging
import os
import secrets
import string
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Annotated

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

#The config block

logger = logging.getLogger(__name__)
logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))

SLUG_CHARS = string.ascii_letters + string.digits
SLUG_LEN = 7
MAX_URL_LEN = 2048

#The domain: dataclasses

@dataclass
class LinkRecord:
    long_url: str
    slug: str
    created_at: datetime
    visits: int = 0
    last_accessed: datetime | None = None

@dataclass
class LinkStore:
    """In-memory store. In production this would be Postgres."""

    _by_slug: dict[str, LinkRecord] = field(default_factory=dict)

    def create(self, long_url: str, slug: str | None = None) -> LinkRecord:
        ...

#The dependency function

_store = LinkStore()


def get_store() -> LinkStore:
    return _store

#The Pydantic schemas

class ShortenRequest(BaseModel):
    long_url: HttpUrl
    custom_slug: str | None = Field(default=None, min_length=3, max_length=32)

#Lifespan: startup and shutdown hooks

@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("URL shortener starting up")
    yield
    logger.info("URL shortener shutting down. %d links in store at exit", len(_store._by_slug))


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

#POST /shorten: putting it together

@app.post("/shorten", response_model=ShortenResponse, status_code=status.HTTP_201_CREATED)
async def shorten(
    payload: ShortenRequest,
    store: Annotated[LinkStore, Depends(get_store)],
) -> ShortenResponse:
    try:
        record = store.create(str(payload.long_url), slug=payload.custom_slug)
    except ValueError as e:
        raise HTTPException(status.HTTP_409_CONFLICT, detail=str(e))
    return ShortenResponse(
        slug=record.slug,
        short_url=f"/{record.slug}",
        long_url=record.long_url,
    )

GET /: path parameters and redirects

@app.get("/{slug}", response_class=RedirectResponse, status_code=status.HTTP_307_TEMPORARY_REDIRECT)
async def follow(
    slug: str,
    store: Annotated[LinkStore, Depends(get_store)],
) -> RedirectResponse:
    record = store.get(slug)
    if record is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"unknown slug: {slug}")
    store.record_visit(slug)
    return RedirectResponse(url=record.long_url)

#HTTPException: why raise instead of return

raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"unknown slug: {slug}")

GET /stats/ and DELETE /

These two endpoints repeat patterns you have already seen, so the third pass is fast.

@app.get("/stats/{slug}", response_model=StatsResponse)
async def stats(
    slug: str,
    store: Annotated[LinkStore, Depends(get_store)],
) -> StatsResponse:
    record = store.get(slug)
    if record is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"unknown slug: {slug}")
    return StatsResponse(...)

@app.delete("/{slug}", status_code=status.HTTP_204_NO_CONTENT)
async def delete(
    slug: str,
    store: Annotated[LinkStore, Depends(get_store)],
) -> None:
    if not store.delete(slug):
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"unknown slug: {slug}")

#The health endpoint

@app.get("/health", response_model=HealthResponse)
async def health(store: Annotated[LinkStore, Depends(get_store)]) -> HealthResponse:
    return HealthResponse(
        status="ok",
        links_in_store=len(store._by_slug),
    )

#Three "how to read code like a senior" tips

#Try it yourself

Server starts on localhost:8000.
POST /shorten {"long_url":"https://example.com"} -> 201 + {"slug":"...","short_url":"/...","long_url":"https://example.com/"}
GET /{slug} -> 307 redirect
GET /stats/{slug} -> 200 + visit count
GET /links?limit=10&offset=0 -> 200 + {"items":[...],"total":N}

# Add to main.py — fill in the TODOs
class LinksResponse(BaseModel):
  items: list[StatsResponse]
  total: int


# TODO: add a list_all method to LinkStore
# def list_all(self, limit: int, offset: int) -> list[LinkRecord]:
#     ...


@app.get("/links", response_model=LinksResponse)
async def list_links(
  limit: int = 10,
  offset: int = 0,
  # TODO: inject the store with Annotated[..., Depends(get_store)]
):
  # TODO: build StatsResponse for each record and return LinksResponse
  pass

#Key Takeaways

• Real Python files have a predictable spine -- imports, config constants, domain types, dependencies, then endpoints/entrypoints. Once you know the spine, you skim by section rather than reading line-by-line
• Pass 1 in 60 seconds, pass 2 in 5 minutes, pass 3 only on what you need -- engineers who try to read everything line-by-line on the first pass never finish. The skim-then-dive method is the highest-leverage reading habit you will learn this year
• "What it does" is the easy half -- "why it is written this way" is where seniority lives -- the same five lines of dependency injection look pointless until you see them swapped out in a test. The why is the entire reason to read code instead of just running it
• The same 15 patterns appear in every Python service -- @dataclass, field(default_factory=...), Annotated[Type, Depends(...)], HTTPException, lifespan @asynccontextmanager, Pydantic request/response model separation, env-var config, structured logging, leading-underscore privacy. You have now seen each one in context
• Reading tests before reading source pays for itself ten times over -- tests are usage examples written by people who already know the code. Source is implementation. Start at the contract; descend into the body only when the contract is not enough

#Where to go next

• Production Python -- now that you can read FastAPI code, learn the logging/CLI/packaging idioms that make a service shippable
• Type Hints & Testing -- the Annotated[] and mypy-compatible patterns you saw in every endpoint
• OOP & Dataclasses -- the background on @dataclass, __post_init__, and field(default_factory=...) that makes the domain section in this lesson click deeper
• Error Handling & Debugging -- the broader theory behind the raise ... except ... raise HTTPException translation layer pattern

"""
url_shortener/main.py
A minimal URL-shortener service with in-memory storage.
Run with: uvicorn main:app --reload
"""
from __future__ import annotations

import logging
import os
import secrets
import string
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Annotated

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


# Config
logger = logging.getLogger(__name__)
logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))

SLUG_CHARS = string.ascii_letters + string.digits
SLUG_LEN = 7
MAX_URL_LEN = 2048


# Domain
@dataclass
class LinkRecord:
    long_url: str
    slug: str
    created_at: datetime
    visits: int = 0
    last_accessed: datetime | None = None


@dataclass
class LinkStore:
    """In-memory store. In production this would be Postgres."""

    _by_slug: dict[str, LinkRecord] = field(default_factory=dict)

    def create(self, long_url: str, slug: str | None = None) -> LinkRecord:
        if slug and slug in self._by_slug:
            raise ValueError(f"slug already taken: {slug}")
        if slug is None:
            slug = self._generate_unique_slug()
        record = LinkRecord(
            long_url=long_url,
            slug=slug,
            created_at=datetime.now(timezone.utc),
        )
        self._by_slug[slug] = record
        return record

    def get(self, slug: str) -> LinkRecord | None:
        return self._by_slug.get(slug)

    def delete(self, slug: str) -> bool:
        return self._by_slug.pop(slug, None) is not None

    def record_visit(self, slug: str) -> None:
        rec = self._by_slug.get(slug)
        if rec is not None:
            rec.visits += 1
            rec.last_accessed = datetime.now(timezone.utc)

    def _generate_unique_slug(self) -> str:
        for _ in range(10):
            candidate = "".join(secrets.choice(SLUG_CHARS) for _ in range(SLUG_LEN))
            if candidate not in self._by_slug:
                return candidate
        raise RuntimeError("could not generate unique slug after 10 attempts")


# Singleton store the dependency returns. Swappable for tests.
_store = LinkStore()


def get_store() -> LinkStore:
    return _store


# Pydantic schemas
class ShortenRequest(BaseModel):
    long_url: HttpUrl
    custom_slug: str | None = Field(default=None, min_length=3, max_length=32)


class ShortenResponse(BaseModel):
    slug: str
    short_url: str
    long_url: str


class StatsResponse(BaseModel):
    slug: str
    long_url: str
    visits: int
    created_at: datetime
    last_accessed: datetime | None


class HealthResponse(BaseModel):
    status: str
    links_in_store: int


# Lifecycle
@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("URL shortener starting up")
    yield
    logger.info("URL shortener shutting down. %d links in store at exit", len(_store._by_slug))


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


# Endpoints
@app.post("/shorten", response_model=ShortenResponse, status_code=status.HTTP_201_CREATED)
async def shorten(
    payload: ShortenRequest,
    store: Annotated[LinkStore, Depends(get_store)],
) -> ShortenResponse:
    try:
        record = store.create(str(payload.long_url), slug=payload.custom_slug)
    except ValueError as e:
        raise HTTPException(status.HTTP_409_CONFLICT, detail=str(e))
    return ShortenResponse(
        slug=record.slug,
        short_url=f"/{record.slug}",
        long_url=record.long_url,
    )


@app.get("/{slug}", response_class=RedirectResponse, status_code=status.HTTP_307_TEMPORARY_REDIRECT)
async def follow(
    slug: str,
    store: Annotated[LinkStore, Depends(get_store)],
) -> RedirectResponse:
    record = store.get(slug)
    if record is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"unknown slug: {slug}")
    store.record_visit(slug)
    return RedirectResponse(url=record.long_url)


@app.get("/stats/{slug}", response_model=StatsResponse)
async def stats(
    slug: str,
    store: Annotated[LinkStore, Depends(get_store)],
) -> StatsResponse:
    record = store.get(slug)
    if record is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"unknown slug: {slug}")
    return StatsResponse(
        slug=record.slug,
        long_url=record.long_url,
        visits=record.visits,
        created_at=record.created_at,
        last_accessed=record.last_accessed,
    )


@app.delete("/{slug}", status_code=status.HTTP_204_NO_CONTENT)
async def delete(
    slug: str,
    store: Annotated[LinkStore, Depends(get_store)],
) -> None:
    if not store.delete(slug):
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"unknown slug: {slug}")


@app.get("/health", response_model=HealthResponse)
async def health(store: Annotated[LinkStore, Depends(get_store)]) -> HealthResponse:
    return HealthResponse(
        status="ok",
        links_in_store=len(store._by_slug),
    )