Flask Essentials: Logging, Observability, and Debuggable Services

Why logging and observability matter in small Flask services

In a small service, most production issues are not “hard bugs” but “hard-to-see” behaviors: a slow external dependency, an unexpected input shape, a missing header, or a code path that only triggers under specific data. Logging and observability make these behaviors visible. The goal is not to log everything; it is to log the right events with enough context to answer: what happened, to whom, when, and why, without leaking sensitive data.

Good service logs are: (1) structured (machine-parsable), (2) contextual (request ID, user ID when available), (3) leveled (debug/info/warn/error), and (4) routed to appropriate handlers for local development and production.

Structured logging with request context

Choose a log format: JSON in production, readable locally

Structured logs are typically JSON objects per line. They are easy to search and aggregate in log systems. Locally, you may prefer a human-friendly format, but keeping the same fields in both environments helps.

Step-by-step: configure logging with a context filter

The pattern below uses Python’s standard logging and a filter that injects request-scoped fields. It works with Flask’s request context and degrades gracefully outside requests (e.g., CLI tasks).

import json, logging, sys, time, uuid, contextvars, re, os, traceback
from flask import Flask, g, request

# Context variables work across async boundaries better than globals
request_id_var = contextvars.ContextVar("request_id", default=None)
user_id_var = contextvars.ContextVar("user_id", default=None)

SENSITIVE_KEYS = {"password", "pass", "token", "access_token", "refresh_token", "authorization", "api_key", "secret"}

class ContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get() or "-"
        record.user_id = user_id_var.get() or "-"
        record.path = getattr(record, "path", None) or (request.path if request else "-")
        record.method = getattr(record, "method", None) or (request.method if request else "-")
        return True

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
            "request_id": getattr(record, "request_id", "-"),
            "user_id": getattr(record, "user_id", "-"),
            "method": getattr(record, "method", "-"),
            "path": getattr(record, "path", "-"),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        # Include extra structured fields if present
        for k, v in record.__dict__.items():
            if k.startswith("_") or k in payload or k in ("args", "msg", "exc_info", "exc_text", "stack_info"):
                continue
            # Keep it JSON-serializable
            try:
                json.dumps(v)
                payload[k] = v
            except TypeError:
                payload[k] = repr(v)
        return json.dumps(payload, ensure_ascii=False)

def configure_logging(app: Flask) -> None:
    root = logging.getLogger()
    root.handlers.clear()

    level = app.config.get("LOG_LEVEL", "INFO").upper()
    root.setLevel(level)

    handler = logging.StreamHandler(sys.stdout)
    handler.addFilter(ContextFilter())

    if app.config.get("LOG_FORMAT", "json") == "json":
        handler.setFormatter(JsonFormatter())
    else:
        handler.setFormatter(logging.Formatter(
            "%(asctime)s %(levelname)s %(name)s request_id=%(request_id)s user_id=%(user_id)s %(method)s %(path)s - %(message)s"
        ))

    root.addHandler(handler)

    # Reduce noisy loggers if needed
    logging.getLogger("werkzeug").setLevel(app.config.get("WERKZEUG_LOG_LEVEL", "WARNING"))

def create_app():
    app = Flask(__name__)
    # app.config[...] set elsewhere
    configure_logging(app)
    return app

Attach request ID and user ID

Request IDs allow you to correlate all logs for a single request across middleware, handlers, and external calls. Prefer accepting an incoming request ID header (from an API gateway) and generating one if missing. User ID should be set only after authentication is known; if you don’t have a user concept, omit it.

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

from flask import Flask, g, request
import uuid, time, logging

log = logging.getLogger(__name__)

def init_request_context(app: Flask) -> None:
    @app.before_request
    def _before_request():
        rid = request.headers.get("X-Request-ID") or str(uuid.uuid4())
        request_id_var.set(rid)
        g.request_start = time.perf_counter()

        # Example: set user_id_var when available (depends on your auth)
        # user_id_var.set(getattr(g, "user_id", None) or "-")

    @app.after_request
    def _after_request(response):
        # Propagate request id back to clients for support/debugging
        response.headers["X-Request-ID"] = request_id_var.get() or "-"
        return response

