Free Ebook cover Flask Essentials: Practical Backend Patterns for Small Services

Flask Essentials: Practical Backend Patterns for Small Services

New course

14 pages

Flask Essentials: Request Lifecycle, Context, and Response Design

Capítulo 4

Estimated reading time: 9 minutes

+ Exercise

From Request Arrival to Response: The High-Level Flow

When a request hits your Flask service, Flask (and the underlying WSGI server) runs a predictable sequence. Understanding this sequence helps you place authentication, logging, metrics, and resource cleanup in the right hooks.

  • WSGI server receives HTTP request and calls your Flask application object.
  • Request context is created (Flask binds request-specific globals like request, g, and session).
  • URL matching (routing) selects the view function for the request.
  • before_request handlers run (zero or more).
  • View function runs and returns a response value (string/dict/Response/etc.).
  • Flask builds a Response (serialization, status code, headers).
  • after_request handlers run (zero or more) to adjust the final Response.
  • Response is returned to the WSGI server.
  • teardown_request handlers run (always) to clean up resources, even on errors.

Important: after_request runs only if a response object exists, while teardown_request runs regardless of whether the request succeeded or failed (it receives an exception if one occurred).

Request Context: request, g, and current_app

What “context” means in Flask

Flask uses context locals (implemented via Werkzeug) so you can access request, g, and current_app without passing them through every function call. These objects are only valid during an active request (or an application context you explicitly push in CLI/tasks).

  • request: incoming HTTP data (method, headers, args, JSON body, etc.).
  • g: a per-request scratchpad for storing request-scoped objects (correlation ID, DB session, timers).
  • current_app: the active Flask app instance; use it for config and app-level resources (loggers, extensions).

Practical rule of thumb

  • Put request-specific data in g.
  • Put app-wide configuration and shared singletons behind current_app (or extensions).
  • Never store request-specific state in module globals; it will leak across requests/threads/workers.

Lifecycle Hooks: before_request, after_request, teardown_request

When to use each hook

HookRunsUse forAvoid
before_requestBefore viewCorrelation ID, auth checks, start timers, open request-scoped resourcesHeavy work, network calls that can be deferred, modifying global state
after_requestAfter view, with ResponseAdd headers, set cookies, attach metrics headers, normalize responseChanging request data, doing cleanup that must run on errors
teardown_requestAlways, after response or errorClose DB sessions, release locks, rollback transactions, final cleanupRaising new exceptions (can mask original errors)

Correlation IDs: propagate and generate

A correlation ID lets you trace a request across logs and downstream calls. A common pattern is:

  • If the client sends X-Request-ID, accept it (after basic validation).
  • Otherwise generate a new ID.
  • Store it in g and return it in the response header.
import time
import uuid
from flask import Flask, g, request


def install_request_hooks(app: Flask) -> None:
    @app.before_request
    def assign_correlation_and_start_timer():
        incoming = request.headers.get("X-Request-ID", "").strip()
        # Keep validation simple for small services: limit length and characters.
        if incoming and len(incoming) <= 128:
            g.correlation_id = incoming
        else:
            g.correlation_id = uuid.uuid4().hex

        g.request_start_ns = time.perf_counter_ns()

    @app.after_request
    def add_observability_headers(response):
        response.headers["X-Request-ID"] = getattr(g, "correlation_id", "")

        start_ns = getattr(g, "request_start_ns", None)
        if start_ns is not None:
            duration_ms = (time.perf_counter_ns() - start_ns) / 1_000_000
            response.headers["Server-Timing"] = f"app;dur={duration_ms:.2f}"
        return response

Responsibility tips:

Continue in our app.

You can listen to the audiobook with the screen off, receive a free certificate for this course, and also have access to 5,000 other free online courses.

Or continue reading below...
Download App

Download the app

  • Keep before_request fast; avoid blocking I/O.
  • Use headers like Server-Timing for lightweight timing; emit detailed metrics to your monitoring system elsewhere.
  • Don’t trust arbitrary client-provided IDs without constraints; at minimum cap length to prevent log injection and header bloat.

