Build Your Own CRUD — The End-to-End That Cements Everything

# The capstone scaffold — ~80 lines for the bones
import hashlib, json, secrets, time, uuid
from datetime import datetime, timezone
from fastapi import FastAPI, HTTPException, Response, Header, status, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field

app = FastAPI(title='Items API', version='1.0.0')
app.add_middleware(
    CORSMiddleware,
    allow_origins=['http://localhost:5173'],
    allow_credentials=True, allow_methods=['*'], allow_headers=['*'],
)
bearer = HTTPBearer()
TOKENS = {'tok_valid_xyz': 'u_42'}

class Item(BaseModel):
    id: str = Field(..., pattern='^itm_')
    name: str
    description: str | None = None
    version: int

_ITEMS: dict[str, dict] = {}
_JOBS: dict[str, dict] = {}

def compute_etag(d: dict) -> str:
    body = json.dumps(d, sort_keys=True, separators=(',', ':')).encode()
    return f'"{hashlib.sha256(body).hexdigest()[:12]}"'

async def current_user(creds: HTTPAuthorizationCredentials = Depends(bearer)):
    uid = TOKENS.get(creds.credentials)
    if not uid:
        raise HTTPException(401, detail='invalid token',
                            headers={'WWW-Authenticate': 'Bearer'})
    return uid

@app.post('/items', status_code=201)
async def create_item(payload: dict, response: Response, uid=Depends(current_user)):
    item_id = 'itm_' + secrets.token_urlsafe(8)
    item = {'id': item_id, 'version': 1, **payload}
    _ITEMS[item_id] = item
    response.headers['Location'] = f'/items/{item_id}'
    response.headers['ETag']     = compute_etag(item)
    return item

# ... continue with GET / PUT / PATCH / DELETE / cursor pagination / action / async
#     (see the exercise for the full build out)

# The optimistic update + conditional GET combo (the harder half)
from fastapi import Header

@app.get('/items/{item_id}')
async def read_item(item_id: str, response: Response,
                    if_none_match: str | None = Header(None, alias='If-None-Match'),
                    uid=Depends(current_user)):
    item = _ITEMS.get(item_id)
    if not item:
        raise HTTPException(404)
    etag = compute_etag(item)
    response.headers['ETag'] = etag
    response.headers['Cache-Control'] = 'private, max-age=60, must-revalidate'
    response.headers['Vary'] = 'Authorization'
    if if_none_match == etag:
        response.status_code = 304
        return None
    return item

@app.put('/items/{item_id}')
async def replace_item(item_id: str, payload: dict, response: Response,
                       if_match: str | None = Header(None, alias='If-Match'),
                       uid=Depends(current_user)):
    existing = _ITEMS.get(item_id)
    if not existing:
        raise HTTPException(404)
    current_etag = compute_etag(existing)
    if if_match and if_match != current_etag:
        raise HTTPException(412, detail='ETag mismatch; refetch and retry')
    new = {**existing, **payload, 'version': existing['version'] + 1}
    _ITEMS[item_id] = new
    response.headers['ETag'] = compute_etag(new)
    return new

# Verify your capstone against every requirement

# Create returns 201 + Location
curl -i -X POST http://localhost:8000/items -H 'Authorization: Bearer tok_valid_xyz' \
  -H 'Content-Type: application/json' -d '{"name":"first"}'

# Save the ETag from the response
ETAG=$(curl -s -i http://localhost:8000/items/itm_xyz -H 'Authorization: Bearer tok_valid_xyz' | grep -i etag | awk '{print $2}' | tr -d '\r')

# Conditional GET — should be 304 with current ETag
curl -i http://localhost:8000/items/itm_xyz -H 'Authorization: Bearer tok_valid_xyz' -H "If-None-Match: $ETAG"

# Optimistic update — should be 412 with stale ETag
curl -i -X PUT http://localhost:8000/items/itm_xyz -H 'Authorization: Bearer tok_valid_xyz' \
  -H 'Content-Type: application/json' -H 'If-Match: "v99-stale"' -d '{"name":"updated"}'

# OpenAPI / Swagger UI available
open http://localhost:8000/docs