Structuring an App Checkpoint
25 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.
This checkpoint puts the module together in one small task app: a router, a service with a limit from settings, a repository, dependencies and an error handler. Then it tests the app with overrides.
Put each concern in its place
A small app can show every idea of this module at once. Settings decide the limits. A repository stores. A service applies the rules. Routes handle HTTP. Dependencies connect them. One handler turns domain errors into answers.
# route -> Depends(get_service)# service -> Depends(get_repo), Depends(get_settings)# errors -> one exception handlerExample 1 · A layered app from the bottom up
from dataclasses import dataclassfrom functools import lru_cachefrom fastapi import APIRouter, Depends, FastAPI, Requestfrom fastapi.responses import JSONResponsefrom fastapi.testclient import TestClientfrom pydantic import BaseModel, Field
# settings@dataclass(frozen=True)class Settings: app_env: str = "development" max_tasks: int = 3
@lru_cachedef get_settings(): return Settings()
# errorsclass AppError(Exception): status_code = 400 code = "app_error"
def __init__(self, message): self.message = message
class NotFound(AppError): status_code = 404 code = "not_found"
class LimitReached(AppError): status_code = 409 code = "limit_reached"
# repositoryclass TaskRepository: def __init__(self): self.rows = {}
def add(self, title): row = {"id": len(self.rows) + 1, "title": title} self.rows[row["id"]] = row return row
def get(self, task_id): return self.rows.get(task_id)
def count(self): return len(self.rows)
shared_repo = TaskRepository()
def get_repo(): return shared_repo
# serviceclass TaskService: def __init__(self, repo, settings): self.repo = repo self.settings = settings
def create(self, title): if self.repo.count() >= self.settings.max_tasks: raise LimitReached(f"At most {self.settings.max_tasks} tasks") return self.repo.add(title)
def read(self, task_id): row = self.repo.get(task_id) if row is None: raise NotFound(f"Task {task_id} does not exist") return row
def get_service(repo=Depends(get_repo), settings=Depends(get_settings)): return TaskService(repo, settings)
# routesclass TaskIn(BaseModel): title: str = Field(min_length=3)
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.post("", status_code=201)def create_task(body: TaskIn, service: TaskService = Depends(get_service)): return service.create(body.title)
@router.get("/{task_id}")def read_task(task_id: int, service: TaskService = Depends(get_service)): return service.read(task_id)
app = FastAPI()app.include_router(router)
@app.exception_handler(AppError)def handle_app_error(request: Request, exc: AppError): body = {"error": {"code": exc.code, "message": exc.message}} return JSONResponse(status_code=exc.status_code, content=body)
client = TestClient(app)r = client.post("/tasks", json={"title": "Write report"})print(r.status_code, r.json())print(client.get("/tasks/1").json())r = client.get("/tasks/9")print(r.status_code, r.json())print(client.post("/tasks", json={"title": "ab"}).status_code)client.post("/tasks", json={"title": "Second task"})client.post("/tasks", json={"title": "Third task"})r = client.post("/tasks", json={"title": "Fourth task"})print(r.status_code, r.json())Output
201 {'id': 1, 'title': 'Write report'}
{'id': 1, 'title': 'Write report'}
404 {'error': {'code': 'not_found', 'message': 'Task 9 does not exist'}}
422
409 {'error': {'code': 'limit_reached', 'message': 'At most 3 tasks'}}What this code does
- The limit of 3 comes from the settings and is enforced by the service.
- The 404 and 409 use the same body shape because one handler builds them.
Read the example below from the bottom up: routes ask for a service, the service gets a repository and settings, and the handler shapes the errors.
Tests replace the edges
The edges of the app are storage and settings. A test overrides them with a fresh repository and small limits, and leaves the routes and the service as they are. Each test then starts clean and does not depend on the order of other tests.
test_repo = TaskRepository()app.dependency_overrides[get_repo] = lambda: test_repoapp.dependency_overrides[get_settings] = lambda: Settings(max_tasks=1)Example 2 · Tests with fresh overrides
from dataclasses import dataclassfrom functools import lru_cachefrom fastapi import APIRouter, Depends, FastAPI, Requestfrom fastapi.responses import JSONResponsefrom fastapi.testclient import TestClientfrom pydantic import BaseModel, Field
# settings@dataclass(frozen=True)class Settings: app_env: str = "development" max_tasks: int = 3
@lru_cachedef get_settings(): return Settings()
# errorsclass AppError(Exception): status_code = 400 code = "app_error"
def __init__(self, message): self.message = message
class NotFound(AppError): status_code = 404 code = "not_found"
class LimitReached(AppError): status_code = 409 code = "limit_reached"
# repositoryclass TaskRepository: def __init__(self): self.rows = {}
def add(self, title): row = {"id": len(self.rows) + 1, "title": title} self.rows[row["id"]] = row return row
def get(self, task_id): return self.rows.get(task_id)
def count(self): return len(self.rows)
shared_repo = TaskRepository()
def get_repo(): return shared_repo
# serviceclass TaskService: def __init__(self, repo, settings): self.repo = repo self.settings = settings
def create(self, title): if self.repo.count() >= self.settings.max_tasks: raise LimitReached(f"At most {self.settings.max_tasks} tasks") return self.repo.add(title)
def read(self, task_id): row = self.repo.get(task_id) if row is None: raise NotFound(f"Task {task_id} does not exist") return row
def get_service(repo=Depends(get_repo), settings=Depends(get_settings)): return TaskService(repo, settings)
# routesclass TaskIn(BaseModel): title: str = Field(min_length=3)
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.post("", status_code=201)def create_task(body: TaskIn, service: TaskService = Depends(get_service)): return service.create(body.title)
@router.get("/{task_id}")def read_task(task_id: int, service: TaskService = Depends(get_service)): return service.read(task_id)
app = FastAPI()app.include_router(router)
@app.exception_handler(AppError)def handle_app_error(request: Request, exc: AppError): body = {"error": {"code": exc.code, "message": exc.message}} return JSONResponse(status_code=exc.status_code, content=body)
client = TestClient(app)
def run_case(limit): fresh = TaskRepository() app.dependency_overrides[get_repo] = lambda: fresh app.dependency_overrides[get_settings] = lambda: Settings(max_tasks=limit) codes = [client.post("/tasks", json={"title": f"Task {n}"}).status_code for n in range(3)] app.dependency_overrides.clear() return codes
print(run_case(1))print(run_case(2))print(run_case(5))print(shared_repo.count())Output
[201, 409, 409] [201, 201, 409] [201, 201, 201] 0
What this code does
- Each case gets a new repository and its own limit, so the cases do not affect each other.
- The shared repository stayed empty, because the overrides replaced it.
Always clear the overrides in the end, so the next test sees the real dependencies.
Review checklist for a structured app
Before you call an app done, check a few things. Routes have no business rules. Services have no HTTP. Only repositories know about storage. Settings are read once. Every error has the same body. Every layer has at least one test.
def test_limit_is_409(): ...
def test_missing_task_is_404(): ...
def test_validation_error_is_422(): ...Example 3 · One body shape for every failure
from dataclasses import dataclassfrom functools import lru_cachefrom fastapi import APIRouter, Depends, FastAPI, Requestfrom fastapi.responses import JSONResponsefrom fastapi.testclient import TestClientfrom pydantic import BaseModel, Field
# settings@dataclass(frozen=True)class Settings: app_env: str = "development" max_tasks: int = 3
@lru_cachedef get_settings(): return Settings()
# errorsclass AppError(Exception): status_code = 400 code = "app_error"
def __init__(self, message): self.message = message
class NotFound(AppError): status_code = 404 code = "not_found"
class LimitReached(AppError): status_code = 409 code = "limit_reached"
# repositoryclass TaskRepository: def __init__(self): self.rows = {}
def add(self, title): row = {"id": len(self.rows) + 1, "title": title} self.rows[row["id"]] = row return row
def get(self, task_id): return self.rows.get(task_id)
def count(self): return len(self.rows)
shared_repo = TaskRepository()
def get_repo(): return shared_repo
# serviceclass TaskService: def __init__(self, repo, settings): self.repo = repo self.settings = settings
def create(self, title): if self.repo.count() >= self.settings.max_tasks: raise LimitReached(f"At most {self.settings.max_tasks} tasks") return self.repo.add(title)
def read(self, task_id): row = self.repo.get(task_id) if row is None: raise NotFound(f"Task {task_id} does not exist") return row
def get_service(repo=Depends(get_repo), settings=Depends(get_settings)): return TaskService(repo, settings)
# routesclass TaskIn(BaseModel): title: str = Field(min_length=3)
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.post("", status_code=201)def create_task(body: TaskIn, service: TaskService = Depends(get_service)): return service.create(body.title)
@router.get("/{task_id}")def read_task(task_id: int, service: TaskService = Depends(get_service)): return service.read(task_id)
app = FastAPI()app.include_router(router)
@app.exception_handler(AppError)def handle_app_error(request: Request, exc: AppError): body = {"error": {"code": exc.code, "message": exc.message}} return JSONResponse(status_code=exc.status_code, content=body)
client = TestClient(app)client.post("/tasks", json={"title": "Seed task"})app.dependency_overrides[get_settings] = lambda: Settings(max_tasks=1)
checks = [ ("limit", client.post("/tasks", json={"title": "Another one"})), ("missing", client.get("/tasks/77")),]for name, r in checks: body = r.json()["error"] print(name, r.status_code, body["code"], sorted(body))
invalid = client.post("/tasks", json={"title": "x"})print("invalid", invalid.status_code, sorted(invalid.json()))app.dependency_overrides.clear()Output
limit 409 limit_reached ['code', 'message'] missing 404 not_found ['code', 'message'] invalid 422 ['detail']
What this code does
- Your own errors share the keys
codeandmessage. - The default 422 still uses
detail; the lesson on error handling showed how to replace it with the same shape.
These three tests cover the three kinds of failure: a business rule, a missing record and bad input.
Finish up
Mark complete & continue
Next: Background Tasks