অনুশীলন ও ইন্টারভিউ ল্যাব

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"}

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.txt

git restore prompts.txt config.py
git status --short
cat config.py
git restore --source=HEAD~1 config.py
cat config.py
git status --short

MODEL = 'large'
MODEL = 'small'
 M config.py

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 --oneline

532fa3f wip
43cb677 Add predict

git reset --soft HEAD~1
git log --oneline
git status --short

43cb677 Add predict
M  model.py
A  settings.py

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.py

Auto-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 = 3

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 settings

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 commit

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 commit

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.md

c8c6d84 Model card v1
v1

git reflog -3
git branch rescue 'HEAD@{1}'
git reset -q --hard rescue
git log --oneline
cat model_card.md

c8c6d84 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 set

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.py

0

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"

pip install git-filter-repo
echo 'bb-live-9f8e7d6c5b4a==>REMOVED' > ../replacements.txt
git filter-repo --replace-text ../replacements.txt
git push --force-with-lease --all

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 main

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()

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"}}

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 429

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}}]}

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"}

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']

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}

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)

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: 1

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: False

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:  False

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: []

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}']

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}

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_forbidden

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"}

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}

# 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 s

kill $(cat server.pid)

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: 500

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: 200

# 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"}

# 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.54s

# 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 note

# 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.50s

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

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'}

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')]

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')

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')

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_long

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 address

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'})

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=118

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

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

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

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)

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

cd ~/projects/lab/git/conflict
git log --oneline -S "TIMEOUT = 30"

1cff844 Raise timeout to 30 s for slow LLM calls

from 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 -> 1

from 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 200