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, andsession). - 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
| Hook | Runs | Use for | Avoid |
|---|---|---|---|
before_request | Before view | Correlation ID, auth checks, start timers, open request-scoped resources | Heavy work, network calls that can be deferred, modifying global state |
after_request | After view, with Response | Add headers, set cookies, attach metrics headers, normalize response | Changing request data, doing cleanup that must run on errors |
teardown_request | Always, after response or error | Close DB sessions, release locks, rollback transactions, final cleanup | Raising 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
gand 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 responseResponsibility 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 the app
- Keep
before_requestfast; avoid blocking I/O. - Use headers like
Server-Timingfor 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:
- Create/open the resource in
before_request. - Store it on
g. - Use it in views/services via a helper accessor.
- Commit/rollback appropriately.
- 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 session4) 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}, 201If 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
| Scenario | Status | Notes |
|---|---|---|
| Successful read | 200 | Return representation |
| Created | 201 | Include Location header when possible |
| Accepted for async processing | 202 | Return job ID/status URL |
| No content | 204 | No response body |
| Validation error | 400 | Client sent invalid data |
| Auth required/failed | 401 | Often with WWW-Authenticate |
| Forbidden | 403 | Authenticated but not allowed |
| Not found | 404 | Resource does not exist |
| Conflict | 409 | Uniqueness/ETag conflict |
| Server error | 500 | Don’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: for201 Createdresponses.
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, headersPagination 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
idorcreated_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,
},
}, 200This 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/jsonor*/*, 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"}, 406Enforce 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),
}, 500This keeps responses JSON-only, includes the correlation ID for support/debugging, and still allows teardown_request to run and clean up request-scoped resources.