ENGINEERING HANDBOOK

Volume 2

Backend Codebase
Walkthrough

A file-by-file, line-by-line tour of the entire backend: entry point, core

infrastructure, models, schemas, repositories, services, endpoints,
middleware, exceptions, utilities, and tests — with every bug flagged, not
hidden.

FastAPI SQLAlchemy Pydantic pytest Layered design

Distributed URL Shortener · FastAPI · PostgreSQL 16 · Redis 7 · SQLAlchemy 2.0 (async) · Alembic · Docker
Internal engineering onboarding & interview-preparation document
Table of Contents
1. How to Read This Volume 4
1.1 The complete file inventory 4

2. Application Entry: [Link] 5

2.1 The application factory 5
2.2 Middleware order is intentional 5
2.3 Route registration order is a correctness requirement 6
2.4 The lifespan: startup and shutdown 6

3. Core Infrastructure 7
3.1 [Link] — typed settings 7
3.2 [Link] — async engine, sessions, get_db 7
3.3 [Link] — pool, key helpers, cache abstraction 8
3.4 [Link] — hashing and JWTs 9
3.5 [Link] — structured logs 10

4. Models Layer 11
4.1 models/__init__.py — the model registry 11
4.2 [Link] 11
4.3 [Link] 11
4.4 [Link] 12

5. Schemas Layer 13
5.1 Validation lives in the schema 13
5.2 Response schemas and the short_url trick 13

6. Repositories Layer 14
6.1 URLRepository — representative methods 14
6.2 UserRepository and AnalyticsRepository 14

7. Services Layer — the Business Logic 15

7.1 URLService.create_short_url 15
7.2 The dedup behaviour has a subtle consequence 15
7.3 [Link] — the hot path in code 16
7.4 The background click recorder 16
7.5 Slug generation and collision handling 16
7.6 update_url — a surprising side effect 17
7.7 AuthService 17
7.8 AnalyticsService 17

toc ii
8. API Endpoints & Dependencies 18
8.1 How a protected endpoint wires up 18
8.2 dependencies/[Link] — authentication as a dependency 19
8.3 The redirect endpoint 20
8.4 The analytics endpoints and an auth quirk 21

9. Cross-Cutting Concerns 22
9.1 middleware/request_id.py 22
9.2 middleware/rate_limit.py 22
9.3 exceptions/ — domain errors and global handlers 24
9.4 utils/request_parser.py 25

10. The Test Suite 26

10.1 [Link] — the fixtures that make it work 26
10.2 What is covered 26

toc iii
1. How to Read This Volume
Volume 1 gave you the map. This volume is the guided tour of the territory: every folder, every file, and
the important lines inside each one. We go top-down — from the application entry point, through the
layers, to the tests — because that mirrors how a request actually flows.
For each significant file you get a file card (purpose, what it imports, who imports it, and what breaks if it
disappears), the code that matters, and a plain-English explanation. Bugs and rough edges are flagged, not
hidden.

1.1 The complete file inventory

The backend is roughly 3,800 lines of Python across these files (excluding the virtual environment and
caches):

Area Files

Entry app/[Link]

Core [Link] , [Link] , [Link] , [Link] , [Link]

API api/v1/[Link] + endpoints/ (auth, urls, analytics, redirect)

Business services/ (auth_service, url_service, analytics_service)

Data repositories/ (user, url, analytics)

Models models/ (user, url, analytics)

Schemas schemas/ (user, auth, url, analytics, common)

Cross-cutting dependencies/[Link] , middleware/ (request_id, rate_limit), exceptions/ , utils/

request_parser.py

Migrations alembic/ (env, versions/initial_schema)

Tests tests/ (conftest, unit, integration)

NOTE — Two stray files you will notice

The repository contains app/models/__init__2.py (empty) and a literal directory named tests/
{unit,integration} — an accidental artefact of a shell mkdir that didn’t expand braces. Neither is imported or
used; both are safe to delete. Spotting dead files like these is good hygiene.

Distributed URL Shortener - Engineering Handbook Page 4 of 26

2. Application Entry: [Link]
Field Detail

Purpose Builds and configures the FastAPI application (the ‘app factory’), registers middleware, exception
handlers, health routes, and all routers, and manages startup/shutdown.

Imports [Link], logging, redis pool helpers, exception handlers, both middlewares, the API router and
the redirect router.

Imported by Uvicorn ( [Link]:app ) and the test suite ( create_app ).

If deleted The service cannot start — there is no application object to serve.

2.1 The application factory

Rather than creating a global app inline, the code uses a factory function. This is a deliberate testability choice:
each test can build a fresh, isolated app.