If you set the user ID later (e.g., after auth), update the context variable at that point:

def set_user_context(user_id: str | None):
    if user_id:
        user_id_var.set(str(user_id))

Log levels and when to use them

Default production level is usually INFO. Use DEBUG temporarily and selectively (see “increase verbosity safely”).

Handlers for local development vs production

Local development

• Stream logs to console (StreamHandler).
• Readable format is fine, but keep the same fields (request_id, user_id, path, method).
• Keep Werkzeug request logs at INFO if you want access logs; otherwise reduce to WARNING.

Production

• Prefer JSON logs to stdout/stderr; let the platform collect them (containers, systemd, PaaS).
• Avoid file handlers inside containers unless you manage rotation.
• Consider separate loggers/handlers for audit events vs application events if you have compliance needs.

If you do need file logging (e.g., VM without centralized logging), use rotation:

from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler("/var/log/app.log", maxBytes=10_000_000, backupCount=5)
handler.setFormatter(JsonFormatter())

What to log (and what not to log)

What to log: boundaries, decisions, and dependencies

• Inputs at boundaries: request method/path, query params, content type, payload shape/keys, and key identifiers (order_id, account_id). Prefer logging summaries over full payloads.
• Key decisions: which branch was taken (e.g., “used cached value”, “applied fallback pricing”, “skipped optional enrichment”).
• External calls: dependency name, endpoint/host, latency, status code, retry count, and high-level error category.
• Performance timing: total request duration, and optionally per-step durations for slow endpoints.
• Correlation identifiers: request_id always; user_id when available; also upstream IDs (trace IDs) if present.

What not to log: secrets and sensitive data

• Passwords, password hashes, one-time codes.
• Access tokens, refresh tokens, API keys, session cookies.
• Authorization headers, full JWTs.
• Full credit card numbers, bank details, private keys.
• Full request/response bodies when they may contain personal data; log only necessary fields, and redact aggressively.

Also avoid logging raw exception objects that may include sensitive payloads; log sanitized context plus stack traces for server-side debugging.

Request/response logging with redaction and timing

Redaction utilities

Redaction should be applied to headers and JSON-like payloads. The goal is to preserve debugging value while removing secrets.

def redact_headers(headers: dict) -> dict:
    out = {}
    for k, v in headers.items():
        lk = k.lower()
        if lk in ("authorization", "cookie", "set-cookie"):
            out[k] = "[REDACTED]"
        else:
            out[k] = v
    return out

def redact_obj(obj):
    if isinstance(obj, dict):
        redacted = {}
        for k, v in obj.items():
            if str(k).lower() in SENSITIVE_KEYS:
                redacted[k] = "[REDACTED]"
            else:
                redacted[k] = redact_obj(v)
        return redacted
    if isinstance(obj, list):
        return [redact_obj(x) for x in obj]
    return obj

Step-by-step: log request start and request end summary

Logging every request and response body can be noisy and risky. A practical default is: log a request start at DEBUG (optional), and always log a request completion summary at INFO with duration and status code.

import logging, time
from flask import Flask, request, g

log = logging.getLogger("http")

def init_http_logging(app: Flask) -> None:
    @app.before_request
    def _log_request_start():
        g.request_start = time.perf_counter()
        # Optional: only in debug level
        if log.isEnabledFor(logging.DEBUG):
            log.debug(
                "request_started",
                extra={
                    "event": "request_started",
                    "query": redact_obj(request.args.to_dict(flat=False)),
                    "headers": redact_headers(dict(request.headers)),
                },
            )

    @app.after_request
    def _log_request_end(response):
        dur_ms = None
        if hasattr(g, "request_start"):
            dur_ms = round((time.perf_counter() - g.request_start) * 1000, 2)

        log.info(
            "request_completed",
            extra={
                "event": "request_completed",
                "status": response.status_code,
                "duration_ms": dur_ms,
                "content_type": response.content_type,
            },
        )
        return response

If you need response logging for a specific endpoint during debugging, log only metadata (status, size) and a redacted subset of the body.

Timing internal steps

For slow endpoints, total duration is not enough. Add step timers around external calls and expensive operations.

import time, logging
log = logging.getLogger(__name__)

