Chapter 6 · Practice and Capstone
Practice and Interview Lab
- Page 21 of 22
- 63 min read
Interviews for backend, ML and AI engineering roles rarely ask you to recite definitions. They hand you a broken repository, a request that fails, an endpoint that leaks data or a vague product idea, and they watch how you think: what you check first, what you ask, which trade-offs you name and how you prove the fix works. Real work is the same. This page is a lab of graded challenges across everything in this tutorial, each with a hint and an explained solution that you can run.
Try each challenge before reading the solution. Speak your reasoning out loud, as you would in an interview. Every Git challenge runs in its own scratch repository, every API challenge runs against the BitByte Practice API at http://localhost:8800 (start it in a second terminal with fastapi run practice_api.py --port 8800), and every FastAPI fix is proved with TestClient or pytest.
What you will learn
- How to approach a debugging or design question the way interviewers expect: clarify, reproduce, fix, prove, explain the trade-off
- Git rescue skills under pressure: restore, reset, conflicts, rebase, reflog, secrets in history, reviewing a diff
- HTTP and API skills: methods, status codes, broken JSON, pagination, 429, timeouts and retries, idempotency, webhook signatures, SSE
- FastAPI and security fixes you can prove with a test: validation, error responses, blocking calls, session leaks, mocking, auth, BOLA, SQL injection, CORS, size limits, SSRF, log redaction
- Eight system-design mini interviews with a structured answer for each
How the lab works
Each challenge has a difficulty and points: Easy (1 point), Medium (2), Hard (3). Give yourself the points only if your answer would also survive the follow-up question in the solution. A scoring guide is at the end.
| Step | What you do | What the interviewer hears |
|---|---|---|
| 1. Clarify | Restate the problem, ask about scale, users and constraints | You don't solve the wrong problem |
| 2. Reproduce | Make the failure happen on purpose, with a command or a test | You work from evidence, not guesses |
| 3. Fix | The smallest correct change first | You keep risk low |
| 4. Prove | Run the same command or test again, plus an edge case | You know it works, not hope |
| 5. Explain | Name the trade-off and what you would do at larger scale | You think beyond the snippet |
Set up the lab
One folder and one virtual environment for the whole page. uv first, with the pip equivalent in the comment:
mkdir -p ~/projects/lab && cd ~/projects/lab
uv venv -q && source .venv/bin/activate
uv pip install -q "fastapi[standard]" httpx2 pytest respx sqlalchemy pyjwt
# pip: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]" httpx2 pytest respx sqlalchemy pyjwt
curl -s http://localhost:8800/health{"status":"ok"}If the last line fails, the Practice API is not running: start it in another terminal and try again.
Part 1: Git challenges
G1. Recover a deleted file and an old version (Easy)
Task: you deleted prompts.txt by accident and broke config.py with a bad edit. Get both back. Then bring back the version of config.py from one commit earlier, when the model was still 'small'.
mkdir -p ~/projects/lab/git && cd ~/projects/lab/git
git init -q recover && cd recover
echo "MODEL = 'small'" > config.py
echo "Answer politely and briefly." > prompts.txt
git add . && git commit -qm "Add config and prompts"
echo "MODEL = 'large'" > config.py
git commit -qam "Use the large model"
rm prompts.txt
echo "MODEL = ???" > config.py
git status --short M config.py
D prompts.txtHint: both files still exist in the index and in the last commit. One command restores from the index; an option picks another commit as the source.
Solution:
git restore prompts.txt config.py
git status --short
cat config.py
git restore --source=HEAD~1 config.py
cat config.py
git status --shortMODEL = 'large'
MODEL = 'small'
M config.pyWhy it works: git restore <file> copies the file from the index (the staging area) back into the working tree, which undoes unstaged edits and deletions. --source=HEAD~1 takes it from an older commit instead, and the change then shows as an ordinary modification you can commit or throw away. If the deletion had been staged (git rm), you would add --staged --worktree. Follow-up: "Can Git recover a file you never committed?" No; that is why you commit small and often.
G2. Undo the last commit without losing work (Easy)
Task: you committed "wip" with a debug print in it. Undo the commit but keep every change so you can fix it and commit properly.
cd ~/projects/lab/git && git init -q undo && cd undo
echo "def predict(x): return x * 2" > model.py
git add model.py && git commit -qm "Add predict"
echo "print('debug')" >> model.py
echo "TIMEOUT_SECONDS = 10" > settings.py
git add . && git commit -qm "wip"
git log --oneline532fa3f wip
43cb677 Add predictHint: git reset moves the branch pointer; the mode decides what happens to the index and the files.
Solution:
git reset --soft HEAD~1
git log --oneline
git status --short43cb677 Add predict
M model.py
A settings.pyWhy it works: --soft moves main back one commit and leaves the changes staged. The default --mixed would leave them unstaged; --hard would delete them. Now remove the debug line, git add and commit with a real message. Follow-up: "And if the commit was already pushed?" Then don't rewrite shared history: use git revert <hash>, which adds a new commit that undoes it.
G3. Resolve a merge conflict (Medium)
Task: two branches changed the same timeout. Merge them and resolve the conflict so that both intentions survive.
cd ~/projects/lab/git && git init -q conflict && cd conflict
printf 'TIMEOUT = 10\nRETRIES = 3\n' > settings.py
git add . && git commit -qm "Add settings"
git switch -qc faster-timeouts
printf 'TIMEOUT = 5\nRETRIES = 3\n' > settings.py
git commit -qam "Lower timeout to 5 s for the weather API"
git switch -q main
printf 'TIMEOUT = 30\nRETRIES = 3\n' > settings.py
git commit -qam "Raise timeout to 30 s for slow LLM calls"
git merge faster-timeouts
cat settings.pyAuto-merging settings.py
CONFLICT (content): Merge conflict in settings.py
Automatic merge failed; fix conflicts and then commit the result.
<<<<<<< HEAD
TIMEOUT = 30
=======
TIMEOUT = 5
>>>>>>> faster-timeouts
RETRIES = 3Hint: a conflict is a question, not an error. Read both commit messages: are the two changes really incompatible?
Solution: they are not. One person wanted a short timeout for a fast API, the other a long one for LLM calls, so the right answer is two settings, not a winner.
printf 'WEATHER_TIMEOUT = 5\nLLM_TIMEOUT = 30\nRETRIES = 3\n' > settings.py
git add settings.py
git commit -q --no-edit
git log --oneline --graph* f971929 Merge branch 'faster-timeouts'
|\
| * a9baa1c Lower timeout to 5 s for the weather API
* | 1cff844 Raise timeout to 30 s for slow LLM calls
|/
* 58777fa Add settingsWhy it works: Git marks the two versions between <<<<<<<, ======= and >>>>>>>. You write the final content, remove every marker, git add to mark it resolved and commit. git merge --abort gets you out if you need to start again. Follow-up: "How do you avoid conflicts?" Small, short-lived branches, merging main in often, and talking to the other author. Here, renaming a setting also means updating every place that reads it, and the tests should catch any you miss.
G4. Rebase a feature branch onto main (Medium)
Task: main moved on while you worked on feature/cache. Make your branch start from the current main so the history is a straight line.
cd ~/projects/lab/git && git init -q rebase && cd rebase
echo "# Weather proxy" > README.md && git add . && git commit -qm "Initial commit"
git switch -qc feature/cache
echo "CACHE_TTL = 300" > cache.py && git add . && git commit -qm "Add cache settings"
echo "def get(key): ..." >> cache.py && git commit -qam "Add cache get"
git switch -q main
echo "fastapi" > requirements.txt && git add . && git commit -qm "Add requirements"
git log --oneline --graph --all* 1d9df1f Add cache get
* 0166514 Add cache settings
| * 3a97ca8 Add requirements
|/
* a5440c5 Initial commitHint: switch to the feature branch first, then replay its commits on top of main.
Solution:
git switch -q feature/cache
git rebase -q main
git log --oneline --graph --all* 6d67bcf Add cache get
* 4f0f149 Add cache settings
* 3a97ca8 Add requirements
* a5440c5 Initial commitWhy it works: rebase replays your two commits on top of main's newest commit. Look at the hashes: they are new commits with the same changes, because a commit's hash includes its parent. That is why the golden rule is: rebase only commits nobody else has. If you had already pushed the branch, you would push with git push --force-with-lease, which refuses if someone else pushed in the meantime. If a rebase stops on a conflict, fix it, git add, then git rebase --continue (or --abort). Follow-up: "Merge or rebase?" Rebase your own branch to keep it current; merge (or squash-merge) the pull request into main; never rebase main itself.
G5. Recover a "lost" commit with reflog (Medium)
Task: a teammate ran git reset --hard HEAD~1 and the commit with the model metrics is gone from git log. Get it back.
cd ~/projects/lab/git && git init -q reflog && cd reflog
echo "v1" > model_card.md && git add . && git commit -qm "Model card v1"
echo "v2: accuracy 0.91 on the test set" > model_card.md && git commit -qam "Model card v2 with metrics"
git reset -q --hard HEAD~1
git log --oneline
cat model_card.mdc8c6d84 Model card v1
v1Hint: Git records every move of HEAD in a local journal, even moves that "lose" commits.
Solution:
git reflog -3
git branch rescue 'HEAD@{1}'
git reset -q --hard rescue
git log --oneline
cat model_card.mdc8c6d84 HEAD@{0}: reset: moving to HEAD~1
1144601 HEAD@{1}: commit: Model card v2 with metrics
c8c6d84 HEAD@{2}: commit (initial): Model card v1
1144601 Model card v2 with metrics
c8c6d84 Model card v1
v2: accuracy 0.91 on the test setWhy it works: the commit was never deleted, only unreferenced. git reflog shows where HEAD was; HEAD@{1} is "one move ago". Putting a branch on it makes it safe again. (The quotes stop some shells, such as PowerShell, from treating the braces specially.) Follow-up: "Does this work for a teammate's commit?" No: the reflog is local and expires (by default unreachable entries after 30 days, others after 90). For commits that were pushed, ask the remote or a teammate's clone.
G6. Find a secret in history (Hard)
Task: someone committed an API key and "fixed" it in the next commit. Prove the key is still in the history, find every commit that contains it, and describe the full response.
cd ~/projects/lab/git && git init -q secrets && cd secrets
echo "import os" > app.py && git add . && git commit -qm "Start app"
echo 'LLM_API_KEY = "bb-live-9f8e7d6c5b4a"' >> app.py && git commit -qam "Call the LLM"
printf 'import os\nLLM_API_KEY = os.environ["LLM_API_KEY"]\n' > app.py && git commit -qam "Read the key from the environment"
grep -c "bb-live-" app.py0Hint: grep only sees the working tree. git log -S finds commits that add or remove a string; git grep can search any list of commits.
Solution:
git log --oneline -S "bb-live-"
git grep -n "bb-live-" $(git rev-list --all)57718d6 Read the key from the environment
3b7c6f7 Call the LLM
3b7c6f72d9e545b3f90afd4b8a371bd2217e3a9e:app.py:2:LLM_API_KEY = "bb-live-9f8e7d6c5b4a"The key is in the history, and anyone with a clone has it. The response, in this order:
- Revoke or rotate the key at the provider first. A pushed secret must be treated as stolen; bots scan public GitHub within minutes. Check the provider's usage logs for misuse.
- Put the new key only in the environment or a secret manager (it already reads
os.environ) and make sure.envis in.gitignore. - Rewrite history if the repository is shared or public, on a fresh clone, then force-push and ask everyone to re-clone:
pip install git-filter-repo
echo 'bb-live-9f8e7d6c5b4a==>REMOVED' > ../replacements.txt
git filter-repo --replace-text ../replacements.txt
git push --force-with-lease --all4. Prevent the next one: keep GitHub secret scanning and push protection on, and add a scanner such as gitleaks to pre-commit and CI.
Follow-up: "Why rotate first if you rewrite history anyway?" Because forks, clones, caches and CI logs already have the old history; rewriting only stops new copies. Step 3 is tidying up, step 1 is the fix. (Git for AI Projects: Data, Models, LFS, DVC and Leaked Secrets covers this in depth.)
G7. Review a pull request diff (Medium)
Task: a teammate asks you to review feature/forecast. Look at exactly what the branch changes compared with main, and write the review.
cd ~/projects/lab/git && git init -q review && cd review
cat > client.py <<'EOF'
import requests
def get_weather(city):
r = requests.get("https://api.example.com/v1/weather", params={"city": city}, timeout=5)
r.raise_for_status()
return r.json()
EOF
git add . && git commit -qm "Add weather client"
git switch -qc feature/forecast
cat > client.py <<'EOF'
import requests
API_KEY = "bb-live-1a2b3c4d"
def get_weather(city):
url = "https://api.example.com/v1/weather?city=" + city
r = requests.get(url, headers={"X-API-Key": API_KEY})
print("weather response:", r.text, API_KEY)
return r.json()
EOF
git commit -qam "Add API key and logging"
git switch -q mainHint: git diff main...feature/forecast (three dots) shows only what the branch added since it split from main, which is what a pull request page shows.
Solution:
git diff --stat main...feature/forecast
git diff main...feature/forecast client.py | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
diff --git a/client.py b/client.py
index e7c432b..f22a4b1 100644
--- a/client.py
+++ b/client.py
@@ -1,7 +1,10 @@
import requests
+API_KEY = "bb-live-1a2b3c4d"
+
def get_weather(city):
- r = requests.get("https://api.example.com/v1/weather", params={"city": city}, timeout=5)
- r.raise_for_status()
+ url = "https://api.example.com/v1/weather?city=" + city
+ r = requests.get(url, headers={"X-API-Key": API_KEY})
+ print("weather response:", r.text, API_KEY)
return r.json()| Line | Comment | Severity |
|---|---|---|
API_KEY = "bb-live-…" | A live secret in source. Rotate it now, then load it from settings/environment. | Blocker |
print(… API_KEY) | Logs the secret and the full response body. Use logging, log status and timing only. | Blocker |
requests.get(url, …) | The timeout was removed: one slow upstream call can hang a worker forever. Put timeout=5 back. | Blocker |
"…?city=" + city | Builds the query by hand: "Cox's Bazar" or & break it, and users can inject extra parameters. Use params=. | Must fix |
removed raise_for_status() | A 401 or 500 now reaches r.json() and fails confusingly or returns an error body as data. | Must fix |
| Commit message | "Add API key and logging" describes the how, not the why; there is no test. | Nit |
Follow-up: "How do you keep a review kind and useful?" Explain why for each comment, separate blockers from nits, suggest the fix, and praise what is good. Approve only when blockers are fixed and CI is green.
Part 2: HTTP and API challenges
H1. Choose the method and status code (Easy)
Task: for each operation of a chat product, choose the HTTP method, a path and the success status code.
Hint: ask two questions for each: does it change state (safe or not)? Does repeating it change the result again (idempotent or not)?
Solution:
| Operation | Method and path | Success | Note |
|---|---|---|---|
| Read a conversation's messages | GET /v1/conversations/{id}/messages | 200 | Safe, cacheable |
| Start a new conversation | POST /v1/conversations | 201 + Location | Not idempotent: add an Idempotency-Key |
| Rename a conversation | PATCH /v1/conversations/{id} | 200 | Partial update |
| Replace a model's whole config | PUT /v1/models/{name}/config | 200 (201 if created) | Idempotent: same body, same result |
| Delete an uploaded document | DELETE /v1/documents/{id} | 204 | Idempotent in effect; a repeat may return 404 |
| Embed 10,000 documents | POST /v1/embedding-jobs | 202 + status URL | Long-running: poll or webhook |
| Search with a large filter | POST /v1/documents/search | 200 | POST because the body is big; it is not cacheable by default |
| Check whether a resource changed | GET with If-None-Match | 304 | No body when unchanged |
You can see a method arrive at the server with the Practice API's echo endpoint:
curl -s -X PATCH http://localhost:8800/v1/echo -H "Content-Type: application/json" -d '{"title": "Exam prep"}'{"method":"PATCH","query":{},"headers":{"user-agent":"curl/8.5.0","accept":"*/*","content-type":"application/json"},"json":{"title":"Exam prep"}}H2. Interpret status codes (Easy)
Task: run these requests. For each status, say what it means, whose fault it is, and whether a client should retry.
cd ~/projects/lab
for path in /v1/courses/3 /v1/courses/99 /v1/weather /v1/me /v1/status/503; do
printf '%-16s %s\n' "$path" "$(curl -s -o /dev/null -w '%{http_code}' "http://localhost:8800$path")"
done
curl -s -o /dev/null -w 'POST no key %{http_code}\n' -X POST http://localhost:8800/v1/reviews \
-H "Content-Type: application/json" -d '{"course_id": 2, "rating": 5, "comment": "Clear and useful"}'
curl -s -o /dev/null -w 'POST wrong key %{http_code}\n' -X POST http://localhost:8800/v1/reviews \
-H "X-API-Key: wrong" -H "Content-Type: application/json" -d '{"course_id": 2, "rating": 5, "comment": "Clear and useful"}'
for i in 1 2 3 4; do curl -s -o /dev/null -w '%{http_code} ' -H "X-Client-Id: lab-h2" http://localhost:8800/v1/limited; done; echo/v1/courses/3 200
/v1/courses/99 404
/v1/weather 422
/v1/me 401
/v1/status/503 503
POST no key 401
POST wrong key 403
200 200 200 429Hint: 4xx means "the client must change something"; 5xx means "the server failed, the same request may work later".
Solution:
| Status | Meaning here | Fault | Retry? |
|---|---|---|---|
| 200 | Course 3 found | — | — |
| 404 | Course 99 does not exist | Client | No |
| 422 | city query parameter missing (validation) | Client | No: fix the request |
| 401 | No credentials (no bearer token, no API key) | Client | Only after getting a token |
| 403 | Credentials understood but not allowed (wrong key) | Client | No |
| 429 | Rate limit: the fourth call within 10 s | Client (too fast) | Yes, after Retry-After |
| 503 | Service temporarily unavailable | Server | Yes, with backoff |
Follow-up: "401 or 403?" 401 means "who are you?" (and should carry WWW-Authenticate); 403 means "I know who you are, and no". For another user's private resource many APIs return 404 so they do not reveal that it exists.
H3. Debug a malformed JSON request (Easy)
Task: these three requests try to create a review and all fail. Read each error and fix the request.
curl -s -X POST http://localhost:8800/v1/reviews -H "X-API-Key: practice-key-123" \
-H "Content-Type: application/json" -d '{"course_id": 2, "rating": 5, "comment": "Clear and useful",}'
echo
curl -s -X POST http://localhost:8800/v1/reviews -H "X-API-Key: practice-key-123" \
-d '{"course_id": 2, "rating": 5, "comment": "Clear and useful"}'
echo
curl -s -X POST http://localhost:8800/v1/reviews -H "X-API-Key: practice-key-123" \
-H "Content-Type: application/json" -d '{"course_id": "two", "rating": 6, "comment": "Clear and useful"}'
echo{"detail":[{"type":"json_invalid","loc":["body",60],"msg":"JSON decode error","input":{},"ctx":{"error":"Expecting property name enclosed in double quotes"}}]}
{"detail":[{"type":"model_attributes_type","loc":["body"],"msg":"Input should be a valid dictionary or object to extract fields from","input":"{\"course_id\": 2, \"rating\": 5, \"comment\": \"Clear and useful\"}"}]}
{"detail":[{"type":"int_parsing","loc":["body","course_id"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"two"},{"type":"less_than_equal","loc":["body","rating"],"msg":"Input should be less than or equal to 5","input":6,"ctx":{"le":5}}]}Hint: look at type and loc in each error: a number in loc is a character position in the body.
Solution: (1) json_invalid at character 60: a trailing comma, which JSON does not allow. (2) model_attributes_type with the whole body as a string: there is no Content-Type: application/json, so curl sent it as a form and the server did not parse it as JSON. (3) Valid JSON, but the values break the schema: course_id is not an integer and rating is above 5. The fix is valid JSON, the header and valid values; curl's --json option sets both Content-Type and Accept for you:
curl -s -X POST http://localhost:8800/v1/reviews -H "X-API-Key: practice-key-123" \
--json '{"course_id": 2, "rating": 5, "comment": "Clear and useful"}'{"id":1,"course_id":2,"rating":5,"comment":"Clear and useful"}Follow-up: "It works in bash but not in PowerShell." PowerShell treats quotes differently; put the JSON in a file and send it with --json @review.json, or build the request in Python with json=, which can never produce invalid JSON.
H4. Handle pagination (Medium)
Task: collect all courses from /v1/courses, four per page, with httpx. Don't hard-code the number of pages, and don't loop forever if the server misbehaves.
Hint: follow next_page until it is null, and cap the number of pages.
Solution:
import httpx
def fetch_all_courses(client: httpx.Client, level: str | None = None, max_pages: int = 50) -> list[dict]:
items, page = [], 1
for _ in range(max_pages):
params = {"page": page, "per_page": 4}
if level:
params["level"] = level
r = client.get("/v1/courses", params=params)
r.raise_for_status()
data = r.json()
items.extend(data["items"])
if data["next_page"] is None:
return items
page = data["next_page"]
raise RuntimeError(f"more than {max_pages} pages: is the server looping?")
with httpx.Client(base_url="http://localhost:8800", timeout=5) as client:
courses = fetch_all_courses(client)
advanced = fetch_all_courses(client, level="advanced")
print(len(courses), [c["id"] for c in courses])
print([c["title"] for c in advanced])10 [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
['RAG Systems', 'AI Agents', 'LLMOps']Why it works: the loop trusts the server's next_page instead of computing pages, reuses one client (one connection pool), checks every status and stops after a safe maximum. Follow-up: "What if rows are inserted while you page?" With offset pagination (page/per_page) items can be skipped or repeated, because positions shift. Cursor pagination (?after=<last id>) is stable, which is why most large APIs use it (Designing Robust APIs: Pagination, Rate Limits, Retries and Versioning).
H5. Recover from 429 Too Many Requests (Medium)
Task: /v1/limited allows 3 requests per 10 seconds per client. Make four calls in a row so that the fourth one succeeds, without hammering the server.
Hint: the 429 response tells you how long to wait.
Solution:
import time
import httpx
def get_respecting_limits(client: httpx.Client, path: str, max_attempts: int = 3, max_wait: int = 30) -> httpx.Response:
for attempt in range(1, max_attempts + 1):
r = client.get(path)
if r.status_code != 429:
return r
wait = min(int(r.headers.get("Retry-After", "1")), max_wait)
print(f" attempt {attempt}: 429, waiting {wait} s as Retry-After says")
time.sleep(wait)
r.raise_for_status()
return r
with httpx.Client(base_url="http://localhost:8800", headers={"X-Client-Id": "lab-h5"}, timeout=5) as client:
for call in range(1, 5):
r = get_respecting_limits(client, "/v1/limited")
print(f"call {call}: {r.status_code} {r.json()}")call 1: 200 {'ok': True, 'remaining': 2}
call 2: 200 {'ok': True, 'remaining': 1}
call 3: 200 {'ok': True, 'remaining': 0}
attempt 1: 429, waiting 10 s as Retry-After says
call 4: 200 {'ok': True, 'remaining': 2}Why it works: the client treats 429 as "slow down", not as an error, waits exactly as long as the server asks, caps the wait so a bad header cannot freeze it for an hour, and gives up after a few attempts. Retry-After can also be an HTTP date; a robust client handles both (that is the Medium exercise at the end). Follow-up: "How do you avoid 429 in the first place?" Throttle on the client (a token bucket or a semaphore), batch requests, cache responses, and read the provider's rate-limit headers before you hit the wall.
H6. Timeout plus retry with backoff (Medium)
Task: write a GET helper that retries 503 and network errors with exponential backoff and jitter, and gives up cleanly when the server is too slow.
Hint: set the timeout on the client; catch httpx.TransportError (timeouts are a subclass); retry only what can succeed later.
Solution:
import random
import time
import httpx
RETRY_STATUSES = {429, 502, 503, 504}
def get_with_retries(client: httpx.Client, path: str, attempts: int = 4, base_delay: float = 0.2) -> httpx.Response:
for attempt in range(1, attempts + 1):
try:
r = client.get(path)
if r.status_code not in RETRY_STATUSES:
return r
reason = str(r.status_code)
except httpx.TransportError as exc:
reason = type(exc).__name__
if attempt == attempts:
raise RuntimeError(f"gave up after {attempts} attempts (last: {reason})")
print(f" attempt {attempt} failed ({reason}), retrying")
time.sleep(base_delay * 2 ** (attempt - 1) * random.uniform(0.5, 1.0))
timeout = httpx.Timeout(5.0, read=0.5)
with httpx.Client(base_url="http://localhost:8800", timeout=timeout) as client:
r = get_with_retries(client, "/v1/flaky?key=lab-h6&fail_times=2")
print(r.status_code, r.json())
try:
get_with_retries(client, "/v1/slow?seconds=2", attempts=2)
except RuntimeError as exc:
print(exc) attempt 1 failed (503), retrying
attempt 2 failed (503), retrying
200 {'ok': True, 'attempt': 3}
attempt 1 failed (ReadTimeout), retrying
gave up after 2 attempts (last: ReadTimeout)Why it works: the delay doubles each time (0.2, 0.4, 0.8 s…) and the random factor (jitter) stops thousands of clients from retrying in lockstep. The read timeout of 0.5 s turns a hanging server into a fast, clear failure. Follow-up: "Would you retry a POST that creates a payment?" Only with an idempotency key (next challenge); otherwise a retry after a timeout can charge twice, because the first request may have succeeded on the server.
H7. Design an idempotent create endpoint (Hard)
Task: POST /v1/payments must never charge twice, even if the client retries after a timeout. Implement an Idempotency-Key header and prove it with tests.
Hint: store the key with a fingerprint of the request body and the response you sent. Same key and same body: replay. Same key and a different body: refuse.
Solution:
import hashlib
import json
from typing import Annotated
from fastapi import FastAPI, Header, HTTPException, Response
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field
class PaymentIn(BaseModel):
user_id: int
amount_bdt: Annotated[int, Field(gt=0)]
app = FastAPI()
PAYMENTS: list[dict] = []
IDEMPOTENCY: dict[str, tuple[str, dict]] = {} # key -> (body fingerprint, response); a DB table or Redis in production
@app.post("/v1/payments", status_code=201)
def create_payment(payment: PaymentIn, response: Response,
idempotency_key: Annotated[str | None, Header()] = None) -> dict:
if not idempotency_key:
raise HTTPException(400, "Idempotency-Key header is required")
fingerprint = hashlib.sha256(json.dumps(payment.model_dump(), sort_keys=True).encode()).hexdigest()
if idempotency_key in IDEMPOTENCY:
saved_fingerprint, saved_response = IDEMPOTENCY[idempotency_key]
if saved_fingerprint != fingerprint:
raise HTTPException(422, "Idempotency-Key was already used with a different body")
response.headers["Idempotent-Replayed"] = "true"
return saved_response
result = {"id": len(PAYMENTS) + 1, **payment.model_dump(), "status": "charged"}
PAYMENTS.append(result)
IDEMPOTENCY[idempotency_key] = (fingerprint, result)
return result
client = TestClient(app)
body = {"user_id": 7, "amount_bdt": 500}
first = client.post("/v1/payments", json=body, headers={"Idempotency-Key": "pay-7-0001"})
retry = client.post("/v1/payments", json=body, headers={"Idempotency-Key": "pay-7-0001"})
misuse = client.post("/v1/payments", json={"user_id": 7, "amount_bdt": 900}, headers={"Idempotency-Key": "pay-7-0001"})
missing = client.post("/v1/payments", json=body)
print(first.status_code, first.json())
print(retry.status_code, retry.json(), retry.headers.get("idempotent-replayed"))
print(misuse.status_code, misuse.json())
print(missing.status_code, missing.json())
print("payments stored:", len(PAYMENTS))201 {'id': 1, 'user_id': 7, 'amount_bdt': 500, 'status': 'charged'}
201 {'id': 1, 'user_id': 7, 'amount_bdt': 500, 'status': 'charged'} true
422 {'detail': 'Idempotency-Key was already used with a different body'}
400 {'detail': 'Idempotency-Key header is required'}
payments stored: 1Why it works: the retry got the same 201 and the same payment ID, and only one payment exists. Reusing the key for a different request is a client bug, so it gets 422 (the status the IETF Idempotency-Key draft suggests). In production: store keys in the database in the same transaction as the payment, with a unique constraint on the key (or Redis SET key value NX EX 86400), expire them after a day or so, scope them per user, and return 409 while the first request with that key is still running. Follow-up: "Why not just check for a duplicate payment?" Two identical payments can be legitimate (two coffees); only the client knows whether a request is a retry, and the key says so.
H8. Verify a webhook signature (Medium)
Task: a payment provider sends webhooks with headers X-Webhook-Timestamp and X-Webhook-Signature: sha256=<hex>, where the hex is an HMAC-SHA256 of "<timestamp>." + raw body with a shared secret. This verifier rejects valid webhooks. Find the bug, then write a correct verifier.
import hashlib
import hmac
import json
SECRET = b"whsec_lab_secret"
def sign(body: bytes, timestamp: int, secret: bytes = SECRET) -> str:
message = str(timestamp).encode() + b"." + body
return "sha256=" + hmac.new(secret, message, hashlib.sha256).hexdigest()
def verify_buggy(body: bytes, timestamp: str, signature: str) -> bool:
data = json.loads(body)
expected = sign(json.dumps(data).encode(), int(timestamp))
return signature == expected
raw = b'{"event":"payment.succeeded","id":"evt_1001"}'
sig = sign(raw, 1768449600)
print(sig)
print("buggy verifier accepts a valid webhook:", verify_buggy(raw, "1768449600", sig))sha256=84f25107ca3bcdbc7ce6aea379019edcb0454083e86fd619b3c96a34b30d563f
buggy verifier accepts a valid webhook: FalseHint: print json.dumps(json.loads(raw)) and compare it with raw, byte by byte.
Solution: the buggy version re-serialises the JSON, which adds spaces after : and ,, so the bytes differ and so does the HMAC. Always sign and verify the raw bytes exactly as received. It also compares with ==, which can leak timing information, and never checks the timestamp, so an attacker can replay an old webhook forever.
import time
def verify(body: bytes, timestamp: str, signature: str, *, now: float | None = None,
tolerance: int = 300, secret: bytes = SECRET) -> bool:
try:
ts = int(timestamp)
except ValueError:
return False
now = time.time() if now is None else now
if abs(now - ts) > tolerance:
return False
expected = sign(body, ts, secret)
return hmac.compare_digest(expected, signature)
now = 1768449700 # 100 seconds after the webhook was signed
print("valid: ", verify(raw, "1768449600", sig, now=now))
print("tampered body: ", verify(raw.replace(b"1001", b"1002"), "1768449600", sig, now=now))
print("wrong secret: ", verify(raw, "1768449600", sign(raw, 1768449600, b"other"), now=now))
print("too old: ", verify(raw, "1768449600", sig, now=now + 3600))
print("bad timestamp: ", verify(raw, "yesterday", sig, now=now))valid: True
tampered body: False
wrong secret: False
too old: False
bad timestamp: FalseFollow-up: "Why include the timestamp in the signed message?" So the attacker cannot attach a fresh timestamp to an old signature: changing it breaks the HMAC. The time window stops old replays; deduplicating by event ID (challenge S7) stops replays inside the window.
H9. Debug a broken SSE stream (Hard)
Task: the browser's EventSource shows nothing from this streaming endpoint, and a Python client that parses Server-Sent Events gets zero events. Find out why and fix it.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient
def parse_sse(text: str) -> list[str]:
"""A minimal SSE parser: an event is one or more data: lines ended by a blank line."""
events, data = [], []
for line in text.splitlines():
if line.startswith("data:"):
data.append(line[5:].strip())
elif line == "" and data:
events.append("\n".join(data))
data = []
return events
broken = FastAPI()
@broken.get("/stream")
async def stream_broken():
async def words():
for word in ["Hello", "from", "BitByte"]:
yield f"data: {word}\n"
return StreamingResponse(words(), media_type="text/plain")
r = TestClient(broken).get("/stream")
print(r.headers["content-type"])
print(repr(r.text))
print("events:", parse_sse(r.text))text/plain; charset=utf-8
'data: Hello\ndata: from\ndata: BitByte\n'
events: []Hint: compare the bytes with the SSE format: what ends an event, and what content type does EventSource require?
Solution: two bugs. Each event must end with a blank line (\n\n), but here every line ends with a single \n, so there is never a complete event. And the content type must be text/event-stream; browsers refuse anything else. FastAPI's built-in SSE support gets both right, encodes data as JSON and adds headers that stop proxies from buffering:
from collections.abc import AsyncIterable
from fastapi.sse import EventSourceResponse, ServerSentEvent
fixed = FastAPI()
@fixed.get("/stream", response_class=EventSourceResponse)
async def stream_fixed() -> AsyncIterable[dict | ServerSentEvent]:
for word in ["Hello", "from", "BitByte"]:
yield {"delta": word}
yield ServerSentEvent(data={"done": True}, event="done")
r = TestClient(fixed).get("/stream")
print(r.headers["content-type"], "|", r.headers["cache-control"], "|", r.headers["x-accel-buffering"])
print("events:", parse_sse(r.text))text/event-stream; charset=utf-8 | no-cache | no
events: ['{"delta":"Hello"}', '{"delta":"from"}', '{"delta":"BitByte"}', '{"done": true}']The client side matters too. Stream the response instead of waiting for all of it, and parse it line by line. Here against the Practice API's fake LLM:
import json
import httpx
with httpx.Client(base_url="http://localhost:8800", timeout=httpx.Timeout(5.0, read=30.0)) as client:
token = client.post("/v1/token", json={"username": "nadia", "password": "s3cret-pass"}).json()["access_token"]
request = {"messages": [{"role": "user", "content": "What is an idempotency key?"}], "stream": True}
parts = []
with client.stream("POST", "/v1/chat", json=request, headers={"Authorization": f"Bearer {token}"}) as r:
r.raise_for_status()
for line in r.iter_lines():
if not line.startswith("data:"):
continue
event = json.loads(line[5:])
if event.get("done"):
usage = event["usage"]
break
parts.append(event["delta"])
print("".join(parts).strip())
print(usage)This is a practice answer about: What is an idempotency key. A real model would say more.
{'prompt_tokens': 5, 'completion_tokens': 17, 'total_tokens': 22}Follow-up: "It works locally but arrives all at once in production." A proxy is buffering: send X-Accel-Buffering: no (FastAPI does), disable buffering or compression for that route at the proxy, and make sure nothing (like a middleware that reads the whole body) collects the stream. Also give the client a long read timeout and handle a stream that ends without its final event.
Part 3: FastAPI and Python challenges
F1. Validate nested JSON (Easy)
Task: a chat endpoint accepts a list of messages. Reject unknown roles, empty content, more than 50 messages, a temperature outside 0–2 and unexpected fields, with errors that point at the exact field.
Hint: nest one model inside another, use Literal for fixed choices, Field for limits and extra="forbid" for unknown fields.
Solution:
from typing import Annotated, Literal
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict, Field
class Message(BaseModel):
role: Literal["system", "user", "assistant"]
content: Annotated[str, Field(min_length=1, max_length=4000)]
class ChatRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
model: str = "practice-llm-1"
messages: Annotated[list[Message], Field(min_length=1, max_length=50)]
temperature: Annotated[float, Field(ge=0, le=2)] = 0.7
app = FastAPI()
@app.post("/v1/chat")
def chat(request: ChatRequest) -> dict:
return {"messages": len(request.messages), "temperature": request.temperature}
client = TestClient(app)
good = {"messages": [{"role": "system", "content": "Be brief."}, {"role": "user", "content": "Hi"}]}
bad = {"messages": [{"role": "user", "content": ""}, {"role": "robot", "content": "Hi"}],
"temperature": 3, "stream_mode": "fast"}
print(client.post("/v1/chat", json=good).json())
r = client.post("/v1/chat", json=bad)
print(r.status_code)
for error in r.json()["detail"]:
print(error["loc"], error["type"]){'messages': 2, 'temperature': 0.7}
422
['body', 'messages', 0, 'content'] string_too_short
['body', 'messages', 1, 'role'] literal_error
['body', 'temperature'] less_than_equal
['body', 'stream_mode'] extra_forbiddenWhy it works: Pydantic validates the whole tree and reports every problem with its path: ["body", "messages", 1, "role"] means "the second message's role". A client can show that next to the right field. Follow-up: "Why limit the list and the strings?" Each message costs tokens and money; limits are your first defence against huge or abusive requests (challenge S5).
F2. Fix wrong error responses (Medium)
Task: this API answers 200 when a course is missing, 500 when the body is wrong, and 200 when a duplicate is created. Fix it so every outcome has the right status code and a clear message.
from fastapi import FastAPI
from fastapi.testclient import TestClient
bad_app = FastAPI()
COURSES = {1: {"id": 1, "title": "Python for AI"}}
@bad_app.get("/courses/{course_id}")
def get_course(course_id: int):
try:
return COURSES[course_id]
except Exception as e:
return {"error": str(e)}
@bad_app.post("/courses")
def create_course(course: dict):
if any(c["title"] == course["title"] for c in COURSES.values()):
return {"error": "exists"}
new = {"id": max(COURSES) + 1, "title": course["title"]}
COURSES[new["id"]] = new
return new
def report(app):
client = TestClient(app, raise_server_exceptions=False)
for method, path, body in [("GET", "/courses/1", None), ("GET", "/courses/42", None),
("POST", "/courses", {"name": "typo"}),
("POST", "/courses", {"title": "Python for AI"}),
("POST", "/courses", {"title": "LLM APIs"})]:
r = client.request(method, path, json=body)
print(f"{method:<4} {path:<12} {str(body):<28} -> {r.status_code} {r.text}")
report(bad_app)GET /courses/1 None -> 200 {"id":1,"title":"Python for AI"}
GET /courses/42 None -> 200 {"error":"42"}
POST /courses {'name': 'typo'} -> 500 Internal Server Error
POST /courses {'title': 'Python for AI'} -> 200 {"error":"exists"}
POST /courses {'title': 'LLM APIs'} -> 200 {"id":2,"title":"LLM APIs"}Hint: let Pydantic validate the body, raise HTTPException for "not found" and "conflict", and set 201 for creation. Never send exception text to the client.
Solution:
from typing import Annotated
from fastapi import HTTPException
from pydantic import BaseModel, Field
class CourseIn(BaseModel):
title: Annotated[str, Field(min_length=3, max_length=100)]
class Course(CourseIn):
id: int
good_app = FastAPI()
COURSES = {1: {"id": 1, "title": "Python for AI"}}
@good_app.get("/courses/{course_id}")
def get_course_fixed(course_id: int) -> Course:
course = COURSES.get(course_id)
if course is None:
raise HTTPException(404, f"Course {course_id} not found")
return course
@good_app.post("/courses", status_code=201)
def create_course_fixed(course: CourseIn) -> Course:
if any(c["title"].lower() == course.title.lower() for c in COURSES.values()):
raise HTTPException(409, f"A course titled {course.title!r} already exists")
new = {"id": max(COURSES) + 1, "title": course.title}
COURSES[new["id"]] = new
return new
report(good_app)GET /courses/1 None -> 200 {"title":"Python for AI","id":1}
GET /courses/42 None -> 404 {"detail":"Course 42 not found"}
POST /courses {'name': 'typo'} -> 422 {"detail":[{"type":"missing","loc":["body","title"],"msg":"Field required","input":{"name":"typo"}}]}
POST /courses {'title': 'Python for AI'} -> 409 {"detail":"A course titled 'Python for AI' already exists"}
POST /courses {'title': 'LLM APIs'} -> 201 {"title":"LLM APIs","id":2}Why it works: a client (or a monitoring dashboard) decides what to do from the status code first. The old 200-with-"error" looked like success; the old 500 blamed the server for a client mistake; str(e) leaked internals (here just '42', elsewhere a SQL query or a file path). Follow-up: "How would you make every error look the same?" Add exception handlers that turn HTTPException and validation errors into one envelope, for example {"error": {"code": "not_found", "message": "…", "request_id": "…"}} (FastAPI Architecture: Routers, Schemas and Error Handling).
F3. A blocking call in an async route (Medium)
Task: under load, an endpoint that "only waits one second" makes the whole server slow. Reproduce it with three concurrent requests and fix it two ways.
# file: blocking_app.py
import asyncio
import time
from fastapi import FastAPI
app = FastAPI()
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/bad")
async def bad():
time.sleep(1) # blocks the event loop: no other request can run meanwhile
return {"route": "bad"}
@app.get("/fixed-async")
async def fixed_async():
await asyncio.sleep(1) # gives control back to the event loop while waiting
return {"route": "fixed-async"}
@app.get("/fixed-def")
def fixed_def():
time.sleep(1) # a plain def route runs in FastAPI's threadpool
return {"route": "fixed-def"}cd ~/projects/lab
fastapi run blocking_app.py --port 8000 > server.log 2>&1 &
echo $! > server.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health{"status":"ok"}import asyncio
import time
import httpx
async def three_at_once(path: str) -> float:
async with httpx.AsyncClient(base_url="http://localhost:8000", timeout=10) as client:
start = time.perf_counter()
await asyncio.gather(*(client.get(path) for _ in range(3)))
return time.perf_counter() - start
for path in ["/bad", "/fixed-async", "/fixed-def"]:
print(f"{path:<12} 3 concurrent requests took about {round(asyncio.run(three_at_once(path)))} s")/bad 3 concurrent requests took about 3 s
/fixed-async 3 concurrent requests took about 1 s
/fixed-def 3 concurrent requests took about 1 skill $(cat server.pid)Hint: an async def route runs on the event loop. What happens to the loop during time.sleep?
Solution: the three requests to /bad ran one after another (3 s) because time.sleep froze the single event loop; the fixed routes overlapped (1 s). The rule: inside async def, only await non-blocking work (httpx.AsyncClient, async DB drivers, asyncio.sleep). If you must call blocking code (requests, a sync DB driver, model.predict on a large model), declare the route with plain def, or push the call to a thread with await run_in_threadpool(fn, …) from fastapi.concurrency. Follow-up: "And CPU-heavy inference?" Threads don't help much with CPU-bound Python; use a process pool, a separate model server, or a job queue (Serving an ML Model).
F4. A database session leak (Hard)
Task: the API works for the first request, then every request fails after a pause. The pool is deliberately tiny (1 connection, 1-second timeout) so the bug shows at once.
from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker
engine = create_engine("sqlite:///lab.db", pool_size=1, max_overflow=0, pool_timeout=1)
SessionLocal = sessionmaker(engine)
leaked = [] # stands in for wherever a real leak keeps the session alive
def get_db_leaky():
session = SessionLocal()
leaked.append(session)
return session
leaky = FastAPI()
@leaky.get("/ping-db")
def ping_db(db: Annotated[Session, Depends(get_db_leaky)]):
return {"db": db.execute(text("SELECT 1")).scalar_one()}
client = TestClient(leaky, raise_server_exceptions=False)
for i in range(1, 4):
r = client.get("/ping-db")
note = f", connections checked out: {engine.pool.checkedout()}" if i == 1 else ""
print(f"request {i}: {r.status_code}{note}")request 1: 200, connections checked out: 1
request 2: 500
request 3: 500Hint: who closes the session, and when?
Solution: nobody. Each session keeps its connection checked out of the pool until it is closed, so the second request waits pool_timeout seconds for a connection that never comes back, and fails with SQLAlchemy's "QueuePool limit … reached" error. (The leaked list plays the part of whatever keeps the session alive in a real leak, such as a cache or a global. Without it, Python's garbage collector might close an abandoned session at some unpredictable moment; that is luck, not a fix.) A dependency with yield closes the session after the response, even when the route raises:
engine = create_engine("sqlite:///lab.db", pool_size=1, max_overflow=0, pool_timeout=1)
SessionLocal = sessionmaker(engine)
def get_db():
with SessionLocal() as session: # closes the session (and returns the connection) at the end
yield session
SessionDep = Annotated[Session, Depends(get_db)]
fixed = FastAPI()
@fixed.get("/ping-db")
def ping_db_fixed(db: SessionDep):
return {"db": db.execute(text("SELECT 1")).scalar_one()}
client = TestClient(fixed, raise_server_exceptions=False)
for i in range(1, 4):
r = client.get("/ping-db")
note = f", connections checked out: {engine.pool.checkedout()}" if i == 1 else ""
print(f"request {i}: {r.status_code}{note}")request 1: 200, connections checked out: 0
request 2: 200
request 3: 200Why it works: FastAPI runs the code after yield when the request is done, and the with block guarantees close(). In production the symptom is the same but slower: latency climbs, then timeouts appear under load. Watch pool metrics (checkedout, waits) and keep transactions short. Follow-up: "Other ways to leak?" Sessions created in background tasks or scripts without with, a global session shared between requests (also a correctness bug), and long-running streams that hold a session open while the LLM talks (Databases with SQLAlchemy and Alembic).
F5. Mock an external API in tests (Medium)
Task: this endpoint calls an upstream weather API. Test success, unknown city, upstream failure and upstream timeout, without any real network call.
# file: weather_app.py
import os
import httpx
from fastapi import FastAPI, HTTPException
UPSTREAM = os.environ.get("WEATHER_UPSTREAM", "https://weather.example.com")
app = FastAPI()
upstream = httpx.Client(base_url=UPSTREAM, timeout=5)
@app.get("/weather/{city}")
def weather(city: str) -> dict:
try:
r = upstream.get("/v1/weather", params={"city": city})
except httpx.TransportError:
raise HTTPException(502, "Weather service unreachable")
if r.status_code == 404:
raise HTTPException(404, f"Unknown city: {city}")
if r.status_code >= 400:
raise HTTPException(502, "Weather service failed")
data = r.json()
return {"city": data["city"], "summary": f"{data['condition']}, {data['temp_c']} °C"}Hint: respx intercepts httpx requests and lets you script responses and errors. By default it fails any request you did not mock, which proves nothing leaks to the network.
Solution:
# file: test_weather_app.py
import httpx
import respx
from fastapi.testclient import TestClient
from weather_app import app
client = TestClient(app)
WEATHER = "https://weather.example.com/v1/weather"
@respx.mock
def test_weather_ok():
route = respx.get(WEATHER, params={"city": "Sylhet"}).mock(return_value=httpx.Response(
200, json={"city": "Sylhet", "temp_c": 27.4, "humidity": 85, "condition": "Rain"}))
r = client.get("/weather/Sylhet")
assert r.status_code == 200
assert r.json() == {"city": "Sylhet", "summary": "Rain, 27.4 °C"}
assert route.call_count == 1
@respx.mock
def test_unknown_city_is_404():
respx.get(WEATHER).mock(return_value=httpx.Response(404, json={"detail": "No weather data"}))
assert client.get("/weather/Narnia").status_code == 404
@respx.mock
def test_upstream_error_is_502():
respx.get(WEATHER).mock(return_value=httpx.Response(503))
r = client.get("/weather/Dhaka")
assert r.status_code == 502
assert r.json() == {"detail": "Weather service failed"}
@respx.mock
def test_upstream_timeout_is_502():
respx.get(WEATHER).mock(side_effect=httpx.ConnectTimeout("timed out"))
assert client.get("/weather/Dhaka").json() == {"detail": "Weather service unreachable"}python -m pytest -q test_weather_app.py.... [100%]
4 passed in 0.54sWhy it works: the tests are fast, deterministic and offline, and they cover the failure paths you can't trigger on demand against a real service. TestClient itself uses httpx2 here, so respx (which patches httpx) only intercepts your app's outbound calls. Follow-up: "Mock or dependency override?" Mocking HTTP tests your real client code; injecting a fake client with app.dependency_overrides is simpler when the client is a dependency. For LLM providers, a fake provider class behind the same interface is usually best (Testing Professional APIs).
F6. Test an authenticated endpoint (Medium)
Task: write tests for a JWT-protected endpoint: no token, garbage token, expired token, token signed with the wrong key, the owner, another user, and a unit test that skips auth.
# file: notes_app.py
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
SECRET = "lab-only-secret-0123456789abcdef0123" # from settings in a real app
ALGORITHM = "HS256"
NOTES = {1: {"id": 1, "owner": "nadia", "text": "Rotate the API key"},
2: {"id": 2, "owner": "rafi", "text": "Rafi's private note"}}
app = FastAPI()
bearer = HTTPBearer(auto_error=False)
def create_token(username: str, minutes: int = 15, secret: str = SECRET) -> str:
expires = datetime.now(timezone.utc) + timedelta(minutes=minutes)
return jwt.encode({"sub": username, "exp": expires}, secret, algorithm=ALGORITHM)
def current_user(creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> str:
if creds is None:
raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})
try:
payload = jwt.decode(creds.credentials, SECRET, algorithms=[ALGORITHM])
except jwt.InvalidTokenError:
raise HTTPException(401, "Invalid or expired token", headers={"WWW-Authenticate": "Bearer"})
return payload["sub"]
CurrentUser = Annotated[str, Depends(current_user)]
@app.get("/notes/{note_id}")
def read_note(note_id: int, user: CurrentUser) -> dict:
note = NOTES.get(note_id)
if note is None or note["owner"] != user:
raise HTTPException(404, "Note not found")
return noteHint: create real tokens in the tests with the app's own helper; pytest.mark.parametrize covers the bad-token cases in one test.
Solution:
# file: test_notes_app.py
import pytest
from fastapi.testclient import TestClient
from notes_app import app, create_token, current_user
client = TestClient(app)
def auth(token: str) -> dict:
return {"Authorization": f"Bearer {token}"}
def test_no_token_is_401():
r = client.get("/notes/1")
assert r.status_code == 401
assert r.headers["www-authenticate"] == "Bearer"
@pytest.mark.parametrize("token", [
"not-a-jwt",
create_token("nadia", minutes=-1),
create_token("nadia", secret="attacker-secret-0123456789abcdef0123"),
])
def test_bad_tokens_are_401(token):
assert client.get("/notes/1", headers=auth(token)).status_code == 401
def test_owner_can_read_own_note():
r = client.get("/notes/1", headers=auth(create_token("nadia")))
assert r.status_code == 200
assert r.json()["owner"] == "nadia"
def test_other_users_note_is_404():
assert client.get("/notes/2", headers=auth(create_token("nadia"))).status_code == 404
def test_override_auth_in_unit_tests():
app.dependency_overrides[current_user] = lambda: "rafi"
try:
assert client.get("/notes/2").status_code == 200
finally:
app.dependency_overrides.clear()python -m pytest -q test_notes_app.py....... [100%]
7 passed in 0.50sWhy it works: the tests cover the auth layer from the outside (real tokens through the real dependency) and include the most important negative test: a valid user reading someone else's data. The override test shows how other unit tests can skip auth, and the finally keeps the override from leaking into later tests. Follow-up: "What would you add?" Tests for missing claims, the none algorithm (PyJWT rejects it because you pass algorithms=[…]), revoked tokens, and roles (Authentication and Authorisation).
Part 4: Security challenges
S1. An exposed API key (Easy)
Task: a teammate shipped a web page that calls the LLM provider directly from the browser with the company key in the JavaScript. What do you do, in order, and how should the key be handled instead?
Hint: anything sent to a browser or a mobile app is public, minified or not.
Solution: (1) Revoke or rotate the key now and check the provider's usage and billing for abuse; set a spending limit. (2) Move the call to your backend: the browser calls your API with the user's session, your API checks auth and per-user limits, then calls the provider with a key that only the server knows. (3) Load the key from the environment into a settings object that won't print it by accident:
import os
from pydantic import SecretStr
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
llm_api_key: SecretStr
llm_model: str = "practice-llm-1"
os.environ["LLM_API_KEY"] = "bb-live-rotated-5e6f" # in real life: .env file, platform secrets or a secret manager
settings = Settings()
print(settings)
print(f"key loaded: {len(settings.llm_api_key.get_secret_value())} characters")llm_api_key=SecretStr('**********') llm_model='practice-llm-1'
key loaded: 20 characters(4) Remove it from the code and, if it was committed, follow G6. Follow-up: "The frontend needs to stream answers." Then the backend streams to the browser with SSE (H9) while it streams from the provider; the key still never leaves the server.
S2. A broken access check: BOLA (Medium)
Task: any logged-in user can read any invoice by changing the ID in the URL. This is Broken Object Level Authorization, number one in the OWASP API Security Top 10 (2023). Show it, fix it, prove it.
from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.testclient import TestClient
TOKENS = {"token-nadia": "nadia", "token-rafi": "rafi"}
INVOICES = {101: {"id": 101, "owner": "nadia", "amount_bdt": 1200},
102: {"id": 102, "owner": "rafi", "amount_bdt": 54000}}
def current_user(authorization: Annotated[str, Header()] = "") -> str:
user = TOKENS.get(authorization.removeprefix("Bearer "))
if user is None:
raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})
return user
app = FastAPI()
@app.get("/v1/invoices/{invoice_id}")
def get_invoice_vulnerable(invoice_id: int, user: Annotated[str, Depends(current_user)]):
invoice = INVOICES.get(invoice_id) # checks that you are logged in, never that it is yours
if invoice is None:
raise HTTPException(404, "Invoice not found")
return invoice
@app.get("/v2/invoices/{invoice_id}")
def get_invoice_fixed(invoice_id: int, user: Annotated[str, Depends(current_user)]):
invoice = INVOICES.get(invoice_id)
if invoice is None or invoice["owner"] != user:
raise HTTPException(404, "Invoice not found")
return invoice
client = TestClient(app)
nadia = {"Authorization": "Bearer token-nadia"}
for version in ("v1", "v2"):
own = client.get(f"/{version}/invoices/101", headers=nadia)
other = client.get(f"/{version}/invoices/102", headers=nadia)
print(f"{version}: own invoice {own.status_code}, Rafi's invoice {other.status_code} {other.json()}")v1: own invoice 200, Rafi's invoice 200 {'id': 102, 'owner': 'rafi', 'amount_bdt': 54000}
v2: own invoice 200, Rafi's invoice 404 {'detail': 'Invoice not found'}Hint: authentication answers "who are you?"; every object access also needs "is this yours?"
Why the fix works: the ownership check happens on every object, and a foreign object looks exactly like a missing one (404), so IDs cannot be probed. With a database, put the owner in the query itself (WHERE id = :id AND owner_id = :user_id) so the check can't be forgotten in one route; random IDs (UUIDs) help but are not a fix. Follow-up: "What about admins?" Make roles explicit and test them: function-level checks (BFLA) and object-level checks (BOLA) are separate tests (API Security and the OWASP API Top 10).
S3. SQL injection (Easy)
Task: a lookup by email builds SQL with an f-string. Show the attack and the fix.
import sqlite3
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT UNIQUE, plan TEXT)")
con.executemany("INSERT INTO users (email, plan) VALUES (?, ?)",
[("nadia@example.com", "student"), ("rafi@example.com", "pro"), ("admin@example.com", "admin")])
def find_user_unsafe(email: str):
return con.execute(f"SELECT id, email, plan FROM users WHERE email = '{email}'").fetchall()
def find_user_safe(email: str):
return con.execute("SELECT id, email, plan FROM users WHERE email = ?", (email,)).fetchall()
attack = "x' OR '1'='1"
print("unsafe:", find_user_unsafe(attack))
print("safe: ", find_user_safe(attack))
print("safe: ", find_user_safe("rafi@example.com"))unsafe: [(1, 'nadia@example.com', 'student'), (2, 'rafi@example.com', 'pro'), (3, 'admin@example.com', 'admin')]
safe: []
safe: [(2, 'rafi@example.com', 'pro')]Hint: write out the final SQL string the unsafe function builds.
Solution: the unsafe query becomes WHERE email = 'x' OR '1'='1', which is true for every row, so the attacker gets all users. With a placeholder the driver sends the value separately from the SQL text, so it is only ever data. Parameters work for values, not for identifiers: for a user-chosen sort column, map it through an allow-list ({"newest": "created_at DESC"}). The same applies to SQLAlchemy's text("… WHERE email = :email") with a parameter dict, and to an LLM that writes SQL: run its queries with a read-only role, an allow-list of tables and a row limit. Follow-up: "Does an ORM make you safe?" Mostly, until someone uses raw text() with an f-string.
S4. Unsafe CORS (Medium)
Task: the API uses cookie sessions and this CORS configuration "to make the frontend work". What can an evil website do?
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.testclient import TestClient
def make_app(origins: list[str]) -> FastAPI:
app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=origins, allow_credentials=True,
allow_methods=["*"], allow_headers=["*"])
@app.get("/v1/me")
def me():
return {"user": "nadia", "plan": "student"}
return app
def cors_headers(app: FastAPI, origin: str) -> tuple:
r = TestClient(app).get("/v1/me", headers={"Origin": origin, "Cookie": "session=abc"})
return r.headers.get("access-control-allow-origin"), r.headers.get("access-control-allow-credentials")
unsafe = make_app(["*"])
print("unsafe, evil site:", cors_headers(unsafe, "https://evil.example"))unsafe, evil site: ('https://evil.example', 'true')Hint: browsers forbid Access-Control-Allow-Origin: * together with credentials. Look at what the server sends instead.
Solution: with "*" and credentials, Starlette echoes back whatever origin asks, plus allow-credentials: true. So any website the user visits can call the API with the user's cookies and read the answer. List the exact origins you trust:
safe = make_app(["https://app.bitbyte.example", "http://localhost:5173"])
print("safe, evil site: ", cors_headers(safe, "https://evil.example"))
print("safe, our app: ", cors_headers(safe, "https://app.bitbyte.example"))safe, evil site: (None, 'true')
safe, our app: ('https://app.bitbyte.example', 'true')Why it works: without an allow-origin header, the browser hides the response from the evil page. Remember what CORS is: a browser rule about which pages may read responses. It is not authentication; curl and scripts ignore it, so every endpoint still needs real auth. With cookies, also use SameSite and CSRF protection for state-changing requests. Load the allowed origins from settings per environment.
S5. Request size limits (Medium)
Task: someone posts a 2 MB body and a chat with 500 messages to your LLM endpoint. Reject both early, with the right status codes.
Hint: there are two layers: the raw size of the body, and the shape of the data inside.
Solution:
from typing import Annotated
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field
MAX_BODY_BYTES = 64_000
class Msg(BaseModel):
role: str
content: Annotated[str, Field(max_length=4000)]
class Chat(BaseModel):
messages: Annotated[list[Msg], Field(min_length=1, max_length=50)]
app = FastAPI()
@app.middleware("http")
async def limit_body_size(request: Request, call_next):
length = request.headers.get("content-length")
if length is not None and int(length) > MAX_BODY_BYTES:
return JSONResponse({"detail": f"Body larger than {MAX_BODY_BYTES} bytes"}, status_code=413)
return await call_next(request)
@app.post("/v1/chat")
def chat(body: Chat) -> dict:
return {"messages": len(body.messages)}
client = TestClient(app)
small = {"messages": [{"role": "user", "content": "Hi"}]}
huge = {"messages": [{"role": "user", "content": "x" * 2_000_000}]}
many = {"messages": [{"role": "user", "content": "Hi"}] * 500}
long_text = {"messages": [{"role": "user", "content": "y" * 5000}]}
for name, body in [("small", small), ("huge", huge), ("many", many), ("long_text", long_text)]:
r = client.post("/v1/chat", json=body)
print(f"{name:<10} {r.status_code} {r.json()['detail'][0]['type'] if r.status_code == 422 else r.json()}")small 200 {'messages': 1}
huge 413 {'detail': 'Body larger than 64000 bytes'}
many 422 too_long
long_text 422 string_too_longWhy it works: the middleware refuses an oversized body with 413 before anything parses it; Pydantic limits the number and length of messages (422). Body size is best enforced in front of the app too (the proxy or platform: for example nginx client_max_body_size), because a client can send a chunked body without Content-Length, which this simple check doesn't see. Add per-user rate limits and token budgets: OWASP calls this "unrestricted resource consumption", and with LLMs it costs real money. Follow-up: "File uploads?" Stream them to storage with a size cap, check the type by content, not by file name.
S6. SSRF: fetching a user-supplied URL (Hard)
Task: your "summarise this web page" feature downloads any URL a user sends. An attacker sends http://169.254.169.254/latest/meta-data/ (the cloud metadata service, which can hand out credentials) or http://localhost:8800/… (your internal services). Write the check that runs before every fetch.
Hint: check the scheme, then resolve the host name and check every address it resolves to; a harmless-looking name can point at 127.0.0.1.
Solution:
import ipaddress
import socket
from urllib.parse import urlsplit
def resolve_all(host: str) -> list[str]:
return [info[4][0] for info in socket.getaddrinfo(host, None)]
def check_url(url: str, resolve=resolve_all) -> str:
parts = urlsplit(url)
if parts.scheme not in ("http", "https"):
return f"blocked: scheme {parts.scheme!r}"
if not parts.hostname:
return "blocked: no host"
try:
addresses = resolve(parts.hostname)
except (socket.gaierror, KeyError):
return "blocked: host does not resolve"
for address in addresses:
ip = ipaddress.ip_address(address)
if not ip.is_global:
return f"blocked: {parts.hostname} -> {ip} is not a public address"
return "allowed"
FAKE_DNS = {"docs.example.org": ["93.184.215.14"], "evil.example": ["127.0.0.1"], "localhost": ["127.0.0.1"]} # stands in for DNS
for url in ["https://docs.example.org/guide", "http://169.254.169.254/latest/meta-data/",
"http://localhost/admin", "http://10.0.0.5/admin", "http://[::1]:6379/",
"file:///etc/passwd", "http://evil.example/page"]:
print(f"{url:<42} {check_url(url, resolve=lambda h: FAKE_DNS.get(h) or resolve_all(h))}")https://docs.example.org/guide allowed
http://169.254.169.254/latest/meta-data/ blocked: 169.254.169.254 -> 169.254.169.254 is not a public address
http://localhost/admin blocked: localhost -> 127.0.0.1 is not a public address
http://10.0.0.5/admin blocked: 10.0.0.5 -> 10.0.0.5 is not a public address
http://[::1]:6379/ blocked: ::1 -> ::1 is not a public address
file:///etc/passwd blocked: scheme 'file'
http://evil.example/page blocked: evil.example -> 127.0.0.1 is not a public addressWhy it works: ip.is_global is false for loopback, private, link-local (the metadata service), and other reserved ranges, for IPv4 and IPv6. Two more rules make it complete: don't follow redirects automatically (a public URL can redirect to 127.0.0.1; re-check each hop), and connect to the address you checked, not a fresh lookup (DNS rebinding: the name can resolve differently a second later). The strongest defence is an allow-list of domains, plus running the fetcher where it cannot reach internal networks. Also cap response size and time. Follow-up: "Where does SSRF hide in AI apps?" RAG ingestion from URLs, agents with a "fetch" tool, image URLs for vision models, and webhooks with user-chosen target URLs.
S7. Webhook security end to end (Medium)
Task: build the receiving endpoint for the webhooks from H8: reject bad signatures and stale timestamps, and process each event only once even if the provider delivers it twice.
Hint: read the raw body with await request.body() before parsing; remember the event IDs you've processed.
Solution (it reuses sign and verify from H8):
import json
import time
from typing import Annotated
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.testclient import TestClient
app = FastAPI()
PROCESSED: set[str] = set() # a table with a unique event_id column in production
@app.post("/webhooks/payments")
async def payment_webhook(request: Request,
x_webhook_timestamp: Annotated[str, Header()],
x_webhook_signature: Annotated[str, Header()]) -> dict:
raw = await request.body()
if not verify(raw, x_webhook_timestamp, x_webhook_signature):
raise HTTPException(401, "Invalid signature or timestamp")
event = json.loads(raw)
if event["id"] in PROCESSED:
return {"status": "duplicate ignored"}
PROCESSED.add(event["id"])
# enqueue the real work here and return quickly
return {"status": "accepted"}
def deliver(body: bytes, timestamp: int, secret: bytes = SECRET):
headers = {"X-Webhook-Timestamp": str(timestamp), "X-Webhook-Signature": sign(body, timestamp, secret),
"Content-Type": "application/json"}
r = client.post("/webhooks/payments", content=body, headers=headers)
return r.status_code, r.json()
client = TestClient(app)
event = b'{"event":"payment.succeeded","id":"evt_2001"}'
now = int(time.time())
print("first delivery: ", deliver(event, now))
print("redelivery: ", deliver(event, now))
print("forged: ", deliver(event, now, secret=b"guess"))
print("replayed later: ", deliver(event, now - 3600))first delivery: (200, {'status': 'accepted'})
redelivery: (200, {'status': 'duplicate ignored'})
forged: (401, {'detail': 'Invalid signature or timestamp'})
replayed later: (401, {'detail': 'Invalid signature or timestamp'})Why it works: signature plus timestamp proves who sent it and when; the processed-ID set makes delivery idempotent, which matters because providers retry until they get a 2xx. Answer fast and do slow work in a queue, or the provider times out and retries. Use HTTPS, keep the secret in settings, support two secrets during rotation, and log the event ID (not the payload). Follow-up: "What if your endpoint was down for an hour?" Providers retry with backoff for a while; also reconcile by polling the provider's API for events you missed.
S8. Sensitive data in logs (Easy)
Task: logs contain bearer tokens, API keys and emails, and they are shipped to a third-party log service. Redact them before they leave the process.
Hint: a logging.Filter sees every record before a handler writes it.
Solution:
import logging
import re
import sys
class RedactFilter(logging.Filter):
RULES = [
(re.compile(r"(?i)(bearer\s+)[\w.~+/-]+=*"), r"\1[REDACTED]"),
(re.compile(r"bb-live-\w+"), "[REDACTED_KEY]"),
(re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"), "[EMAIL]"),
]
def filter(self, record: logging.LogRecord) -> bool:
message = record.getMessage()
for pattern, replacement in self.RULES:
message = pattern.sub(replacement, message)
record.msg, record.args = message, None
return True
logging.basicConfig(format="%(levelname)s %(name)s %(message)s", stream=sys.stdout, force=True)
for handler in logging.getLogger().handlers:
handler.addFilter(RedactFilter())
log = logging.getLogger("lab.s8")
log.setLevel(logging.INFO)
log.info("login ok for %s", "nadia@example.com")
log.info("calling provider with header Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.abc.def")
log.warning("provider rejected key %s", "bb-live-9f8e7d6c5b4a")
log.info("chat request id=%s tokens=%d", "req-42", 118)INFO lab.s8 login ok for [EMAIL]
INFO lab.s8 calling provider with header Authorization: Bearer [REDACTED]
WARNING lab.s8 provider rejected key [REDACTED_KEY]
INFO lab.s8 chat request id=req-42 tokens=118Why it works: the filter rewrites the final message, so it catches values passed as arguments too. But redaction is a safety net, not the plan: don't log secrets or whole bodies in the first place. Log IDs, status codes, timings and token counts; keep prompts and model outputs (which often contain personal data) out of general logs, or store them separately with access control and a retention period. Follow-up: "Where else do secrets leak?" Exception tracebacks with local variables, error messages returned to clients, CI logs (set -x), and URLs with keys in the query string, which end up in proxy logs.
Part 5: System-design mini interviews
A mini interview lasts 15–20 minutes. Interviewers don't expect one right answer; they grade how you structure the conversation. Use the same five steps every time: requirements (functional and non-functional: users, traffic, latency, cost), API, data, architecture (draw it), failure modes and trade-offs. The answers below are model answers to practise against.
D1. A weather API proxy (Easy)
Prompt: "Our app shows weather for Bangladeshi cities. The upstream provider allows 1,000 calls a day and is sometimes slow. Design our backend."
Hint: weather changes slowly; most users ask about the same few cities.
mobile app ──▶ GET /v1/weather/{city} ──▶ FastAPI ──▶ Redis cache (TTL 10 min) ──miss──▶ provider
│ (timeout 3 s,
└── on provider error: serve stale value retry once)
(up to 1 h old) with "stale": true| Decision | Choice and why |
|---|---|
| Caching | Cache per normalised city (weather:dhaka), TTL 10 minutes: 5 cities × 6 per hour × 24 = 720 calls a day, under the quota whatever the traffic |
| Stampede | One request refreshes a key while others wait or get the stale value (a short Redis lock) |
| Failure | Timeout 3 s, one retry with jitter, then serve stale data marked "stale": true; if there is none, 503 with Retry-After |
| Validation | Allow-list of cities (also blocks cache-key abuse); normalise case and spelling |
| Protection | Per-client rate limit; the provider key stays on the server |
| Observability | Cache hit rate, upstream latency and errors, remaining quota |
Follow-up: "Traffic grows 100×?" Nothing changes upstream (the cache absorbs it); scale the API horizontally, since all state is in Redis.
D2. An authentication API (Medium)
Prompt: "Design sign-up and login for our AI study app: web and mobile clients, about 50,000 users."
Hint: separate the identity problem (who are you) from the session problem (how do you stay logged in).
| Area | Model answer |
|---|---|
| Endpoints | POST /auth/register, POST /auth/login, POST /auth/refresh, POST /auth/logout, GET /me; also password reset and email verification |
| Passwords | Argon2 hashes (pwdlib), never stored or logged in plain text; a minimum length and a check against breached passwords |
| Tokens | Short-lived JWT access token (about 15 min) with sub, exp, roles; a refresh token stored server-side (hashed) and rotated on every use, with reuse detection that revokes the whole family |
| Clients | Browser: refresh token in an HttpOnly, Secure, SameSite cookie, plus CSRF protection; mobile: secure OS storage |
| Abuse | Rate limit login per account and per IP, generic error messages ("wrong email or password"), optional MFA, alerts on unusual logins |
| Data | users(id, email unique, password_hash, created_at), refresh_tokens(id, user_id, token_hash, family_id, expires_at, revoked_at) |
| Build or buy | "Sign in with Google" or a hosted identity provider uses OAuth 2.0/OIDC with the authorization code flow and PKCE; often the better choice than building your own |
Follow-up: "How do you log out a JWT?" You can't change a token that was issued; keep access tokens short, revoke the refresh token, and keep a small denylist of token IDs (jti) in Redis for emergencies.
D3. An ML prediction endpoint (Medium)
Prompt: "We trained a scikit-learn model that predicts whether a student will drop a course. Serve it to our web app."
client ──▶ POST /v1/predictions ──▶ FastAPI (validate features with Pydantic)
│
├── model loaded once at startup (lifespan), version "dropout-2026-10-01"
├── predict in a threadpool (CPU work off the event loop)
└── log: request id, model version, latency, prediction (no personal data)
│
▼
monitoring: latency, error rate, input drift- API: request with typed, range-checked features; response
{"prediction": 1, "probability": 0.82, "model_version": "…"}; a batch endpoint for up to N rows. - Model lifecycle: the model file is versioned (registry or object storage), loaded at startup with a warm-up prediction; the readiness check fails until it is loaded; roll out a new version beside the old one and compare.
- Performance: small models predict in milliseconds; scale with more processes or containers; GPU only if the model needs it.
- Correctness: the same preprocessing pipeline as in training (save the whole
Pipeline, not just the estimator); tests with a tiny model.
Follow-up: "Predictions got worse over time." Monitor input distributions and, once labels arrive, accuracy; retrain on a schedule or on drift (Serving an ML Model).
D4. An LLM streaming chat API (Hard)
Prompt: "Design the backend for a ChatGPT-style tutor: streamed answers, conversation history, several model providers."
browser ──POST /v1/chat (SSE)──▶ FastAPI ──▶ auth + per-user limits (Redis)
│
├──▶ load history (PostgreSQL), build prompt (versioned template)
├──▶ provider adapter ──▶ OpenAI Responses / Anthropic Messages / local model
│ (timeout, retry only before the first token, fallback model)
├──◀ deltas ──▶ re-emitted to the browser as SSE events
└──▶ on finish: save answer, tokens, cost, model, prompt version| Concern | Model answer |
|---|---|
| Streaming | SSE (text/event-stream) from the server to the browser: simple, works over HTTP, auto-reconnect; WebSockets only if you need two-way real time |
| Provider independence | One adapter interface (complete(), stream() yielding text deltas); provider-specific events mapped to your own delta/done/error events |
| Failures | Retry only before the first token is sent; mid-stream errors become an error event; client disconnects cancel the upstream call to save money |
| Cost and limits | Per-user token budget checked before the call, actual usage recorded after (D5); cap input length and max_output_tokens |
| Data | conversations, messages(role, content, tokens, model, prompt_version); trim or summarise history to fit the context window |
| Safety | Treat model output as untrusted (escape it in HTML, never execute it); defend against prompt injection in tools and retrieved documents; redact secrets from logs |
| Scale | Long-lived connections: async workers, generous proxy timeouts, buffering off; horizontal scaling because state is in PostgreSQL and Redis |
Follow-up: "How do you know answers are good?" Log prompt and model versions with every answer, build a small evaluation set, and compare versions before switching (Building an LLM Backend with Streaming).
D5. Per-user AI usage limits (Hard)
Prompt: "Free users get 20,000 tokens a day, paid users 500,000. Enforce it across 10 API servers."
Hint: you don't know the output tokens until the answer is finished.
- Where the counter lives: Redis, shared by all servers, key
usage:{user_id}:{yyyy-mm-dd}with a TTL of a few days. Increments withINCRBYare atomic, so ten servers can't race. - Reserve, then settle: before the call, estimate (input tokens +
max_output_tokens) and reserve that amount atomically (a small Lua script: check the remaining budget, increment if it fits). After the call, adjust by the real usage the provider reports. Without a reservation, 50 parallel requests could all pass the check. - Response: 429 with a clear message and
Retry-After(seconds until the day resets), plus headers such asX-RateLimit-Remaining; aGET /v1/usageendpoint for the UI. - Two separate limits: a request rate limit (requests per minute, token bucket) protects the service; the token budget controls cost. Also a global budget alarm for the whole company.
- Source of truth: write each call's usage to PostgreSQL for billing and analytics; Redis is the fast, approximate gate. If Redis is down, decide on purpose: fail closed for free users, fail open with a low cap for paid ones.
Follow-up: "Why per day and not a rolling 24 hours?" Simpler to explain and to reset; a sliding window is fairer but needs more storage. Say which you chose and why.
D6. Background processing (Medium)
Prompt: "Users upload a 200-page PDF; we extract text, chunk it and create embeddings. It takes two minutes."
POST /v1/documents ──▶ save file to object storage ──▶ insert job (status=queued) ──▶ 202 Accepted
Location: /v1/jobs/42
queue (Redis/RabbitMQ/SQS) ──▶ worker(s)
│ extract → chunk → embed
GET /v1/jobs/42 ◀── {"status": "running", "progress": 0.6} ◀── job table ◀──────────┘ updates status
(or a webhook / SSE notification when done)- Why not
BackgroundTasks? It runs in the web process: lost on restart or deploy, no retries, no visibility, and it competes with requests for CPU. Use a real queue and workers (Celery, RQ, Dramatiq, arq, or a cloud queue) for anything slow or important. - Reliability: jobs are idempotent (re-running a step doesn't duplicate chunks: upsert by document ID and chunk number), retried with backoff, and moved to a dead-letter queue after N failures; a per-job timeout.
- Limits: file size and page caps, per-user concurrent job limits, and worker concurrency sized to the embedding provider's rate limits.
Follow-up: "How does the user find out it's done?" Polling the status URL is simplest; add SSE or a webhook for a better experience.
D7. A reliable third-party integration (Hard)
Prompt: "When a student pays, we must enrol them in the course and send a receipt email. The payment provider sends webhooks; the email service is sometimes down."
- Inbound: verify webhook signatures, deduplicate by event ID, respond 2xx fast (S7); the provider's retries plus your dedup give exactly-once effects.
- Atomic state change: in one database transaction, mark the payment as paid, create the enrolment, and insert an "email to send" row in an outbox table. Either all happen or none.
- Outbound: a worker reads the outbox, calls the email service with an idempotency key, retries with backoff, marks the row as sent; a circuit breaker stops hammering a service that is down.
- Safety net: a nightly reconciliation job compares the provider's payments with your enrolments and alerts on differences, because webhooks can be missed.
- Observability: a correlation ID from webhook to email in every log line; alerts on outbox backlog age.
Follow-up: "Why not call the email service inside the webhook handler?" If it is slow or down, the handler times out, the provider retries, and you risk duplicate enrolments or lost emails. The outbox decouples them.
D8. A CI/CD pipeline (Medium)
Prompt: "Design CI/CD for our FastAPI + PostgreSQL service on GitHub."
pull request ──▶ CI: lint (ruff) → tests with PostgreSQL service → dependency + secret scans → docker build
│ required checks must pass, one review, then merge to main
main ──▶ build image once, tag with the commit SHA ──▶ push to GHCR
──▶ deploy to staging ──▶ run migrations ──▶ smoke tests
──▶ manual approval (environment protection) ──▶ deploy the SAME image to production ──▶ health check
└── rollback = redeploy the previous image tag- Security: top-level
permissions: contents: read, extra permissions per job only; secrets in GitHub environments; third-party actions pinned to full commit SHAs; never run untrusted fork code with secrets (pull_request_targetdanger). - Speed: cache uv downloads, run jobs in parallel, fail fast on lint.
- Migrations: backward-compatible (expand, deploy, then contract), so the old version still works during the rollout and a rollback is safe.
- Release: tags and release notes for versions; Dependabot keeps actions and packages current.
Follow-up: "A deploy broke production." Roll back first (redeploy the previous tag), investigate second, then add the missing test (CI/CD with GitHub Actions, Deployment and Production Readiness).
Score yourself
| Part | Challenges | Points |
|---|---|---|
| Git | G1 1, G2 1, G3 2, G4 2, G5 2, G6 3, G7 2 | 13 |
| HTTP and APIs | H1 1, H2 1, H3 1, H4 2, H5 2, H6 2, H7 3, H8 2, H9 3 | 17 |
| FastAPI and Python | F1 1, F2 2, F3 2, F4 3, F5 2, F6 2 | 12 |
| Security | S1 1, S2 2, S3 1, S4 2, S5 2, S6 3, S7 2, S8 1 | 14 |
| System design | D1 1, D2 2, D3 2, D4 3, D5 3, D6 2, D7 3, D8 2 | 18 |
| Total | 74 |
55 or more: ready for junior-to-mid backend and AI-engineering interviews; do the capstone. 35–54: solid; revisit the pages behind the challenges you missed and redo them in a week. Under 35: work through Parts 1–3 again slowly, typing every command yourself.
Common mistakes
- Jumping to the fix. In interviews and in incidents, an unreproduced bug is a guess. Fix: reproduce first (a command or a failing test), then change one thing.
- Destroying evidence.
git reset --hard, deleting logs or restarting before you look. Fix: copy what you need, usegit stashor a rescue branch, and read the reflog. - Retrying everything. Retrying 400s, or retrying a payment POST without an idempotency key. Fix: retry only timeouts, 429 and 5xx, with backoff, jitter and a limit, and only idempotent requests.
- Security as an afterthought. "We'll add auth checks later" is how BOLA and leaked keys happen. Fix: every endpoint gets an ownership test, and secrets never touch the code.
- Design answers without numbers or trade-offs. "Use Redis and Kubernetes" says nothing. Fix: estimate traffic and cost, state what you chose, what you gave up, and what you'd change at 10× scale.
Try it yourself
- Easy: In the
conflictrepository from G3, find which commit introduced the textTIMEOUT = 30, using one Git command. - Medium: Write
parse_retry_after(value, now)that accepts both forms ofRetry-After(seconds, or an HTTP date likeThu, 15 Jan 2026 04:00:30 GMT), returns the seconds to wait, never negative and at most 60, and returns 1 for garbage. - Hard: Write a fixed-window rate limiter as a FastAPI dependency: 3 requests per 10 seconds per
X-Client-Id, answering 429 withRetry-After. Inject the clock so a test can move time forward without sleeping, and prove it withTestClient.
Answers
cd ~/projects/lab/git/conflict
git log --oneline -S "TIMEOUT = 30"1cff844 Raise timeout to 30 s for slow LLM callsfrom datetime import datetime, timezone
from email.utils import parsedate_to_datetime
def parse_retry_after(value: str | None, now: datetime, cap: int = 60) -> int:
if not value:
return 1
value = value.strip()
if value.isdigit():
seconds = int(value)
else:
try:
seconds = int((parsedate_to_datetime(value) - now).total_seconds())
except (TypeError, ValueError):
return 1
return max(0, min(seconds, cap))
now = datetime(2026, 1, 15, 4, 0, 0, tzinfo=timezone.utc)
for value in ["7", "120", "Thu, 15 Jan 2026 04:00:30 GMT", "Thu, 15 Jan 2026 03:59:00 GMT", "soon", None]:
print(repr(value), "->", parse_retry_after(value, now))'7' -> 7
'120' -> 60
'Thu, 15 Jan 2026 04:00:30 GMT' -> 30
'Thu, 15 Jan 2026 03:59:00 GMT' -> 0
'soon' -> 1
None -> 1from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.testclient import TestClient
class FixedWindowLimiter:
def __init__(self, limit: int, window: int, clock):
self.limit, self.window, self.clock = limit, window, clock
self.counts: dict[tuple[str, int], int] = {}
def hit(self, client_id: str) -> int | None:
"""Count one request. Returns None if allowed, else the seconds until the window resets."""
now = self.clock()
window_start = int(now // self.window) * self.window
key = (client_id, window_start)
self.counts[key] = self.counts.get(key, 0) + 1
if self.counts[key] > self.limit:
return max(1, int(window_start + self.window - now))
return None
fake_now = [1000.0]
limiter = FixedWindowLimiter(limit=3, window=10, clock=lambda: fake_now[0])
def rate_limit(x_client_id: Annotated[str, Header()] = "anonymous") -> None:
retry_after = limiter.hit(x_client_id)
if retry_after is not None:
raise HTTPException(429, "Too many requests", headers={"Retry-After": str(retry_after)})
app = FastAPI()
@app.get("/v1/answer", dependencies=[Depends(rate_limit)])
def answer() -> dict:
return {"ok": True}
client = TestClient(app)
codes = [client.get("/v1/answer", headers={"X-Client-Id": "a"}).status_code for _ in range(4)]
blocked = client.get("/v1/answer", headers={"X-Client-Id": "a"})
other = client.get("/v1/answer", headers={"X-Client-Id": "b"}).status_code
fake_now[0] += 10
after = client.get("/v1/answer", headers={"X-Client-Id": "a"}).status_code
print(codes, blocked.status_code, blocked.headers["retry-after"], other, after)[200, 200, 200, 429] 429 10 200 200Notice in the last answer that the counter for a blocked request still increases, and that a fixed window allows a burst of up to 6 requests across a window boundary; a sliding window or token bucket smooths that out, and Redis makes it work across servers (Caching, Rate Limiting and Reliability with Redis).
Summary
- Work every problem the same way: clarify, reproduce, fix the smallest thing, prove it, explain the trade-off.
- Git rescue:
restorefor files,reset --softto redo a commit,revertfor shared history, the reflog for "lost" commits, and rotate-first for leaked secrets. - Reliable API clients page safely, honour
Retry-After, set timeouts, retry with backoff and jitter only what is safe, and use idempotency keys for writes. - FastAPI bugs that matter in production: blocking calls in
async def, sessions that are never closed, wrong status codes, and missing tests for failure paths and other users' data. - Security checks you should do without thinking: ownership on every object, parameters in SQL, exact CORS origins, size limits, SSRF checks, signed and deduplicated webhooks, clean logs.
- System-design answers follow requirements → API → data → architecture → failure modes, with numbers and trade-offs.
Next: Capstone Project, Cheat Sheet and What's Next brings all of this together in one production-ready AI backend for your portfolio, with a cheat sheet for the whole tutorial.