app/[Link]

def create_app() -> FastAPI:

app = FastAPI(
title="URL Shortener API",
description=(...),
version="1.0.0",
docs_url="/docs", redoc_url="/redoc", openapi_url="/[Link]",
lifespan=lifespan,
)
# Middleware
app.add_middleware(CORSMiddleware, allow_origins=settings.CORS_ORIGINS_LIST, ...)
app.add_middleware(RateLimitMiddleware)
app.add_middleware(RequestIDMiddleware)
# Exception handlers
register_exception_handlers(app)
# Health routes (registered FIRST)
@[Link]("/health") ...
@[Link]("/health/ready") ...
# API routes (/api/v1/*)
app.include_router(api_router)
# Redirect catch-all (MUST be last)
app.include_router(redirect_router)
return app

app = create_app()

2.2 Middleware order is intentional

Middleware added with add_middleware wraps the app from the inside out, so the last one added runs first on
the way in. The order here means RequestIDMiddleware runs first (so every subsequent log line carries a
request ID), then rate limiting, then CORS.

Distributed URL Shortener - Engineering Handbook Page 5 of 26

 Request order through the middleware stack and the headers added on the way out.

2.3 Route registration order is a correctness requirement

KEY IDEA — Why the redirect router is registered LAST

The redirect route is GET /{slug} — a catch-all that matches any single path segment. If it were registered
before /health or /docs , it would swallow them (treating ‘health’ as a slug). FastAPI matches routes in
registration order, so health and /api/v1/* are registered first and the catch-all last. The redirect handler also
keeps an explicit exclusion set as a second line of defence.

2.4 The lifespan: startup and shutdown

The lifespan async context manager runs code once at startup (before serving) and once at shutdown. Here
it configures logging and verifies Redis connectivity on boot, and closes the Redis pool cleanly on exit.

app/[Link]

@asynccontextmanager
async def lifespan(app: FastAPI):
configure_logging([Link])
[Link]("Starting URL Shortener", environment=[Link], ...)
try:
redis = await get_redis_pool()
await [Link]()
[Link]("Redis connection established")
except Exception as e:
[Link]("Redis connection failed", error=str(e))
yield # <-- app serves requests here
[Link]("Shutting down URL Shortener")
await close_redis_pool()

NOTE — Redis failure at startup is logged, not fatal

If Redis is unreachable at boot, the app logs an error but still starts. That aligns with the fail-open philosophy: the
service can run degraded (no cache, no rate limiting) rather than refusing to boot.

Distributed URL Shortener - Engineering Handbook Page 6 of 26

3. Core Infrastructure
The core/ package holds the process-wide singletons and utilities every other layer depends on:
configuration, the database engine, the Redis pool, security primitives, and logging.

3.1 [Link] — typed settings

Field Detail

Purpose Parses and validates all environment variables into a typed Settings object; derives the database
URLs and CORS list; exposes a cached singleton.

Imports pydantic, pydantic-settings.

Imported by Almost everything — database, redis, security, main all read settings .

If deleted The app cannot read its configuration; nothing that needs a URL, key, or limit can initialise.

Using Pydantic Settings means a wrong type (say, a non-numeric port) fails loudly at startup rather than silently
later. The database URLs are computed properties, so credentials live in one place:

app/core/[Link]

@property
def DATABASE_URL(self) -> str: # async, used by FastAPI at runtime
return (f"postgresql+asyncpg://{self.DATABASE_USERNAME}:{self.DATABASE_PASSWORD}"
f"@{self.DATABASE_HOSTNAME}:{self.DATABASE_PORT}/{self.DATABASE_NAME}")

@property
def SYNC_DATABASE_URL(self) -> str: # sync, used only by Alembic
return (f"postgresql+psycopg2://{self.DATABASE_USERNAME}:{self.DATABASE_PASSWORD}"
f"@{self.DATABASE_HOSTNAME}:{self.DATABASE_PORT}/{self.DATABASE_NAME}")

KEY IDEA — Two database URLs, one config

The runtime app uses the async asyncpg driver; Alembic’s schema inspection needs a sync psycopg2
connection. Deriving both from the same credentials avoids drift.

The settings object is wrapped in @lru_cache so parsing happens once, and exposed as a module-level
settings alias for convenient imports.

3.2 [Link] — async engine, sessions, get_db

Field Detail

Purpose Creates the async SQLAlchemy engine (the connection pool) and session factory; defines the
declarative Base ; provides the get_db per-request dependency.

Imports sqlalchemy async, [Link].

Imported by Every repository, every model (via Base ), every route needing a session, Alembic.

If deleted No database access anywhere; models can’t be defined; the app is inert.

The engine is created once per process and configured with pool parameters straight from settings:

Distributed URL Shortener - Engineering Handbook Page 7 of 26

 app/core/[Link]

engine = create_async_engine(
settings.DATABASE_URL,
echo=not settings.is_production, # log SQL in dev only
pool_size=settings.DB_POOL_SIZE, # 10 warm connections
max_overflow=settings.DB_MAX_OVERFLOW, # +20 under burst
pool_timeout=settings.DB_POOL_TIMEOUT, # wait max 30s for a free conn
pool_recycle=settings.DB_POOL_RECYCLE, # recycle conns older than 30 min
pool_pre_ping=True, # verify a conn is alive before using it
)

The get_db dependency yields a session and guarantees cleanup. The try/except/finally rolls back on
any exception and always closes the session, so a failed request never leaks a connection back to the pool in
a dirty state:

app/core/[Link]

async def get_db() -> AsyncGenerator[AsyncSession, None]:

async with AsyncSessionLocal() as session:
try:
yield session
except Exception:
await [Link]()
raise
finally:
await [Link]()

Q1 INTERMEDIATE What is the difference between the Engine and the Session?

The Engine is the process-level connection pool — expensive to create, created once. The Session is a per-
request unit of work — cheap, tracks ORM object state, and is not thread-safe, so each request gets its own.
Creating an Engine per request would exhaust the database’s connection limit almost immediately.

3.3 [Link] — pool, key helpers, cache abstraction

Field Detail

Purpose Manages the singleton Redis connection pool; defines TTL constants and namespaced key builders;
provides the RedisCache helper class with fail-open operations.

Imports [Link], [Link].

Imported by Services (caching, blocklist), middleware (rate limiting), main (startup ping), dependencies.

If deleted No caching, no rate limiting, no token revocation; redirects always hit the database.

Keys are built by small helper functions so the naming scheme is centralised and consistent (e.g.
url:slug:aB3cD4eF , auth:blocklist:<jti> , ratelimit:ip:[Link] ). The RedisCache class wraps raw
operations and — crucially — fails open:

Distributed URL Shortener - Engineering Handbook Page 8 of 26

 app/core/[Link]

class RedisCache:
async def get(self, key: str) -> str | None:
try:
return await self._r.get(key)
except Exception:
return None # cache error -> behave like a miss

async def set(self, key: str, value: str, ttl: int) -> None:
try:
await self._r.setex(key, ttl, value)
except Exception:
pass # cache is optional; never break the request

async def increment(self, key: str, ttl: int | None = None) -> int:
count = await self._r.incr(key) # atomic
if count == 1 and ttl:
await self._r.expire(key, ttl)
return count

KEY IDEA — Atomicity is why rate limiting is correct across replicas

Redis INCR is atomic and single-threaded, so even with ten app replicas hammering the same key there is no
read-modify-write race. That is the whole reason rate limiting and click counting can be safely centralised in Redis.

3.4 [Link] — hashing and JWTs

Field Detail

Purpose Password hashing/verification with bcrypt; creation, decoding, and JTI extraction of JWT access and
refresh tokens.

Imports bcrypt, python-jose, [Link].

Imported by AuthService (login, register, refresh, revocation), the security unit tests.

If deleted No authentication is possible — passwords can’t be hashed or verified, tokens can’t be issued or
validated.

Passwords use bcrypt directly (not passlib) with a work factor of 12 (212 rounds). bcrypt is deliberately slow
and auto-salts each hash, so identical passwords produce different hashes and brute-forcing a stolen
database is expensive.

app/core/[Link]

_BCRYPT_ROUNDS = 12

def hash_password(plain_password: str) -> str:

salt = [Link](rounds=_BCRYPT_ROUNDS)
return [Link](plain_password.encode(), salt).decode()

def verify_password(plain_password: str, hashed_password: str) -> bool:

try:
return [Link](plain_password.encode(), hashed_password.encode())
except Exception:
return False

Distributed URL Shortener - Engineering Handbook Page 9 of 26

Tokens carry standard claims plus two important extras: a unique jti (JWT ID, a UUID) used for revocation,
and a type (‘access’ or ‘refresh’) so a refresh token can never be used where an access token is expected.

app/core/[Link]

def create_access_token(data, expires_delta=None) -> str:

to_encode = [Link]()
expire = [Link](UTC) + (expires_delta or timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({"exp": expire, "iat": [Link](UTC),
"jti": str(uuid4()), "type": "access"})
return [Link](to_encode, settings.SECRET_KEY, algorithm=[Link])

BUG / GOTCHA — bcrypt&rsquo;s 72-byte limit

bcrypt silently truncates input beyond 72 bytes. Passwords are capped at 100 characters by the schema, so a >72-
byte password would have its tail ignored. It is a minor, well-known bcrypt caveat — worth mentioning if asked
about password handling.

3.5 [Link] — structured logs

Field Detail

Purpose Configures structlog for JSON output in production and colourised console output in
development; silences noisy third-party loggers.

Imports logging, structlog.

Imported by main (at startup), any module calling get_logger .

If deleted Logs become unstructured/no correlation; observability degrades but the app still runs.

Structured logs are machine-parseable (searchable by request_id , user_id , slug ) which is essential once
you run many replicas. The request-ID binding that makes cross-instance tracing possible is set by
RequestIDMiddleware (Chapter 9).

Distributed URL Shortener - Engineering Handbook Page 10 of 26

4. Models Layer
Models are SQLAlchemy 2.0 declarative classes that map Python objects to database tables using typed
Mapped[...] annotations. Volume 3 covers the schema design; here we read the code.

4.1 models/__init__.py — the model registry

This file imports all three models so that [Link] knows about every table. Alembic’s autogenerate
compares [Link] against the live database; a model that isn’t imported here would be invisible to
migrations.

app/models/__init__.py

from [Link] import URLAnalytics # noqa: F401

from [Link] import URL # noqa: F401
from [Link] import User # noqa: F401
__all__ = ["User", "URL", "URLAnalytics"]

4.2 [Link]
The User model uses a UUID primary key (unguessable, no enumeration, no sequential-write hotspot), a
unique+indexed email and username, a role string for RBAC, and soft-delete via is_active . It declares a
one-to-many relationship to URL .

app/core/../models/[Link]

class User(Base):
__tablename__ = "users"
id: Mapped[[Link]] = mapped_column(UUID(as_uuid=True),
primary_key=True, default=uuid.uuid4, index=True)
email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, index=True)
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
role: Mapped[str] = mapped_column(String(20), default="user", nullable=False)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
is_verified: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
urls: Mapped[list["URL"]] = relationship("URL", back_populates="owner", lazy="select")

4.3 [Link]
The heart of the domain. Key columns: a unique+indexed slug (the redirect lookup key), the long_url
(Text), a url_hash (SHA-256, indexed) for duplicate detection, a denormalized click_count (BigInteger),
and an expires_at that is nullable (NULL = never expire). Composite indexes support the common queries.

Distributed URL Shortener - Engineering Handbook Page 11 of 26

 app/models/[Link]

class URL(Base):
__tablename__ = "urls"
slug: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, index=True)
long_url: Mapped[str] = mapped_column(Text, nullable=False)
url_hash: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
click_count: Mapped[int] = mapped_column(BigInteger, default=0, nullable=False)
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
owner_id: Mapped[[Link]] = mapped_column(
UUID(as_uuid=True), ForeignKey("[Link]", ondelete="CASCADE"), nullable=False,
index=True)
__table_args__ = (
Index("ix_urls_owner_active", "owner_id", "is_active"),
Index("ix_urls_owner_hash", "owner_id", "url_hash"), # duplicate detection
)

KEY IDEA — The index on slug is the performance keystone

The unique B-tree index on slug is what makes a redirect lookup O(log n) even at a billion rows. On a cache miss
the query is SELECT ... WHERE slug = ? — a single index seek.

4.4 [Link]
Each row is one click event: a hashed visitor fingerprint (privacy-preserving), IP, country, browser/OS/device,
referrer, and a timestamp. Composite indexes support the aggregation queries (by date, by country, by visitor).
Volume 3 explains the denormalization strategy between this event log and the click_count column.

Distributed URL Shortener - Engineering Handbook Page 12 of 26

5. Schemas Layer
Schemas are Pydantic v2 models that define and validate the shape of requests and responses. They are the
API’s contract and its first line of input defence. Models (ORM) and schemas (API) are deliberately separate:
never expose an ORM object with its hashed_password directly.

5.1 Validation lives in the schema

For example, registration enforces username format and password strength declaratively:

app/schemas/[Link]

class UserCreate(BaseModel):
email: EmailStr
username: str = Field(min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_-]+$")
password: str = Field(min_length=8, max_length=100)

@field_validator("password")
@classmethod
def password_strength(cls, v: str) -> str:
if not any([Link]() for c in v):
raise ValueError("Password must contain at least one uppercase letter")
if not any([Link]() for c in v):
raise ValueError("Password must contain at least one digit")
return v

KEY IDEA — Invalid input never reaches your logic

Because FastAPI validates the body against the schema before the handler runs, a malformed email or weak
password is rejected with a 422 automatically. Your service code can assume its inputs are already well-formed.

5.2 Response schemas and the short_url trick

URLResponse sets from_attributes=True so it can be built directly from an ORM object. It also has a
short_url field defaulted to empty string and filled in by the service layer, because building it needs the
configured BASE_URL .

app/schemas/[Link]

class URLResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: [Link]
slug: str
long_url: str
short_url: str = "" # populated by URLService._to_response()
click_count: int
expires_at: datetime | None
...

Other schema files: [Link] (login/refresh/logout requests, the token response, and an internal
TokenPayload ), [Link] (the breakdown and leaderboard shapes), and [Link] (a uniform
APIResponse envelope, an ErrorResponse , and pagination helpers).

Distributed URL Shortener - Engineering Handbook Page 13 of 26

6. Repositories Layer
Repositories are the only place raw SQL/ORM queries live. Each wraps one model, takes an AsyncSession
in its constructor, and exposes intention-revealing methods ( get_by_slug , hash_exists_for_owner ).
Services call repositories; repositories never call services.

6.1 URLRepository — representative methods

Two methods are worth reading closely. First, the atomic click increment — it uses a single SQL UPDATE ...
SET click_count = click_count + 1 , which avoids a read-modify-write race entirely:

app/repositories/url_repository.py

async def increment_click_count(self, slug: str) -> None:

await self._db.execute(
update(URL).where([Link] == slug)
.values(click_count=URL.click_count + 1))
await self._db.commit()

Second, paginated listing returns both the page of items and the total count in one method, so the service can
compute page counts:

app/repositories/url_repository.py

async def get_by_owner(self, owner_id, *, page=1, size=20) -> tuple[list[URL], int]:

base_query = select(URL).where(URL.owner_id == owner_id, URL.is_active == True)
total = (await self._db.execute(
select([Link]()).select_from(base_query.subquery()))).scalar_one()
result = await self._db.execute(
base_query.order_by(URL.created_at.desc())
.offset((page - 1) * size).limit(size))
return [Link]().all(), total

TIP — Offset pagination has a known scaling caveat

OFFSET pagination is simple but degrades on deep pages (the database still scans and discards the skipped
rows). For very large result sets, keyset (‘seek’) pagination is preferred. For a per-user URL list this is a non-issue,
but it’s a good thing to name in an interview.

6.2 UserRepository and AnalyticsRepository

UserRepository handles create/get-by-id/email/username, existence checks, and a soft-delete ( is_active =
False ). It lower-cases emails on write and read so lookups are case-insensitive. AnalyticsRepository
inserts click events and runs the aggregation queries ( COUNT , COUNT(DISTINCT visitor_hash) , and GROUP
BY for date/country/browser/OS/device).

Distributed URL Shortener - Engineering Handbook Page 14 of 26

7. Services Layer — the Business Logic
Services orchestrate a use case across repositories and Redis. This is where the interesting logic — and the
interesting bugs — live.

7.1 URLService.create_short_url
The creation flow: hash the URL, short-circuit if the user already shortened it, validate or generate a slug,
persist, and prime the cache.

app/services/url_service.py

async def create_short_url(self, data, owner_id) -> URLResponse:

long_url = str(data.long_url)
url_hash = _hash_url(long_url)
existing = await self._url_repo.hash_exists_for_owner(url_hash, owner_id)
if existing:
return self._to_response(existing) # dedup: return existing
if data.custom_alias:
if await self._url_repo.slug_exists(data.custom_alias):
raise AliasAlreadyExistsException(...) # 409
slug, is_custom = data.custom_alias, True
else:
slug, is_custom = await self._generate_unique_slug(), False

expires_at = data.expires_at
if expires_at is None:
expires_at = [Link](UTC) + timedelta(minutes=30) # <-- see BUG
url = await self._url_repo.create(
slug=slug, long_url=long_url, url_hash=url_hash, owner_id=owner_id,
title=[Link], is_custom_alias=is_custom,
expires_at=data.expires_at) # <-- passes original None
await self._cache.set(url_cache_key(slug), long_url, ttl=URL_CACHE_TTL)
return self._to_response(url)

BUG / GOTCHA — The expiry default is dead code

Look at the last two lines carefully. The code computes a 30-minute default into the local variable expires_at ,
but then passes data.expires_at (the original None ) to create() . So the computed default is never used,
and new URLs are stored with a NULL expiry (never expire). The fix is to pass the local expires_at variable.
This is the root cause of the ‘expiry doesn’t work’ behaviour noted in Volume 1.

7.2 The dedup behaviour has a subtle consequence

NOTE — Custom alias can be silently ignored

Because the duplicate check runs before the custom-alias branch, if you have already shortened a URL and then
submit the same URL again with a custom alias, the service returns the pre-existing short URL and ignores your
alias. Defensible (one URL per user), but surprising — a good design-discussion point.

Distributed URL Shortener - Engineering Handbook Page 15 of 26

 7.3 [Link] — the hot path in code
app/services/url_service.py

async def redirect(self, slug, background_tasks, request_meta) -> str:

cached_url = await self._cache.get(url_cache_key(slug))
if cached_url:
background_tasks.add_task(self._record_click_background, slug=slug,
request_meta=request_meta)
return cached_url # <-- no expiry re-check here
url = await self._url_repo.get_by_slug(slug)
if not url or not url.is_active:
raise URLNotFoundException(...) # 404
if url.expires_at and url.expires_at < [Link](UTC):
raise URLExpiredException(...) # 410
await self._cache.set(url_cache_key(slug), url.long_url, ttl=URL_CACHE_TTL)
background_tasks.add_task(self._record_click_background, slug=slug,
request_meta=request_meta)
return url.long_url

BUG / GOTCHA — Expiry is only checked on the cache-miss path

On a cache hit the function returns immediately without checking expires_at . So even if the expiry bug in 7.1
were fixed, a link that expired while cached would keep redirecting until its 1-hour Redis TTL lapsed. A correct
implementation caches the expiry alongside the URL (or shortens the TTL to the remaining lifetime) and checks it
on hits too. Volume 6 shows the fix.

7.4 The background click recorder

Runs after the response is sent; looks up the URL, increments the counter, and inserts the analytics row. The
docstring itself explains the BackgroundTask-vs-Celery decision (covered fully in Volume 3):

app/services/url_service.py

async def _record_click_background(self, slug, request_meta) -> None:

url = await self._url_repo.get_by_slug(slug)
if not url:
return
await self._url_repo.increment_click_count(slug)
await self._analytics_repo.record_click(url_id=[Link], ip_address=..., ...)

7.5 Slug generation and collision handling

app/services/url_service.py

async def _generate_unique_slug(self, max_retries: int = 3) -> str:

for attempt in range(max_retries):
slug = _generate_slug(settings.NANOID_LENGTH)
if not await self._url_repo.slug_exists(slug):
return slug
if attempt == max_retries - 1:
return _generate_slug(settings.NANOID_LENGTH + 2) # longer on repeat clash
return _generate_slug(settings.NANOID_LENGTH + 2)

Distributed URL Shortener - Engineering Handbook Page 16 of 26

 NOTE — A tiny logical gap

On the final fallback the longer slug is returned without re-checking uniqueness. With 6410 combinations the risk is
astronomically small, and the unique index would reject a genuine clash anyway, but strictly the loop could verify
the fallback too.

7.6 update_url — a surprising side effect

app/services/url_service.py

async def update_url(self, slug, data, owner_id) -> URLResponse:

url = await self._get_owned_url(slug, owner_id)
update_fields = data.model_dump(exclude_unset=True)
if "expires_at" not in update_fields:
update_fields["expires_at"] = [Link](UTC) + timedelta(minutes=30) # !!
if "long_url" in update_fields:
update_fields["long_url"] = str(update_fields["long_url"])
update_fields["url_hash"] = _hash_url(update_fields["long_url"])
updated = await self._url_repo.update(url, **update_fields)
await self._cache.delete(url_cache_key(slug)) # invalidate cache
return self._to_response(updated)

BUG / GOTCHA — Editing the title silently sets a 30-minute expiry

If the update payload doesn’t include expires_at , the code forces a new expiry 30 minutes out. So changing
only a link’s title makes the link expire in half an hour — almost certainly unintended. The correct behaviour is to
leave expires_at untouched when the client didn’t send it. Note the one thing this method gets right: it
invalidates the Redis cache after updating.

7.7 AuthService
Handles register, login, refresh, logout, and current-user resolution. Login verifies the bcrypt hash and issues
tokens; refresh validates the token type and checks the blocklist; logout writes both JTIs to the Redis blocklist
with a TTL equal to the refresh token’s lifetime so the blocklist self-cleans.

SECURITY — A login timing side-channel (documented honestly)

The login code comments claim it hashes a dummy password when the user doesn’t exist (to keep response time
constant), but the implementation actually returns early on an unknown email. That leaves a small user-
enumeration timing side-channel: ‘unknown email’ returns faster than ‘known email, wrong password’. The fix is
to always run a bcrypt verification against a dummy hash. Volume 4 revisits this.

7.8 AnalyticsService
Thin orchestration over AnalyticsRepository : it looks up the URL, then gathers total clicks, unique visitors,
and the five breakdowns into a single URLAnalyticsResponse . Ownership is enforced at the endpoint
(Chapter 8), not here.

Distributed URL Shortener - Engineering Handbook Page 17 of 26

8. API Endpoints & Dependencies
Endpoints are thin: they declare the route, its request/response schemas, the required dependencies, and
delegate to a service. The heavy lifting is dependency injection.

8.1 How a protected endpoint wires up

Consider POST /api/v1/urls/ . It declares two dependencies: the current user (which forces authentication)
and a URL service (assembled from a database session and a Redis client).

app/api/v1/endpoints/[Link]

def _get_url_service(db: AsyncSession = Depends(get_db),

redis: [Link] = Depends(get_redis)) -> URLService:
return URLService(db=db, cache=RedisCache(redis))

@[Link]("/", response_model=URLResponse, status_code=201)

async def create_url(data: URLCreate, current_user: CurrentUser,
url_service: URLService = Depends(_get_url_service)):
return await url_service.create_short_url(data, owner_id=current_user.id)

FastAPI resolves the whole dependency tree per request, in the right order, and cleans it up afterwards:

Distributed URL Shortener - Engineering Handbook Page 18 of 26

 The dependency tree FastAPI resolves for create_url: the current user and the service, each built from a session and a Redis client.

8.2 dependencies/[Link] — authentication as a dependency

get_current_user extracts the bearer token, validates it via AuthService , and returns a User — or raises
401. Two Annotated aliases make routes read cleanly: CurrentUser requires any authenticated user;
AdminUser requires the admin role via a dependency factory.

app/dependencies/[Link]

def require_role(*roles: str):

async def _check_role(current_user: User = Depends(get_current_user)) -> User:
if current_user.role not in roles:
raise HTTPException(status_code=403, detail=f"Requires role: {...}")
return current_user
return _check_role

CurrentUser = Annotated[User, Depends(get_current_user)]

AdminUser = Annotated[User, Depends(require_role("admin"))]

The full protected-request validation sequence:

Distributed URL Shortener - Engineering Handbook Page 19 of 26

 get_current_user: reject missing credentials, decode, verify type, check the Redis blocklist, load the user, verify active.

8.3 The redirect endpoint

Guards against treating system paths as slugs, extracts request metadata for analytics, and returns a 302.
One rough edge: a leftover debug print .

app/api/v1/endpoints/[Link]

@[Link]("/{slug}", response_class=RedirectResponse, status_code=302)

async def redirect_to_url(slug, request, background_tasks,
url_service=Depends(_get_url_service)):
if slug in _EXCLUDED_PATHS: # health, docs, api, ...
raise HTTPException(status_code=404, detail="Not found")
request_meta = extract_request_meta(request)
long_url = await url_service.redirect(slug, background_tasks, request_meta)
print("LONG URL =", repr(long_url), type(long_url)) # <-- debug leftover
return RedirectResponse(url=long_url, status_code=302)

BUG / GOTCHA — Remove the debug print

The print(...) line is development scaffolding that shipped by accident. It writes to stdout on every redirect —
noise in production and a tiny performance cost. Delete it (and rely on structured logging if you need the value).

Distributed URL Shortener - Engineering Handbook Page 20 of 26

 8.4 The analytics endpoints and an auth quirk

NOTE — &lsquo;Top URLs&rsquo; is accidentally auth-required

The leaderboard route declares current_user: CurrentUser = None intending ‘optional auth’, but
CurrentUser is a hard dependency that raises 401 when no token is present. The default value doesn’t make it
optional, so the endpoint actually requires authentication despite the README calling it public. To make it truly
optional you’d use a dependency that returns None when unauthenticated.

Distributed URL Shortener - Engineering Handbook Page 21 of 26

9. Cross-Cutting Concerns

9.1 middleware/request_id.py
Attaches a UUID to every request, exposes it on [Link].request_id and the X-Request-ID
response header, and binds it into the structlog context so every log line during that request carries the same
ID — the foundation of distributed tracing.

9.2 middleware/rate_limit.py
Per-IP fixed-window limiting with an atomic Redis INCR ; sets the TTL on the first request of a window; returns
429 with Retry-After and X-RateLimit-* headers when the limit is exceeded; and fails open if Redis
errors.

Distributed URL Shortener - Engineering Handbook Page 22 of 26

Distributed URL Shortener - Engineering Handbook Page 23 of 26
 The rate-limit decision, including the fail-open branch.

NOTE — &lsquo;Sliding window&rsquo; in the docstring, fixed window in the code

The module docstring is candid: the implementation is really a fixed window with a sliding reset (one counter per IP
with a TTL), not a true sliding window. The docstring even sketches the sorted-set approach a true sliding window
would need. Fine for this scope; know the distinction for interviews.

9.3 exceptions/ — domain errors and global handlers

Domain exceptions ( URLNotFoundException , AliasAlreadyExistsException , CredentialsException , etc.)
are raised deep in services and translated to HTTP responses in one central place, so handlers stay free of
try/except boilerplate and every error has a consistent JSON shape.

Each domain exception maps to a specific HTTP status; everything unexpected becomes a logged 500.

app/exceptions/[Link]

@app.exception_handler(URLNotFoundException)
async def url_not_found_handler(request, exc):
return _error_response([Link], status.HTTP_404_NOT_FOUND)

@app.exception_handler(Exception) # catch-all
async def generic_error_handler(request, exc):
[Link]("Unexpected error", exc_info=exc)
return _error_response("An unexpected error occurred", 500,
request_id=getattr([Link], "request_id", None))

SECURITY — Errors never leak internals

The catch-all logs the full exception server-side but returns a generic message to the client, with the request ID so
support can correlate. Never returning stack traces or SQL errors to clients is a basic but important security
property.

Distributed URL Shortener - Engineering Handbook Page 24 of 26

 NOTE — One defined exception is unused

RateLimitExceededException exists and has a handler, but the rate-limit middleware returns a 429
JSONResponse directly instead of raising it, so the exception is effectively dead code.

9.4 utils/request_parser.py
Extracts analytics metadata from a request: the client IP (honouring X-Forwarded-For / X-Real-IP ), a salted
one-way hash of the IP for privacy-preserving unique-visitor counting, and browser/OS/device parsed from the
User-Agent. Country is read from a CF-IPCountry header if present (a real deployment would use a GeoIP
database or a CDN header).

SECURITY — Privacy by design

Unique visitors are counted by a salted SHA-256 of the IP, not the raw IP, so you can measure reach without
persisting personal data. The salt should be rotated periodically — the code notes this.

Distributed URL Shortener - Engineering Handbook Page 25 of 26

10. The Test Suite
Tests are the executable specification and a fast feedback loop. This suite runs entirely in-memory: a fresh
SQLite database per test and a mocked Redis, so no external services are needed to run pytest .

10.1 [Link] — the fixtures that make it work

• Per-test SQLite engine using an in-memory database and a StaticPool , with tables created from
[Link] .

• Mock Redis: a plain dict standing in for Redis, with async get / setex / incr / exists so caching, rate
limiting, and the blocklist behave in tests.
• Dependency overrides: the app’s get_db and get_redis are replaced with the test doubles — the exact
mechanism Volume 1 praised layering for.
• User/auth fixtures: registered_user and auth_headers give tests a logged-in user with a bearer token
in one line.

tests/[Link]

app.dependency_overrides[get_db] = override_get_db
app.dependency_overrides[get_redis] = override_get_redis
async with AsyncClient(transport=ASGITransport(app=app), base_url="[Link] as ac:
yield ac

10.2 What is covered

File Type Covers

unit/test_security.py Unit bcrypt hashing round-trips, JWT create/decode, JTI extraction,

tamper/expiry rejection

unit/test_url_service.py Unit slug length/alphabet/uniqueness, URL hashing, short-URL

building

integration/test_auth.py Integration register/login/refresh/me happy and error paths (duplicate, weak

password, wrong password, invalid token)

integration/test_urls.py Integration create (incl. custom alias, dedup, validation), list+pagination, get,
update, delete, and redirect

integration/ Integration liveness and readiness probes

test_health.py

TIP — The tests double as usage documentation

If you’re unsure how an endpoint behaves, read its integration test — it shows the exact request shape and the
expected status code. For instance, test_create_url_deduplication confirms that shortening the same URL
twice returns the same slug.

NOTE — Where to go next

Volume 3 goes deep on the data and speed layers this volume introduced: the PostgreSQL schema and indexing,
SQLAlchemy async internals, Alembic migrations, the Redis caching strategy, and the background-task-vs-Celery
decision.

Distributed URL Shortener - Engineering Handbook Page 26 of 26