High-level design
Part 2 of 3 · URL ShortenerURL Shortener - Flask + SQLite Full Solution, Concurrency & Tests
Full Flask+SQLite URL shortener with ownership delete, TTL, atomic hit UPDATE, concurrent hit tests, HTTP status contracts.
- 1Gist
- 2Maps
- 3Q&A
- 4Sandbox
Voice readout needs Web Speech Synthesis in this browser.
Question ladder
L1
Why pin the in-memory connection?
Answer
Each new connection to :memory: is an empty database. The schema would vanish between requests.
L2
Where does CODE_TAKEN come from?
Answer
INSERT hits the UNIQUE constraint, IntegrityError becomes ValueError CODE_TAKEN, and Flask maps that to 409.
L3
What does the hit UPDATE require?
Answer
code matches, deleted=0, and expires_at is null or still in the future. rowcount must be 1.
L4
What is the concurrent test?
Answer
8 threads call resolve 50 times each. The metadata hits field must be 400 and the error list must be empty.
L5
How does delete decide 403 versus 404?
Answer
Missing or already deleted is NOT_FOUND. A wrong owner_token is FORBIDDEN. The matching owner sets deleted=1 and returns 204.
L6
What does metadata return for an expired row?
Answer
The row is still visible with expired true. Redirect of that same code is 404.
L7
What would you change for Postgres?
Answer
Drop the process lock in favor of UPDATE ... RETURNING and a transactional INSERT. Add an idempotency key on create if clients retry.
Failure modes
Closed memory connection
Closing the only :memory: connection drops the schema. The pin stays open.
Lost hit updates
A SELECT then a Python increment drops concurrent hits. The conditional UPDATE does not.
Expired redirect still 302
Forgetting the expiry predicate in the UPDATE would count and redirect a dead link.
Misconceptions
The Flask route is the concurrency control.
The route calls the store. The lock and the conditional UPDATE are the control.
404 on redirect means the row was hard-deleted.
Expiry and soft delete also return 404 on redirect. Metadata can still show an expired row.
A green single-threaded test proves the hit count.
Only the 8-thread test would catch a lost update.
Interviewer traps
Add Redis in the middle of the coding solution.
Keep this page on SQLite. Point at cache-aside and the scale-up sibling.
Swallow IntegrityError as 500.
A taken code is a client conflict: 409 and error CODE_TAKEN.
Design scenario
Same prompt for every reader.
Requirements
App factory, pinned SQLite, ownership delete, TTL, atomic hits, and an 8-thread stress test.
Traffic / scale
Threads inside one process. Hundreds of redirects against one code.
Latency
In-process SQLite. The interesting latency is lock wait, not a network RTT.
Consistency
Exactly one successful increment per successful resolve.
Availability
A failed UNIQUE insert is 409, not a 500 and not a duplicate row.
Failure assumptions
- The test client and the store share one process.
- Expiry can land between create and resolve.
Constraints
- Do not open a new :memory: connection per request.
- Do not increment hits in Python after a plain SELECT.
Prompt
Implement the shortener so a unittest file can prove status codes and concurrent hits without a network.
API
Which handler status codes does test_http_contracts assert?
Data
Which WHERE clause makes the hit update conditional?
Architecture
What does the process lock cover, and what do you refuse to add in this file?
How hits stay correct
Prefer
Conditional UPDATE under the store lock
The database adds one only while the same predicates that mean alive still hold.
- rowcount 1 is the proof.
- The concurrent test expects 400 hits.
- Expiry fails the UPDATE and the redirect is 404.
Alternative
Read the count in Python and write it back
Two threads read the same integer and the loser overwrites the winner.
- The stress test fails.
- A plain SELECT is not a lock on the row.
- SQLite can do the add in one statement.
Run the store tests and the HTTP contract test. Then break the hit UPDATE so it writes a Python variable, and confirm the 400-hit assertion fails.
Overview
Full runnable implementation of the URL shortener coding spec: URLShortenerStore on SQLite with a pinned in-memory connection for tests, Flask routes with strict status codes, ownership-checked soft delete, TTL, atomic hit increments, and a unittest suite including an 8-thread hit stress test (400 increments).
Step-by-step (implementation)
- Schema +
meta.auto_seqinitializer. base62_encodeandCODE_RE/URL_REvalidators.createcustom vs auto paths; map IntegrityError to CODE_TAKEN.resolvewith conditionalUPDATE hits=hits+1 WHERE deleted=0 AND (expires_at IS NULL OR expires_at > now).metadata,delete,stats.- Flask app factory wiring JSON errors to 400/403/404/409/201/204/302.
- Tests: unit store + HTTP contracts + concurrent hits.
Resolve under the store lock
Diagram 1. The redirect path returns the long URL only after a conditional hit update matches one row.
- 1
Acquire the store lock
One RLock wraps the transaction so the counter and the hit update do not interleave. - 2
Select the row by code
Missing rows never reach the update. - 3
Alive?
Deleted or expired rows return None. Flask turns that into 404. - 4
Conditional hit UPDATE
hits = hits + 1 only when deleted=0 and expiry is null or still in the future. - 5
rowcount equals 1?
Zero rows means the link died between the read and the write. Return None. - 6
Commit and return the long URL
The caller sends 302 Location. Metadata later shows the new hit count.
Decisions
- 1
1. Acquire store lock
- next2. BEGIN logical tx
- 2
2. BEGIN logical tx
- next3. SELECT url row by code
- 3
3. SELECT url row by code
- next4. Alive? not deleted and not expired
- ?
4. Alive? not deleted and not expired
- No5. Failure: return None / 404
- Yes6. UPDATE hits = hits + 1 with same predicates
- 5
5. Failure: return None / 404
- 6
6. UPDATE hits = hits + 1 with same predicates
- next7. rowcount equals 1?
- ?
7. rowcount equals 1?
- No5. Failure: return None / 404
- Yes8. Commit and return long_url
- 8
8. Commit and return long_url
Lesson map
URL Shortener - Flask + SQLite Full Solution, Concurrency & Tests
Full Flask+SQLite URL shortener with ownership delete, TTL, atomic hit UPDATE, concurrent hit tests, HTTP status contracts.
Architecture. Architecture
Select a node to see why it exists, or an edge to see the protocol, direction, effect, and consequence.
Mermaid export
flowchart TB a["1. Acquire store lock"] b["2. BEGIN logical tx"] c["3. SELECT url row by code"] d["4. Alive? not deleted and not expired"] e["5. Failure: return None / 404"] f["6. UPDATE hits = hits + 1 with same predicates"] g["7. rowcount equals 1?"] h["8. Commit and return long_url"] a -->|continues| b b -->|continues| c c -->|continues| d d -->|No| e d -->|Yes| f f -->|continues| g g -->|No| e g -->|Yes| h
Solution code
"""URL Shortener core: Flask + SQLite.
Sandbox: pip install flask; python app.py
Tests: python test_app.py
"""
from __future__ import annotations
import os
import re
import sqlite3
import string
import threading
import time
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Any, Optional
from flask import Flask, jsonify, redirect, request
ALPHABET = string.digits + string.ascii_letters # base62
CODE_RE = re.compile(r"^[A-Za-z0-9_-]{4,32}$")
URL_RE = re.compile(r"^https?://[^\s]{3,2048}$", re.I)
def utcnow() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def base62_encode(n: int) -> str:
if n <= 0:
return "0"
out = []
while n:
n, r = divmod(n, 62)
out.append(ALPHABET[r])
return "".join(reversed(out))
class URLShortenerStore:
"""SQLite-backed store with a process lock for read-modify-write safety."""
def __init__(self, db_path: str = ":memory:") -> None:
self.db_path = db_path
self._lock = threading.RLock()
# Keep one pinned connection so ":memory:" schema survives across opens.
self._pin: Optional[sqlite3.Connection] = None
if db_path == ":memory:":
self._pin = sqlite3.connect(":memory:", check_same_thread=False)
self._pin.row_factory = sqlite3.Row
self._init_db()
def _connect(self) -> sqlite3.Connection:
if self._pin is not None:
return self._pin
conn = sqlite3.connect(self.db_path, check_same_thread=False, timeout=30)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys=ON")
return conn
def _init_db(self) -> None:
with self._lock:
conn = self._connect()
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS urls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT NOT NULL UNIQUE,
long_url TEXT NOT NULL,
owner_token TEXT NOT NULL,
created_at TEXT NOT NULL,
expires_at TEXT,
hits INTEGER NOT NULL DEFAULT 0,
deleted INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_urls_code ON urls(code);
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value INTEGER NOT NULL
);
INSERT OR IGNORE INTO meta(key, value) VALUES ('auto_seq', 1000000);
"""
)
conn.commit()
# Do not close the pinned in-memory connection.
@contextmanager
def _tx(self):
with self._lock:
conn = self._connect()
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
if self._pin is None:
conn.close()
def create(
self,
long_url: str,
owner_token: str,
custom_code: Optional[str] = None,
ttl_seconds: Optional[int] = None,
) -> dict[str, Any]:
if not URL_RE.match(long_url or ""):
raise ValueError("INVALID_URL")
if not owner_token or len(owner_token) > 128:
raise ValueError("INVALID_OWNER")
if ttl_seconds is not None and (ttl_seconds < 1 or ttl_seconds > 365 * 24 * 3600):
raise ValueError("INVALID_TTL")
expires_at = None
if ttl_seconds is not None:
expires_at = datetime.fromtimestamp(
time.time() + ttl_seconds, tz=timezone.utc
).replace(microsecond=0).isoformat()
with self._tx() as conn:
if custom_code is not None:
if not CODE_RE.match(custom_code):
raise ValueError("INVALID_CODE")
try:
conn.execute(
"INSERT INTO urls(code,long_url,owner_token,created_at,expires_at)"
" VALUES(?,?,?,?,?)",
(custom_code, long_url, owner_token, utcnow(), expires_at),
)
code = custom_code
except sqlite3.IntegrityError:
raise ValueError("CODE_TAKEN")
else:
# Auto code from monotonic sequence -> base62; retry on rare collision.
for _ in range(8):
row = conn.execute(
"SELECT value FROM meta WHERE key='auto_seq'"
).fetchone()
seq = int(row["value"]) + 1
conn.execute(
"UPDATE meta SET value=? WHERE key='auto_seq'", (seq,)
)
code = base62_encode(seq)
try:
conn.execute(
"INSERT INTO urls(code,long_url,owner_token,created_at,expires_at)"
" VALUES(?,?,?,?,?)",
(code, long_url, owner_token, utcnow(), expires_at),
)
break
except sqlite3.IntegrityError:
continue
else:
raise RuntimeError("CODE_COLLISION")
row = conn.execute(
"SELECT code,long_url,created_at,expires_at,hits FROM urls WHERE code=?",
(code,),
).fetchone()
return dict(row)
def _alive(self, row: sqlite3.Row) -> bool:
if row is None or row["deleted"]:
return False
if row["expires_at"]:
if row["expires_at"] <= utcnow():
return False
return True
def resolve(self, code: str) -> Optional[str]:
"""Redirect path: atomic hit increment if still alive."""
with self._tx() as conn:
row = conn.execute(
"SELECT * FROM urls WHERE code=?", (code,)
).fetchone()
if not self._alive(row):
return None
# Atomic conditional UPDATE protects concurrent hit counting.
cur = conn.execute(
"UPDATE urls SET hits = hits + 1"
" WHERE code=? AND deleted=0"
" AND (expires_at IS NULL OR expires_at > ?)",
(code, utcnow()),
)
if cur.rowcount != 1:
return None
return row["long_url"]
def metadata(self, code: str) -> Optional[dict[str, Any]]:
with self._tx() as conn:
row = conn.execute(
"SELECT code,long_url,created_at,expires_at,hits,deleted FROM urls WHERE code=?",
(code,),
).fetchone()
if row is None or row["deleted"]:
return None
expired = bool(row["expires_at"] and row["expires_at"] <= utcnow())
d = dict(row)
d["expired"] = expired
d.pop("deleted", None)
return d
def delete(self, code: str, owner_token: str) -> str:
"""Returns OK | NOT_FOUND | FORBIDDEN."""
with self._tx() as conn:
row = conn.execute(
"SELECT owner_token, deleted FROM urls WHERE code=?", (code,)
).fetchone()
if row is None or row["deleted"]:
return "NOT_FOUND"
if row["owner_token"] != owner_token:
return "FORBIDDEN"
conn.execute("UPDATE urls SET deleted=1 WHERE code=?", (code,))
return "OK"
def stats(self) -> dict[str, int]:
with self._tx() as conn:
total = conn.execute(
"SELECT COUNT(*) AS c FROM urls WHERE deleted=0"
).fetchone()["c"]
active = conn.execute(
"SELECT COUNT(*) AS c FROM urls WHERE deleted=0"
" AND (expires_at IS NULL OR expires_at > ?)",
(utcnow(),),
).fetchone()["c"]
hits = conn.execute(
"SELECT COALESCE(SUM(hits),0) AS h FROM urls WHERE deleted=0"
).fetchone()["h"]
return {"total_urls": total, "active_urls": active, "total_hits": hits}
def create_app(db_path: str = ":memory:") -> Flask:
app = Flask(__name__)
store = URLShortenerStore(db_path)
app.config["STORE"] = store
@app.post("/api/v1/urls")
def create_url():
body = request.get_json(silent=True) or {}
try:
data = store.create(
long_url=body.get("url", ""),
owner_token=body.get("owner_token", ""),
custom_code=body.get("code"),
ttl_seconds=body.get("ttl_seconds"),
)
except ValueError as e:
code = str(e)
status = {
"INVALID_URL": 400,
"INVALID_OWNER": 400,
"INVALID_TTL": 400,
"INVALID_CODE": 400,
"CODE_TAKEN": 409,
}.get(code, 400)
return jsonify({"error": code}), status
except RuntimeError as e:
return jsonify({"error": str(e)}), 500
return jsonify(data), 201
@app.get("/r/<code>")
def redirect_code(code: str):
url = store.resolve(code)
if url is None:
return jsonify({"error": "NOT_FOUND"}), 404
return redirect(url, code=302)
@app.get("/api/v1/urls/<code>")
def get_meta(code: str):
meta = store.metadata(code)
if meta is None:
return jsonify({"error": "NOT_FOUND"}), 404
return jsonify(meta), 200
@app.delete("/api/v1/urls/<code>")
def delete_url(code: str):
body = request.get_json(silent=True) or {}
owner = body.get("owner_token") or request.headers.get("X-Owner-Token", "")
result = store.delete(code, owner)
if result == "NOT_FOUND":
return jsonify({"error": "NOT_FOUND"}), 404
if result == "FORBIDDEN":
return jsonify({"error": "FORBIDDEN"}), 403
return "", 204
@app.get("/api/v1/stats")
def get_stats():
return jsonify(store.stats()), 200
return app
if __name__ == "__main__":
path = os.environ.get("URL_DB", ":memory:")
create_app(path).run(host="127.0.0.1", port=int(os.environ.get("PORT", "5055")), debug=False)
Tests
"""Tests for URL Shortener Flask + SQLite service."""
from __future__ import annotations
import threading
import unittest
from app import URLShortenerStore, create_app
class StoreTests(unittest.TestCase):
def setUp(self):
self.store = URLShortenerStore(":memory:")
def test_create_auto_and_redirect_hits(self):
row = self.store.create("https://example.com/a", "own-1")
self.assertTrue(row["code"])
url = self.store.resolve(row["code"])
self.assertEqual(url, "https://example.com/a")
meta = self.store.metadata(row["code"])
self.assertEqual(meta["hits"], 1)
def test_custom_code_conflict(self):
self.store.create("https://example.com/a", "own-1", custom_code="myLink")
with self.assertRaises(ValueError) as ctx:
self.store.create("https://example.com/b", "own-2", custom_code="myLink")
self.assertEqual(str(ctx.exception), "CODE_TAKEN")
def test_invalid_url_and_code(self):
with self.assertRaises(ValueError):
self.store.create("ftp://bad", "own-1")
with self.assertRaises(ValueError):
self.store.create("https://ok.com", "own-1", custom_code="!!")
def test_ttl_expiry(self):
row = self.store.create("https://example.com/x", "own-1", ttl_seconds=1)
# Force expiry by rewriting expires_at into the past via resolve after sleep
import time
time.sleep(1.1)
self.assertIsNone(self.store.resolve(row["code"]))
meta = self.store.metadata(row["code"])
self.assertTrue(meta["expired"])
def test_delete_ownership(self):
row = self.store.create("https://example.com/a", "own-1", custom_code="delMe")
self.assertEqual(self.store.delete("delMe", "wrong"), "FORBIDDEN")
self.assertEqual(self.store.delete("delMe", "own-1"), "OK")
self.assertEqual(self.store.delete("delMe", "own-1"), "NOT_FOUND")
self.assertIsNone(self.store.resolve("delMe"))
def test_stats(self):
self.store.create("https://example.com/1", "o")
self.store.create("https://example.com/2", "o")
self.store.resolve(self.store.create("https://example.com/3", "o")["code"])
s = self.store.stats()
self.assertEqual(s["total_urls"], 3)
self.assertEqual(s["active_urls"], 3)
self.assertEqual(s["total_hits"], 1)
def test_concurrent_hits(self):
row = self.store.create("https://example.com/c", "o", custom_code="hitMe")
errors = []
def worker():
try:
for _ in range(50):
assert self.store.resolve("hitMe") == "https://example.com/c"
except Exception as e:
errors.append(e)
threads = [threading.Thread(target=worker) for _ in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
self.assertEqual(errors, [])
self.assertEqual(self.store.metadata("hitMe")["hits"], 400)
class ApiTests(unittest.TestCase):
def setUp(self):
self.app = create_app(":memory:")
self.client = self.app.test_client()
def test_http_contracts(self):
r = self.client.post(
"/api/v1/urls",
json={"url": "https://example.com/z", "owner_token": "tok", "code": "zCode"},
)
self.assertEqual(r.status_code, 201)
self.assertEqual(r.get_json()["code"], "zCode")
r = self.client.post(
"/api/v1/urls",
json={"url": "https://example.com/z", "owner_token": "tok", "code": "zCode"},
)
self.assertEqual(r.status_code, 409)
r = self.client.get("/r/zCode", follow_redirects=False)
self.assertEqual(r.status_code, 302)
self.assertEqual(r.headers["Location"], "https://example.com/z")
r = self.client.get("/api/v1/urls/zCode")
self.assertEqual(r.status_code, 200)
self.assertEqual(r.get_json()["hits"], 1)
r = self.client.delete("/api/v1/urls/zCode", json={"owner_token": "bad"})
self.assertEqual(r.status_code, 403)
r = self.client.delete("/api/v1/urls/zCode", json={"owner_token": "tok"})
self.assertEqual(r.status_code, 204)
r = self.client.get("/r/zCode")
self.assertEqual(r.status_code, 404)
r = self.client.get("/api/v1/stats")
self.assertEqual(r.status_code, 200)
self.assertIn("total_urls", r.get_json())
if __name__ == "__main__":
unittest.main(verbosity=2)
Test run (venv + Flask): Ran 8 tests in ~1.1s — OK including test_concurrent_hits (8 threads x 50 resolves = 400 hits).
Concepts used, learn more
Read the underlying idea on its own study page. This lesson applies it. It does not replace those pages.
- API Design — Naming, Paths, Routing & Contracts
- API Idempotency Keys
- Rate Limiting: Token Bucket, Leaky Bucket & Sliding Window
- Mutexes, Condition Variables, Deadlocks & Happens-Before
- Mutex vs RWLock
- Atomics vs Locks
- Redis Cache-Aside, Invalidation & Stampede Prevention
- Consistent Hashing: Rings, Virtual Nodes & Replica Placement
- Database Sharding & Partitioning — Keys, Hotspots & Rebalancing
- Partition Strategies — Range, Hash, List & Composite
- CDNs, Cache Hierarchy & Origin Shielding
- Load Balancing — L4 vs L7, Algorithms & Health Checks
- Negative Caching
- Rendezvous Hashing (HRW): Highest Random Weight
- Hashing, Frequency Maps & Counting — Two Sum Family & Anagrams
| Concept | How the code uses it |
|---|---|
| UNIQUE + IntegrityError | Custom code conflict -> 409 |
| Conditional UPDATE | Hit increment without lost updates |
| Soft delete flag | Delete without reclaiming row immediately |
| App factory | Inject store/db path for tests |
| Ownership token | Delete authorization |
Interview Q&A
What does the HTTP contract test cover in one method?
Answer
201 on create, 409 on the same code, 302 with Location, metadata hits 1, 403 then 204 on delete, 404 on a later redirect, and 200 stats that include total_urls.
Why is the app an app factory?
Answer
create_app(db_path) injects the store so tests can pass :memory: and production can pass a file path. The suite does not bind a port.
When would you move hits off the request?
Answer
When redirect QPS makes the write on the hot path the bottleneck. Enqueue the hit and accept a slightly stale count. That tradeoff is the scale-up page, with rate limiting if create is being abused.
Why does delete return 204 with an empty body?
Answer
The contract says the row is gone from the caller's point of view. 204 has no JSON. A wrong owner is 403 with FORBIDDEN, and a missing code is 404 with NOT_FOUND.
What does the TTL test force?
Answer
It creates a link with ttl_seconds 1, sleeps just past that second, and asserts resolve returns None while metadata still shows expired true.
Which concept pages explain the pieces this file uses?
Answer
HTTP status and resource shape are API design. Retrying create safely is idempotency keys. The process lock is mutexes. The rest of the read path is the scale-up page.
Pitfalls
- Closing the only
:memory:connection between requests. - Incrementing hits in Python after a plain SELECT (race).
- Forgetting to treat expired rows as 404 on redirect while still returning metadata with
expired: true.
Interview follow-ups
- Swap SQLite for Postgres: replace lock with
UPDATE ... RETURNINGand transactionalINSERT. - Add idempotency key on create for client retries.
- Move hit increments to an async queue when QPS spikes (accept slightly stale counts).
Go Deeper
Related
The series pager also walks these pages.