Teardown: cleanup that runs even on errors

teardown_request receives an exception argument (or None). Use it to ensure resources are always released.

from flask import Flask, g


def install_teardown(app: Flask) -> None:
    @app.teardown_request
    def cleanup(exception):
        # Never raise from teardown; log if needed.
        session = g.pop("db_session", None)
        if session is not None:
            try:
                if exception is not None:
                    session.rollback()
                session.close()
            except Exception:
                # Avoid masking the original exception.
                current_app.logger.exception("Failed to cleanup db session")

Pattern: Request-Scoped Resources (DB Session Example)

Small services often need a per-request database session or unit-of-work. A safe pattern is:

  1. Create/open the resource in before_request.
  2. Store it on g.
  3. Use it in views/services via a helper accessor.
  4. Commit/rollback appropriately.
  5. Close it in teardown_request.

Step-by-step implementation

1) Provide a session factory on the app

How you create the factory depends on your DB library. The key is: the factory is app-wide, but the session instance is request-scoped.

from flask import Flask


def create_session_factory():
    # Placeholder: return a callable that creates a new session.
    # For SQLAlchemy, this might be sessionmaker(bind=engine).
    ...


def init_db(app: Flask) -> None:
    app.session_factory = create_session_factory()

2) Open the session per request

from flask import Flask, g, current_app


def install_db_session_hooks(app: Flask) -> None:
    @app.before_request
    def open_db_session():
        g.db_session = current_app.session_factory()

    @app.teardown_request
    def close_db_session(exception):
        session = g.pop("db_session", None)
        if session is None:
            return
        try:
            if exception is not None:
                session.rollback()
            session.close()
        except Exception:
            current_app.logger.exception("DB session cleanup failed")

3) Accessor helper to avoid passing g everywhere

from flask import g


def db_session():
    session = getattr(g, "db_session", None)
    if session is None:
        raise RuntimeError("DB session not initialized for this request")
    return session

4) Commit strategy: explicit is safer

A common small-service approach is to commit explicitly in the view (or service layer) when the operation succeeds. This keeps transaction boundaries obvious.

from flask import Blueprint, request

bp = Blueprint("items", __name__)

@bp.post("/items")
def create_item():
    payload = request.get_json(silent=True) or {}
    name = (payload.get("name") or "").strip()
    if not name:
        return {"error": "name is required"}, 400

    s = db_session()
    item = Item(name=name)
    s.add(item)
    s.commit()

    return {"id": item.id, "name": item.name}, 201

If an exception occurs before commit(), teardown_request will rollback and close the session. If you commit and then later fail, you may need compensating logic; keep handlers small to reduce that risk.

Response Design for Small Services

JSON serialization: be explicit about what you return

Flask can convert a Python dict to JSON automatically in modern versions, but you still need to control:

  • Which fields are exposed (avoid returning ORM objects directly).
  • How you serialize dates/decimals (convert to strings or numbers intentionally).
  • Error payload shape consistency.
from datetime import datetime, timezone


def item_to_dict(item):
    return {
        "id": item.id,
        "name": item.name,
        "created_at": item.created_at.replace(tzinfo=timezone.utc).isoformat(),
    }

Status codes: map outcomes to HTTP semantics

ScenarioStatusNotes
Successful read200Return representation
Created201Include Location header when possible
Accepted for async processing202Return job ID/status URL
No content204No response body
Validation error400Client sent invalid data
Auth required/failed401Often with WWW-Authenticate
Forbidden403Authenticated but not allowed
Not found404Resource does not exist
Conflict409Uniqueness/ETag conflict
Server error500Don’t leak internals

Headers: small additions that improve operability

  • X-Request-ID: correlation/tracing.
  • Server-Timing: coarse performance insight.
  • Cache-Control: prevent caching for sensitive endpoints, or enable caching for stable reads.
  • Location: for 201 Created responses.
from flask import url_for

