REST API Patterns

Resource-based URLs, HTTP method conventions, route organisation by domain, and explicit parameter extraction.

Show:

Status: Extracted
Purpose: REST API design principles, route organization, and endpoint patterns.

REST API Principles

RESTful Design

Principles:

Resource-based URLs: URLs represent resources, not actions
HTTP Methods: Use standard HTTP methods (GET, POST, PUT, PATCH, DELETE)
Stateless: Each request contains all information needed
Uniform Interface: Consistent patterns across all endpoints

Route Organization

Route Structure

Pattern: /api/v1/{resource}/{identifier}/{sub-resource}

Examples:

GET /api/v1/accounts - List accounts
GET /api/v1/accounts/{account_id} - Get account by ID
POST /api/v1/accounts - Create account
PUT /api/v1/accounts/{account_id} - Update account
DELETE /api/v1/accounts/{account_id} - Delete account
GET /api/v1/accounts/{account_id}/messages - Get account messages

Route Organization by Domain

Pattern:

api/
├── routes/
│   ├── accounts.py         # Account routes
│   ├── messages.py         # Message routes
│   └── admin.py            # Admin routes

Route Module Structure:

# api/routes/accounts.py
from fastapi import APIRouter, Depends
 
router = APIRouter(prefix="/api/v1/accounts", tags=["accounts"])
 
@router.get("/")
async def list_accounts(...):
    """List accounts."""
    ...
 
@router.post("/")
async def create_account(...):
    """Create account."""
    ...
 
@router.get("/{account_id}")
async def get_account(account_id: str, ...):
    """Get account by ID."""
    ...

HTTP Methods

Standard HTTP Methods

GET: Retrieve resources (read operations)

GET /api/v1/accounts - List resources
GET /api/v1/accounts/{id} - Get single resource

POST: Create resources (write operations)

POST /api/v1/accounts - Create resource

PUT: Full update (write operations)

PUT /api/v1/accounts/{id} - Full update

PATCH: Partial update (write operations)

PATCH /api/v1/accounts/{id} - Partial update

DELETE: Delete resources (write operations)

DELETE /api/v1/accounts/{id} - Delete resource

Endpoint Patterns

CRUD Operations

Create:

@router.post("/")
async def create_account(
    request: CreateAccountRequest,
    uow: IUnitOfWork = Depends(get_unit_of_work),
) -> AccountResponse:
    """Create account."""
    ...

Read:

@router.get("/{account_id}")
async def get_account(
    account_id: str = Path(...),
    account_repository: IAccountRepository = Depends(get_account_repository),
) -> AccountResponse:
    """Get account by ID."""
    ...

Update:

@router.put("/{account_id}")
async def update_account(
    account_id: str = Path(...),
    request: UpdateAccountRequest = Body(...),
    uow: IUnitOfWork = Depends(get_unit_of_work),
) -> AccountResponse:
    """Update account."""
    ...

Delete:

@router.delete("/{account_id}")
async def delete_account(
    account_id: str = Path(...),
    uow: IUnitOfWork = Depends(get_unit_of_work),
) -> None:
    """Delete account."""
    ...

Explicit Parameter Extraction

Parameter Patterns

Rationale:

Makes parameter sources explicit (Path(), Body(), Query())
Improves code readability and maintainability
Reduces ambiguity about parameter origins
Better IDE support and type checking

Pattern:

@router.get("/{account_id}")
async def get_account(
    account_id: str = Path(..., description="Account ID"),
    include_messages: bool = Query(False, description="Include messages"),
    limit: int = Query(10, ge=1, le=100, description="Limit results"),
) -> AccountResponse:
    """Get account by ID."""
    ...

Response Descriptions Helper

Standardized Response Documentation

Rationale:

Standardizes OpenAPI response documentation
Reduces duplication across route definitions
Ensures consistent error response documentation
Improves API documentation clarity for consumers

Implementation:

# api/responses.py
def account_responses() -> dict:
    """Standard response descriptions for account endpoints."""
    return {
        200: {"description": "Success"},
        400: {"description": "Bad Request"},
        404: {"description": "Account not found"},
        500: {"description": "Internal server error"},
    }
 
# Usage in route
@router.get(
    "/{account_id}",
    responses=account_responses(),
)
async def get_account(...):
    """Get account by ID."""
    ...

Registration & Execution Order

# api/middleware/__init__.py
def register_middleware(app: FastAPI, settings: Settings) -> None:
    # Registered first = executes LAST (outermost)
    register_cors_middleware(app, settings)
 
    # Registered second = executes FOURTH
    app.add_middleware(JWTAuthMiddleware, jwt_service=get_jwt_service())
 
    # Registered third = executes THIRD
    app.add_middleware(RequestLoggingMiddleware, logging_service=get_logging_service())
 
    # Registered fourth = executes SECOND
    app.add_middleware(RequestContextMiddleware)
 
    # Exception handlers — closest to routes
    register_exception_handlers(app)

Actual execution order per request:

RequestContextMiddleware — sets request ID, actor context shell, timing start
RequestLoggingMiddleware — logs request start (has request ID available)
JWTAuthMiddleware — validates token, populates actor context
Route handler
Exception handlers (on error)

