Django's built-in logging is powerful but defaults to plain text output. By adding a custom handler that ships structured JSON logs to LogFlow, you get full-text search, alerting, and anomaly detection across your Django application.

Prerequisites:

Python 3.10+, Django 4.x or 5.x
A LogFlow account (free tier works)
Your API key from Settings → API Key

Step 1 — Install Dependencies

pip install requests python-dotenv

We'll use the LogFlow HTTP API directly with requests. No extra library needed.

Step 2 — Create a LogFlow Handler

Create your_project/logflow_handler.py:

import json
import logging
import threading
from datetime import datetime, timezone
from typing import Any

import requests


class LogFlowHandler(logging.Handler):
    """
    A logging handler that ships structured log records to LogFlow
    via the batch ingestion API. Non-blocking — logs are sent in a
    background thread so your application doesn't stall on network issues.
    """

    API_URL = "https://api.getlogflow.com/v1/logs/batch"

    def __init__(self, api_key: str, service: str, environment: str = "production"):
        super().__init__()
        self.api_key = api_key
        self.service = service
        self.environment = environment
        self._session = requests.Session()
        self._session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })

    def emit(self, record: logging.LogRecord) -> None:
        # Run in background thread — never block the main request
        thread = threading.Thread(target=self._send, args=(record,), daemon=True)
        thread.start()

    def _build_payload(self, record: logging.LogRecord) -> dict[str, Any]:
        level_map = {
            logging.DEBUG: "debug",
            logging.INFO: "info",
            logging.WARNING: "warn",
            logging.ERROR: "error",
            logging.CRITICAL: "fatal",
        }

        attributes: dict[str, str] = {
            "logger": record.name,
            "module": record.module,
            "function": record.funcName,
            "line": str(record.lineno),
        }

        # Attach exception info if present
        if record.exc_info:
            import traceback
            attributes["exception"] = "".join(
                traceback.format_exception(*record.exc_info)
            )

        # Attach any extra fields passed via extra={}
        for key, value in record.__dict__.items():
            if key.startswith("_") or key in (
                "name", "msg", "args", "levelname", "levelno",
                "pathname", "filename", "module", "exc_info",
                "exc_text", "stack_info", "lineno", "funcName",
                "created", "msecs", "relativeCreated", "thread",
                "threadName", "processName", "process", "message",
            ):
                continue
            attributes[key] = str(value)

        return {
            "level": level_map.get(record.levelno, "info"),
            "message": record.getMessage(),
            "service": self.service,
            "timestamp": datetime.fromtimestamp(
                record.created, tz=timezone.utc
            ).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
            "attributes": attributes,
        }

    def _send(self, record: logging.LogRecord) -> None:
        try:
            payload = self._build_payload(record)
            self._session.post(
                self.API_URL,
                data=json.dumps({"logs": [payload]}),
                timeout=5,
            )
        except Exception:
            # Never let logging errors crash the application
            pass

Step 3 — Configure Django Logging

In settings.py, add LogFlow to Django's LOGGING dict:

import os
from dotenv import load_dotenv

load_dotenv()

LOGFLOW_API_KEY = os.environ.get("LOGFLOW_API_KEY", "")
ENVIRONMENT = os.environ.get("DJANGO_ENV", "production")

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,

    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "simple",
        },
        "logflow": {
            "()": "your_project.logflow_handler.LogFlowHandler",
            "api_key": LOGFLOW_API_KEY,
            "service": "django-api",
            "environment": ENVIRONMENT,
            "level": "INFO",    # Don't ship DEBUG to LogFlow
        },
    },

    "formatters": {
        "simple": {
            "format": "{levelname} {name} {message}",
            "style": "{",
        },
    },

    "loggers": {
        # Your application logs
        "your_app": {
            "handlers": ["console", "logflow"],
            "level": "DEBUG",   # DEBUG locally, INFO filters at handler level
            "propagate": False,
        },
        # Django request logs (access log)
        "django.request": {
            "handlers": ["logflow"],
            "level": "WARNING",  # Only 4xx/5xx — too noisy otherwise
            "propagate": True,
        },
        # Database query logs — be careful, this can be very verbose
        "django.db.backends": {
            "handlers": [],      # Enable only when debugging query issues
            "level": "DEBUG",
            "propagate": False,
        },
    },

    "root": {
        "handlers": ["console"],
        "level": "WARNING",
    },
}

