This tutorial sets up structured JSON logging in Python and ships logs to LogFlow via the HTTP API. Works with any Python framework — Flask, Django, FastAPI, or plain scripts.

Prerequisites

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

Step 1 — Install the Requests Library

LogFlow doesn't require a special Python SDK. Use the standard requests library (or httpx, urllib3, or even urllib):

pip install requests

Step 2 — Create a Logger Module

Create logflow.py:

import logging
import json
import threading
import queue
import atexit
import requests

LOGFLOW_API_KEY = "lf_your_api_key"  # Use env vars in production
LOGFLOW_URL = "https://api.getlogflow.com/v1/logs"
SERVICE_NAME = "my-python-app"

# Background queue for non-blocking log shipping
_log_queue: queue.Queue = queue.Queue(maxsize=10_000)
_stop_event = threading.Event()


class JSONFormatter(logging.Formatter):
    """Format log records as JSON with extra fields."""

    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": self.formatTime(record, datefmt="%Y-%m-%dT%H:%M:%S.000Z"),
            "level": record.levelname.lower(),
            "message": record.getMessage(),
            "service": SERVICE_NAME,
            "module": record.module,
            "function": record.funcName,
        }
        # Include any extra fields passed via logger.info("msg", extra={...})
        for key in ("user_id", "request_id", "order_id", "duration_ms", "path", "status", "error", "trace_id"):
            if hasattr(record, key):
                log_entry[key] = getattr(record, key)
        return json.dumps(log_entry)


class LogFlowHandler(logging.Handler):
    """Sends log records to LogFlow API in a background thread."""

    def emit(self, record: logging.LogRecord):
        try:
            msg = self.format(record)
            _log_queue.put_nowait(msg)
        except queue.Full:
            pass  # Drop logs rather than blocking the app


def _flush_worker():
    """Background thread that batches and sends logs."""
    session = requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {LOGFLOW_API_KEY}",
        "Content-Type": "application/json",
    })

    while not _stop_event.is_set():
        batch = []
        try:
            # Wait up to 2 seconds for the first item
            item = _log_queue.get(timeout=2)
            batch.append(json.loads(item))
        except queue.Empty:
            continue

        # Drain up to 99 more items
        while len(batch) < 100:
            try:
                item = _log_queue.get_nowait()
                batch.append(json.loads(item))
            except queue.Empty:
                break

        if batch:
            try:
                session.post(LOGFLOW_URL, json=batch, timeout=5)
            except requests.RequestException:
                pass  # Log shipping should never crash the app

    # Final flush on shutdown
    batch = []
    while not _log_queue.empty():
        try:
            batch.append(json.loads(_log_queue.get_nowait()))
        except queue.Empty:
            break
    if batch:
        try:
            session.post(LOGFLOW_URL, json=batch, timeout=5)
        except requests.RequestException:
            pass


# Start background thread
_worker = threading.Thread(target=_flush_worker, daemon=True)
_worker.start()


def _shutdown():
    _stop_event.set()
    _worker.join(timeout=5)


atexit.register(_shutdown)


def get_logger(name: str) -> logging.Logger:
    """Get a configured logger that outputs JSON and ships to LogFlow."""
    logger = logging.getLogger(name)
    if not logger.handlers:
        # Console handler (JSON to stdout)
        console = logging.StreamHandler()
        console.setFormatter(JSONFormatter())
        logger.addHandler(console)

        # LogFlow handler (background shipping)
        logflow = LogFlowHandler()
        logflow.setFormatter(JSONFormatter())
        logger.addHandler(logflow)

        logger.setLevel(logging.INFO)
    return logger

Step 3 — Use the Logger

from logflow import get_logger

logger = get_logger(__name__)

# Basic logging
logger.info("server started", extra={"port": 8000})

# With context
logger.info("order.created", extra={
    "order_id": "ORD-1234",
    "user_id": 42,
    "amount": 99.00,
})

# Errors with exception info
try:
    process_payment(order)
except Exception as e:
    logger.error("payment.failed", extra={
        "order_id": order.id,
        "error": str(e),
    }, exc_info=True)

Step 4 — Add Request Context (Web Apps)

For Flask, Django, or FastAPI, add a middleware that attaches request ID and user info to every log:

import contextvars
import uuid

request_id_var = contextvars.ContextVar("request_id", default="-")
user_id_var = contextvars.ContextVar("user_id", default=None)


# Flask middleware example
from flask import Flask, request, g

app = Flask(__name__)

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

@app.after_request
def log_request(response):
    logger.info("request.handled", extra={
        "request_id": g.request_id,
        "path": request.path,
        "method": request.method,
        "status": response.status_code,
    })
    return response

For framework-specific details, see:

Step 5 — Configure Log Levels per Module

Keep noisy libraries quiet:

import logging

# Quiet third-party noise
logging.getLogger("urllib3").setLevel(logging.WARNING)
logging.getLogger("requests").setLevel(logging.WARNING)
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)

# Verbose for your code when debugging
logging.getLogger("myapp.payments").setLevel(logging.DEBUG)

Step 6 — Verify in LogFlow

Open LogFlow Logs Explorer and search:

service:my-python-app

You should see your structured logs with all extra fields searchable and filterable.

Next Steps

Set up alerts — get notified in Slack when errors spike
Python logging best practices — deeper guide
Structured logging guide — why JSON matters
PII masking — automatically redact sensitive data

Logging in Python with LogFlow