Structured Logging in Flask

Flask's built-in logging uses Python's standard logging module, but it outputs plain text by default. This tutorial adds structured JSON logging with request context, error tracking, and log shipping to LogFlow.

Prerequisites

• Python 3.9+ with Flask installed
• A LogFlow account (free tier works)
• Your API key from Settings → API Key

Step 1 — Install Dependencies

pip install flask requests

Step 2 — Create the Logger

Create logger.py in your project:

import logging
import json
import os
import contextvars

# Context variables for per-request data
request_id_var = contextvars.ContextVar("request_id", default="-")

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record, datefmt="%Y-%m-%dT%H:%M:%S.000Z"),
            "level": record.levelname.lower(),
            "message": record.getMessage(),
            "service": os.getenv("SERVICE_NAME", "flask-app"),
            "request_id": request_id_var.get(),
        }
        # Copy extra fields
        for key in record.__dict__:
            if key not in logging.LogRecord("", 0, "", 0, "", (), None).__dict__ and key != "request_id":
                log_entry[key] = record.__dict__[key]
        return json.dumps(log_entry)

def setup_logging():
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())

    root = logging.getLogger()
    root.handlers = [handler]
    root.setLevel(logging.INFO)

    # Quiet noisy libraries
    logging.getLogger("werkzeug").setLevel(logging.WARNING)
    logging.getLogger("urllib3").setLevel(logging.WARNING)

Step 3 — Add Request Logging Middleware

Update your Flask app (app.py):

import uuid
import time
import logging
from flask import Flask, request, g
from logger import setup_logging, request_id_var

setup_logging()
logger = logging.getLogger(__name__)

app = Flask(__name__)

@app.before_request
def before_request():
    g.start_time = time.time()
    g.request_id = request.headers.get("X-Request-ID", str(uuid.uuid4())[:8])
    request_id_var.set(g.request_id)

@app.after_request
def after_request(response):
    duration_ms = round((time.time() - g.start_time) * 1000)

    # Skip health check noise
    if request.path == "/health":
        return response

    logger.info("request.handled", extra={
        "method": request.method,
        "path": request.path,
        "status": response.status_code,
        "duration_ms": duration_ms,
        "ip": request.remote_addr,
        "user_agent": request.headers.get("User-Agent", ""),
    })

    # Add request ID to response headers for debugging
    response.headers["X-Request-ID"] = g.request_id
    return response

@app.errorhandler(Exception)
def handle_exception(e):
    logger.error("request.error", extra={
        "error": str(e),
        "error_type": type(e).__name__,
        "path": request.path,
        "method": request.method,
    }, exc_info=True)
    return {"error": "Internal server error"}, 500

Step 4 — Ship Logs to LogFlow

Add log shipping with the HTTP API. Create logflow_handler.py:

import logging
import json
import threading
import queue
import atexit
import os
import requests as http_requests

LOGFLOW_URL = "https://api.getlogflow.com/v1/logs"
LOGFLOW_API_KEY = os.getenv("LOGFLOW_API_KEY", "")

_queue: queue.Queue = queue.Queue(maxsize=10_000)
_stop = threading.Event()

class LogFlowHandler(logging.Handler):
    def emit(self, record):
        try:
            _queue.put_nowait(self.format(record))
        except queue.Full:
            pass

def _worker():
    session = http_requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {LOGFLOW_API_KEY}",
        "Content-Type": "application/json",
    })

    while not _stop.is_set():
        batch = []
        try:
            batch.append(json.loads(_queue.get(timeout=2)))
        except queue.Empty:
            continue

        while len(batch) < 50:
            try:
                batch.append(json.loads(_queue.get_nowait()))
            except queue.Empty:
                break

        try:
            session.post(LOGFLOW_URL, json=batch, timeout=5)
        except Exception:
            pass

_thread = threading.Thread(target=_worker, daemon=True)
_thread.start()
atexit.register(lambda: (_stop.set(), _thread.join(3)))

Add the handler in setup_logging():

from logflow_handler import LogFlowHandler
from logger import JSONFormatter

def setup_logging():
    formatter = JSONFormatter()

    console = logging.StreamHandler()
    console.setFormatter(formatter)

    logflow = LogFlowHandler()
    logflow.setFormatter(formatter)

    root = logging.getLogger()
    root.handlers = [console, logflow]
    root.setLevel(logging.INFO)

Set your API key in the environment:

export LOGFLOW_API_KEY=lf_your_api_key_here

Step 5 — Add User Context

If your app has authentication, add user info to logs:

from flask_login import current_user

@app.before_request
def add_user_context():
    if current_user and current_user.is_authenticated:
        g.user_id = current_user.id
    else:
        g.user_id = None

Then include g.user_id in your log extra dict. This lets you search all logs for a specific user in LogFlow's Logs Explorer: user_id:42.

Step 6 — Verify in LogFlow

Run your Flask app and make some requests:

flask run
curl http://localhost:5000/
curl http://localhost:5000/nonexistent

Open LogFlow and search:

service:flask-app

You should see structured request logs with method, path, status, and duration.

Next Steps

• Set up Slack alerts — get notified when error rates spike
• Python logging best practices — deeper patterns
• Structured logging guide — why JSON matters
• Django logging tutorial — if you also use Django
• FastAPI logging tutorial — async alternative