On response (unwinding):

Route handler returns
JWTAuthMiddleware — no action on response
RequestLoggingMiddleware — logs response with status code and duration
RequestContextMiddleware — adds X-Request-ID response header
CORS — adds Access-Control-Allow-* headers

Each Middleware

1. CORS (`api/middleware/cors.py`)

from starlette.middleware.cors import CORSMiddleware
 
def register_cors_middleware(app: FastAPI, settings: Settings) -> None:
    origins = [o.strip() for o in settings.cors_allowed_origins.split(",")]
    app.add_middleware(
        CORSMiddleware,
        allow_origins=origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

Must be registered first so it's the outermost middleware and handles OPTIONS preflight before auth middleware runs.

2. Request Context (`api/middleware/request_context.py`)

Sets up request.state.ctx — a dict shared across the request lifecycle. Everything else reads from here.

import uuid, time
from starlette.middleware.base import BaseHTTPMiddleware
 
class RequestContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request.state.ctx = {
            "request_id": str(uuid.uuid4()),
            "path": request.url.path,
            "method": request.method,
            "start_time": time.perf_counter(),
            "client_ip": request.client.host if request.client else None,
            "actor": {
                "actor_id": None,
                "actor_type": "anonymous",
                "roles": (),
                "account_ids": (),
            },
        }
        response = await call_next(request)
        response.headers["X-Request-ID"] = request.state.ctx["request_id"]
        return response

3. JWT Auth (`api/middleware/jwt_auth.py`)

Validates the Authorization: Bearer <token> header and updates request.state.ctx["actor"] with the authenticated identity.

class JWTAuthMiddleware(BaseHTTPMiddleware):
    PUBLIC_PATHS = {"/health", "/docs", "/openapi.json", "/auth/login", "/auth/signup"}
 
    def __init__(self, app, jwt_service: JWTService):
        super().__init__(app)
        self._jwt = jwt_service
 
    async def dispatch(self, request: Request, call_next):
        if request.url.path in self.PUBLIC_PATHS:
            return await call_next(request)
 
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return await call_next(request)  # Let route handle 401 if needed
 
        token = auth_header.removeprefix("Bearer ")
        if not self._jwt.validate_token(token):
            return JSONResponse({"detail": "Invalid or expired token"}, status_code=401)
 
        claims = self._jwt.extract_claims(token)
        request.state.ctx["actor"] = {
            "actor_id": claims["account_id"],
            "actor_type": claims.get("account_type", "user"),
            "roles": tuple(claims.get("roles", [])),
            "account_ids": (claims["account_id"],),
        }
        request.state.ctx["jwt"] = claims
        return await call_next(request)

4. Request Logging (`api/middleware/logging.py`)

Logs structured request start and end events. Uses request.state.ctx for request metadata.

class RequestLoggingMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, logging_service: LoggingService):
        super().__init__(app)
        self._logger = logging_service
 
    async def dispatch(self, request: Request, call_next):
        ctx = getattr(request.state, "ctx", {})
        self._logger.info("request.start", extra={
            "request_id": ctx.get("request_id"),
            "path": ctx.get("path"),
            "method": ctx.get("method"),
        })
 
        response = await call_next(request)
 
        duration_ms = (time.perf_counter() - ctx.get("start_time", 0)) * 1000
        level = "error" if response.status_code >= 500 else \
                "warning" if response.status_code >= 400 else "info"
 
        getattr(self._logger, level)("request.end", extra={
            "request_id": ctx.get("request_id"),
            "status_code": response.status_code,
            "duration_ms": round(duration_ms, 2),
        })
        return response

5. Exception Handlers (`api/middleware/error_handler.py`)

Maps exceptions to structured JSON responses. Registered via register_exception_handlers(app).

def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(EntityNotFoundError)
    async def not_found_handler(request, exc):
        _log_warning("entity.not_found", exc, request)
        return JSONResponse({"detail": str(exc)}, status_code=404)
 
    @app.exception_handler(NotAuthorizedError)
    async def not_authorized_handler(request, exc):
        _log_warning("not.authorized", exc, request)
        return JSONResponse({"detail": str(exc)}, status_code=403)
 
    @app.exception_handler(RequestValidationError)
    async def validation_handler(request, exc):
        return JSONResponse({"detail": exc.errors()}, status_code=422)
 
    @app.exception_handler(Exception)
    async def catchall_handler(request, exc):
        _log_error("unhandled.exception", exc, request)
        ctx = getattr(request.state, "ctx", {})
        return JSONResponse(
            {"detail": "Internal server error", "request_id": ctx.get("request_id")},
            status_code=500,
        )

Webhook routes should return 200 on all errors to prevent provider retries:

@app.exception_handler(Exception)
async def webhook_error_handler(request, exc):
    if request.url.path.startswith("/webhooks/"):
        return JSONResponse({"status": "received"}, status_code=200)
    raise exc

Rules

Never handle auth, logging, or error formatting inside route handlers — that belongs in middleware.
Exception handlers are not technically ASGI middleware but are registered alongside them for the same reason: separation of concerns.
Adding a new middleware? Add it to register_middleware() in api/middleware/__init__.py and document its execution order position in this file.

Session Management Request & Response Patterns