Beyond CRUD Checkpoint
28 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)Fix the mistakes
The slips that cost marks in this module. Spot each one, then check the fix.
A real shop service does more than save rows. After an order it sends a receipt, lets sellers upload product photos and keeps popular reads fast. This checkpoint joins those three jobs and checks the whole module.
Choose where the work runs
Work the client needs for its answer runs inside the route. Short, loss-tolerant work runs as a background task. Work that must survive restarts, be retried or be scheduled goes to a task queue.
@app.post("/orders", status_code=201)def place_order(body: OrderIn, tasks: BackgroundTasks): order = save_order(body) tasks.add_task(send_receipt, order.id) return orderExample 1 · Order with a receipt task and cache invalidation
from fastapi import BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, UploadFilefrom fastapi.testclient import TestClient
class FakeClock: def __init__(self, now=0.0): self.now = now
def __call__(self): return self.now
def advance(self, seconds): self.now += seconds
class TTLCache: def __init__(self, clock, default_ttl=60): self._clock = clock self._default_ttl = default_ttl self._data = {}
def get(self, key, default=None): item = self._data.get(key) if item is None: return default value, expires_at = item if self._clock() >= expires_at: del self._data[key] return default return value
def set(self, key, value, ttl=None): ttl = self._default_ttl if ttl is None else ttl self._data[key] = (value, self._clock() + ttl)
def delete(self, key): self._data.pop(key, None)
clock = FakeClock()cache = TTLCache(clock)stock = {1: 5}outbox = []app = FastAPI()
def send_receipt(email, qty): outbox.append(f"{email}: {qty} sold")
@app.get("/stock/{product_id}")def read_stock(product_id: int): left = cache.get(f"stock:{product_id}") if left is None: left = stock[product_id] cache.set(f"stock:{product_id}", left) return {"left": left}
@app.post("/orders/{product_id}", status_code=201)def order(product_id: int, qty: int, email: str, tasks: BackgroundTasks): if stock[product_id] < qty: raise HTTPException(status_code=409, detail="Not enough stock") stock[product_id] -= qty cache.delete(f"stock:{product_id}") tasks.add_task(send_receipt, email, qty) return {"ok": True}
client = TestClient(app)print(client.get("/stock/1").json())print(client.post("/orders/1", params={"qty": 2, "email": "a@example.com"}).status_code)print(client.get("/stock/1").json(), outbox)print(client.post("/orders/1", params={"qty": 9, "email": "b@example.com"}).status_code, outbox)Output
{'left': 5}
201
{'left': 3} ['a@example.com: 2 sold']
409 ['a@example.com: 2 sold']What this code does
- The order deletes the cached stock, so the next read shows 3, not the old 5.
- The receipt is a background task; the refused order adds none.
The order is saved before the answer. The receipt is slow and not needed for the answer, so it becomes a task.
Validate uploads before storing
Check the declared type, the size and the first bytes. Answer 415 for a wrong type and 413 for a large file. Store the file under a name you create.
async def read_png(file: UploadFile): data = await file.read(MAX_BYTES + 1) if len(data) > MAX_BYTES: raise HTTPException(413, "File too large") if not data.startswith(PNG): raise HTTPException(415, "Not a PNG") return dataExample 2 · A product photo upload with all checks
from fastapi import BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, UploadFilefrom fastapi.testclient import TestClient
PNG = b"\x89PNG\r\n\x1a\n"MAX_BYTES = 24products = {1: "Pen"}saved = {}app = FastAPI()
@app.post("/products/{product_id}/photo", status_code=201)async def upload_photo(product_id: int, photo: UploadFile): if product_id not in products: raise HTTPException(status_code=404, detail="Unknown product") if photo.content_type != "image/png": raise HTTPException(status_code=415, detail="PNG only") data = await photo.read(MAX_BYTES + 1) if len(data) > MAX_BYTES: raise HTTPException(status_code=413, detail="Too large") if not data.startswith(PNG): raise HTTPException(status_code=415, detail="Not a PNG") saved[product_id] = f"product-{product_id}.png" return {"stored_as": saved[product_id]}
client = TestClient(app)def send(pid, content, ctype="image/png"): res = client.post(f"/products/{pid}/photo", files={"photo": ("p.png", content, ctype)}) print(res.status_code, res.json())
send(1, PNG + b"data")send(2, PNG)send(1, PNG + b"x" * 30)send(1, b"not an image")print(saved)Output
201 {'stored_as': 'product-1.png'}
404 {'detail': 'Unknown product'}
413 {'detail': 'Too large'}
415 {'detail': 'Not a PNG'}
{1: 'product-1.png'}What this code does
- The checks run from cheapest to most expensive: product, declared type, size, content.
- Only the first upload is stored, under a name made from the product id.
Reading one byte more than the limit is enough to know the file is too big.
Cache with a TTL and invalidate on writes
A cache-aside read checks the cache, loads on a miss and stores the value with a TTL. A write deletes the key. Test expiry with a fake clock so no test has to sleep.
value = cache.get(key)if value is None: value = load_from_database(key) cache.set(key, value, ttl=60)# on a write: cache.delete(key)Example 3 · Stale data without invalidation
class FakeClock: def __init__(self, now=0.0): self.now = now
def __call__(self): return self.now
def advance(self, seconds): self.now += seconds
class TTLCache: def __init__(self, clock, default_ttl=60): self._clock = clock self._default_ttl = default_ttl self._data = {}
def get(self, key, default=None): item = self._data.get(key) if item is None: return default value, expires_at = item if self._clock() >= expires_at: del self._data[key] return default return value
def set(self, key, value, ttl=None): ttl = self._default_ttl if ttl is None else ttl self._data[key] = (value, self._clock() + ttl)
def delete(self, key): self._data.pop(key, None)
clock = FakeClock()cache = TTLCache(clock, default_ttl=60)db = {"price": 50}
def read_price(): value = cache.get("price") if value is None: value = db["price"] cache.set("price", value) return value
print(read_price())db["price"] = 60print("after update, no delete:", read_price())clock.advance(60)print("after the TTL:", read_price())
db["price"] = 70cache.delete("price")print("after update with delete:", read_price())Output
50 after update, no delete: 50 after the TTL: 60 after update with delete: 70
What this code does
- Without a delete, readers see the old price until the TTL ends.
- With a delete, the next read loads the new price at once.
Use the TTL as a safety net and the delete as the main way to keep data fresh.
Finish up
Mark this lesson complete
This is the last lesson in this section.