@bp.post("/items")
def create_item():
    ...
    s.commit()

    body = item_to_dict(item)
    headers = {"Location": url_for("items.get_item", item_id=item.id, _external=False)}
    return body, 201, headers

Pagination basics: limit, cursor/offset, and metadata

For small services, keep pagination simple and consistent. A typical pattern uses limit and cursor (or offset) query parameters, and returns metadata.

Constraints to enforce:

  • Default limit (e.g., 20).
  • Maximum limit (e.g., 100) to protect the service.
  • Stable ordering (e.g., by id or created_at).
from flask import request

DEFAULT_LIMIT = 20
MAX_LIMIT = 100

@bp.get("/items")
def list_items():
    limit = request.args.get("limit", type=int) or DEFAULT_LIMIT
    limit = max(1, min(limit, MAX_LIMIT))

    cursor = request.args.get("cursor", type=int)

    q = db_session().query(Item).order_by(Item.id.asc())
    if cursor is not None:
        q = q.filter(Item.id > cursor)

    rows = q.limit(limit + 1).all()
    has_more = len(rows) > limit
    rows = rows[:limit]

    next_cursor = rows[-1].id if (has_more and rows) else None

    return {
        "items": [item_to_dict(r) for r in rows],
        "page": {
            "limit": limit,
            "next_cursor": next_cursor,
            "has_more": has_more,
        },
    }, 200

This cursor approach avoids some pitfalls of offset pagination (like skipping/duplicating items when new rows are inserted), while still being easy to implement.

Content Negotiation Constraints (Keep It Simple)

Many small services only support JSON. You can still be explicit about it to avoid surprising clients and to prevent accidental HTML responses.

Enforce JSON responses and basic Accept handling

  • If the client sends Accept: application/json or */*, return JSON.
  • If the client requests an unsupported type, return 406 Not Acceptable.
from flask import request

SUPPORTED = "application/json"

@app.before_request
def enforce_accept_header():
    accept = request.headers.get("Accept")
    if not accept or "*/*" in accept or SUPPORTED in accept:
        return None
    return {"error": "Only application/json is supported"}, 406

Enforce JSON request bodies where required

For endpoints that require JSON input, validate Content-Type and parse safely.

from flask import request

@bp.post("/items")
def create_item():
    if request.mimetype != "application/json":
        return {"error": "Content-Type must be application/json"}, 415

    payload = request.get_json(silent=True)
    if payload is None:
        return {"error": "Invalid JSON"}, 400

    ...

Designing Error Responses That Work with the Lifecycle

Because hooks and teardown run around your view, error responses should be consistent and should not break cleanup. A practical pattern is to raise exceptions for expected errors and register handlers that return JSON.

from werkzeug.exceptions import HTTPException

@app.errorhandler(HTTPException)
def handle_http_exception(e: HTTPException):
    # e.code is the HTTP status
    return {
        "error": e.name,
        "message": e.description,
        "request_id": getattr(g, "correlation_id", None),
    }, e.code

@app.errorhandler(Exception)
def handle_unexpected_exception(e: Exception):
    current_app.logger.exception("Unhandled error")
    return {
        "error": "Internal Server Error",
        "message": "An unexpected error occurred",
        "request_id": getattr(g, "correlation_id", None),
    }, 500

This keeps responses JSON-only, includes the correlation ID for support/debugging, and still allows teardown_request to run and clean up request-scoped resources.

Now answer the exercise about the content:

In a Flask service, where should you place cleanup logic (like closing a per-request DB session) to ensure it runs even if an error occurs before a Response is produced?

You are right! Congratulations, now go to the next page

You missed! Try again.

after_request only runs when a response exists, but teardown_request runs regardless of success or failure and receives the exception (if any). This makes it the right place for guaranteed cleanup like closing DB sessions.

Next chapter

Flask Essentials: Error Handling Patterns and Consistent API Errors

Arrow Right Icon
Download the app to earn free Certification and listen to the courses in the background, even with the screen off.