Add your API key to .env:

LOGFLOW_API_KEY=lf_your_api_key_here
DJANGO_ENV=production

Step 4 — Log in Your Views

Use Python's standard logging module — no LogFlow-specific API in your application code:

import logging
from django.http import JsonResponse
from django.views import View

logger = logging.getLogger("your_app.views")


class OrderView(View):
    def get(self, request, order_id: str) -> JsonResponse:
        logger.info(
            "Fetching order",
            extra={
                "order_id": order_id,
                "user_id": str(request.user.id),
                "request_id": request.headers.get("X-Request-Id", ""),
            },
        )

        try:
            order = Order.objects.get(id=order_id, user=request.user)
            return JsonResponse(order.to_dict())

        except Order.DoesNotExist:
            logger.warning(
                "Order not found",
                extra={"order_id": order_id, "user_id": str(request.user.id)},
            )
            return JsonResponse({"error": "Not found"}, status=404)

        except Exception as e:
            logger.error(
                "Unexpected error fetching order",
                exc_info=True,
                extra={
                    "order_id": order_id,
                    "user_id": str(request.user.id),
                },
            )
            return JsonResponse({"error": "Internal server error"}, status=500)

The exc_info=True flag attaches the full exception traceback to the log — it will appear in the exception attribute in LogFlow.

Step 5 — Add Request Middleware

For request-level logging with duration and status code, add a middleware:

# your_project/middleware.py
import logging
import time
import uuid

logger = logging.getLogger("your_app.requests")


class RequestLoggingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        request_id = str(uuid.uuid4())[:8]
        request.request_id = request_id
        start = time.time()

        response = self.get_response(request)

        duration_ms = round((time.time() - start) * 1000)
        level = logging.ERROR if response.status_code >= 500 else logging.INFO

        logger.log(
            level,
            f"{request.method} {request.path}",
            extra={
                "method": request.method,
                "path": request.path,
                "status_code": str(response.status_code),
                "duration_ms": str(duration_ms),
                "request_id": request_id,
                "user_id": str(request.user.id) if request.user.is_authenticated else "",
                "ip": request.META.get("REMOTE_ADDR", ""),
            },
        )

        return response

MIDDLEWARE = [
    "your_project.middleware.RequestLoggingMiddleware",
    # ... existing middleware ...
]

Step 6 — Celery Task Logging (Optional)

If you use Celery for background tasks, add LogFlow to Celery's logging config:

# celery.py
from celery import Celery
from celery.signals import setup_logging

app = Celery("your_project")

@setup_logging.connect
def configure_logging(**kwargs):
    from django.conf import settings
    import logging.config
    logging.config.dictConfig(settings.LOGGING)

Then log in your tasks using standard logging:

import logging
from celery import shared_task

logger = logging.getLogger("your_app.tasks")


@shared_task
def send_weekly_report(user_id: int) -> None:
    logger.info(
        "Starting weekly report",
        extra={"user_id": str(user_id), "task": "send_weekly_report"},
    )
    try:
        # ... task logic ...
        logger.info(
            "Weekly report sent",
            extra={"user_id": str(user_id), "task": "send_weekly_report"},
        )
    except Exception:
        logger.error(
            "Failed to send weekly report",
            exc_info=True,
            extra={"user_id": str(user_id), "task": "send_weekly_report"},
        )
        raise

Searching in LogFlow

With structured attributes, you can filter precisely in the Logs Explorer:

service:django-api level:error — all Django errors
user_id:12345 — everything this user triggered
status_code:500 — all 500 errors
task:send_weekly_report level:error — failed Celery tasks

Setting Up Alerts

Common Django alerts to create in your Alerts dashboard:

Error rate above 5% — catch request handler failures
Keyword match on "OperationalError" — database connection issues
Keyword match on "IntegrityError" — data constraint violations
Log volume drop — your Django app went silent (process crashed, gunicorn died)

Read Alerting Best Practices for guidance on thresholds and cooldowns.

Structured Logging in Django with LogFlow