class StepTimer:
    def __init__(self, name: str):
        self.name = name
        self.t0 = None
    def __enter__(self):
        self.t0 = time.perf_counter()
        return self
    def __exit__(self, exc_type, exc, tb):
        ms = round((time.perf_counter() - self.t0) * 1000, 2)
        log.info("step_timing", extra={"event": "step_timing", "step": self.name, "duration_ms": ms})

# Usage:
# with StepTimer("fetch_customer"):
#     customer = client.get_customer(customer_id)

Logging external calls (without leaking data)

When calling external services, log: dependency name, operation, URL host/path (not full query if it contains secrets), status, latency, and retry count. Avoid logging full request/response bodies unless you have a safe redaction layer and a strong reason.

import logging, time, urllib.parse
log = logging.getLogger("deps")

def safe_url(url: str) -> str:
    parts = urllib.parse.urlsplit(url)
    # Drop query and fragment by default
    return urllib.parse.urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))

def log_dependency_call(dep: str, url: str, method: str, status: int | None, duration_ms: float, retries: int = 0, err: str | None = None):
    log.info(
        "dependency_call",
        extra={
            "event": "dependency_call",
            "dependency": dep,
            "method": method,
            "url": safe_url(url),
            "status": status,
            "duration_ms": duration_ms,
            "retries": retries,
            "error": err,
        },
    )

Safe verbosity increases during troubleshooting

Turning on global DEBUG in production can leak data and overwhelm log storage. Prefer targeted approaches:

• Per-request debug: allow a privileged header (e.g., X-Debug-Log: 1) only for internal traffic, and only enable extra logs for that request ID.
• Per-logger levels: keep root at INFO, set deps or a specific module to DEBUG.
• Sampling: log detailed payload summaries for 1% of requests or only for slow requests.

Example: per-request debug flag stored in context and used to gate extra logs:

debug_log_var = contextvars.ContextVar("debug_log", default=False)

def init_debug_toggle(app: Flask) -> None:
    @app.before_request
    def _debug_toggle():
        # Only enable if you trust the caller; enforce via network/auth controls
        debug_log_var.set(request.headers.get("X-Debug-Log") == "1")

def debug_enabled() -> bool:
    return bool(debug_log_var.get())

# Usage inside handlers/services
if debug_enabled():
    logging.getLogger(__name__).debug("debug_details", extra={"event": "debug_details", "inputs": "...redacted..."})

Practical troubleshooting workflow

1) Reproduce the issue reliably

• Capture the request ID (from X-Request-ID response header) and the exact inputs that triggered the issue (redacted if needed).
• Reproduce locally or in a staging environment with the same configuration and dependency versions.
• If it is timing-related, reproduce with similar load or add artificial latency to dependencies.

2) Inspect logs using correlation fields

• Filter by request_id to see the full story of a single request.
• Look for dependency_call events and compare their duration_ms to total request time.
• Find the first warning/error in the sequence; later errors are often cascading failures.

3) Increase verbosity safely

• Enable debug logs only for the affected request ID (per-request debug) or only for the relevant logger/module.
• Add temporary logs around suspected decision points (branch selection, cache hits/misses, retries) using structured fields.
• Keep redaction on; never “temporarily” log tokens or passwords.

4) Verify the fix with tests

Once you believe you fixed the issue, lock it in with tests that would have caught it:

• Regression test: the minimal input that used to fail now succeeds (or fails with the correct error).
• Logging test (optional): assert that critical events are logged and that sensitive fields are redacted.
• Performance guard: for known slow paths, assert step timings are emitted or that a timeout path behaves correctly.

Example: verify redaction using Flask’s test client and caplog (pytest):

import json, logging

def test_request_logging_redacts_authorization(client, caplog):
    caplog.set_level(logging.DEBUG)
    client.get("/health", headers={"Authorization": "Bearer secret"})

    # Find any JSON log lines and ensure Authorization is redacted
    for rec in caplog.records:
        msg = rec.getMessage()
        # If using JsonFormatter, message is not JSON; structured fields are in rec.__dict__
        headers = getattr(rec, "headers", None)
        if headers and "Authorization" in headers:
            assert headers["Authorization"] == "[REDACTED]"