Django middlewares to monitor your application with Prometheus.io
npx @tessl/cli install tessl/pypi-django-prometheus@2.4.0Django middlewares to monitor your application with Prometheus.io. Provides comprehensive monitoring capabilities for Django applications through middleware components that automatically collect and export metrics for HTTP requests/responses, database operations, cache access, and model operations.
pip install django-prometheusimport django_prometheusCommon imports for middleware:
from django_prometheus.middleware import PrometheusBeforeMiddleware, PrometheusAfterMiddlewareFor model monitoring:
from django_prometheus.models import ExportModelOperationsMixinFor exporting metrics:
from django_prometheus.exports import ExportToDjangoView, SetupPrometheusEndpointOnPort, SetupPrometheusEndpointOnPortRange, SetupPrometheusExportsFromConfigFor testing utilities:
from django_prometheus.testutils import assert_metric_equal, save_registry, assert_metric_diff, get_metricFor migration monitoring:
from django_prometheus.migrations import ExportMigrations, ExportMigrationsForDatabaseFor utility classes:
from django_prometheus.utils import PowersOf, Time, TimeSinceFor app configuration:
from django_prometheus.apps import DjangoPrometheusConfigAdd to Django settings:
# settings.py
INSTALLED_APPS = [
'django_prometheus',
# ... other apps
]
MIDDLEWARE = [
'django_prometheus.middleware.PrometheusBeforeMiddleware',
# ... other middleware
'django_prometheus.middleware.PrometheusAfterMiddleware',
]
# Optional: Export metrics on a separate port
PROMETHEUS_METRICS_EXPORT_PORT = 8001
PROMETHEUS_METRICS_EXPORT_PORT_RANGE = range(8001, 8005)
PROMETHEUS_METRICS_EXPORT_ADDRESS = ""
# Optional: Custom metric namespace (prefixes all metrics)
PROMETHEUS_METRIC_NAMESPACE = "myproject"
# Optional: Custom histogram latency buckets
PROMETHEUS_LATENCY_BUCKETS = (0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0, 25.0, 50.0, 75.0, float("inf"))
# Optional: Disable migration monitoring
PROMETHEUS_EXPORT_MIGRATIONS = FalseAdd metrics URL to your project:
# urls.py
from django.urls import path, include
urlpatterns = [
path('', include('django_prometheus.urls')),
# ... other patterns
]from django.db import models
from django_prometheus.models import ExportModelOperationsMixin
class User(ExportModelOperationsMixin('user'), models.Model):
username = models.CharField(max_length=100)
email = models.EmailField()Comprehensive monitoring of HTTP requests, responses, exceptions, and latencies through Django middleware. Tracks method, status codes, view names, response sizes, and more.
class PrometheusBeforeMiddleware:
def process_request(self, request): ...
def process_response(self, request, response): ...
class PrometheusAfterMiddleware:
def process_request(self, request): ...
def process_view(self, request, view_func, *view_args, **view_kwargs): ...
def process_template_response(self, request, response): ...
def process_response(self, request, response): ...
def process_exception(self, request, exception): ...Monitor Django model operations (insert, update, delete) by model name through mixin classes that instrument Django ORM operations.
def ExportModelOperationsMixin(model_name: str):
"""
Returns a mixin class for models to export counters for lifecycle operations.
Parameters:
- model_name: str, name of the model for metric labeling
Returns:
Mixin class with instrumented _do_insert, _do_update, delete methods
"""Automatic monitoring of database operations through instrumented Django database backends. Tracks connection counts, query execution times, errors, and bulk operations.
# Database backends with monitoring
django_prometheus.db.backends.postgresql.base.DatabaseWrapper
django_prometheus.db.backends.mysql.base.DatabaseWrapper
django_prometheus.db.backends.sqlite3.base.DatabaseWrapper
django_prometheus.db.backends.spatialite.base.DatabaseWrapper
django_prometheus.db.backends.postgis.base.DatabaseWrapperMonitor Django cache operations (get/hit/miss/fail) through instrumented cache backends for various cache systems.
# Cache backends with monitoring
django_prometheus.cache.backends.filebased.FileBasedCache
django_prometheus.cache.backends.locmem.LocMemCache
django_prometheus.cache.backends.memcached.PyMemcacheCache
django_prometheus.cache.backends.redis.RedisCacheExport Prometheus metrics through Django views or standalone HTTP servers, with support for multiprocess environments.
def ExportToDjangoView(request):
"""
Exports /metrics as a Django view.
Parameters:
- request: Django HttpRequest object
Returns:
HttpResponse with Prometheus metrics data
"""
def SetupPrometheusEndpointOnPort(port: int, addr: str = ""):
"""
Exports Prometheus metrics on HTTPServer running in its own thread.
Parameters:
- port: int, port number to serve metrics on
- addr: str, address to bind to (default: all interfaces)
"""
def SetupPrometheusEndpointOnPortRange(port_range, addr: str = ""):
"""
Like SetupPrometheusEndpointOnPort, but tries several ports.
Parameters:
- port_range: iterable of ports to try
- addr: str, address to bind to (default: all interfaces)
Returns:
int: chosen port, or None if no port available
"""
def SetupPrometheusExportsFromConfig():
"""
Exports metrics based on Django settings configuration.
Reads PROMETHEUS_METRICS_EXPORT_* settings and starts appropriate server.
"""Track applied and unapplied Django migrations across database connections to monitor deployment and migration status.
def ExportMigrations():
"""
Exports counts of unapplied migrations.
Called during app startup via AppConfig.ready().
"""
def ExportMigrationsForDatabase(alias: str, executor):
"""
Exports migration counts for specific database.
Parameters:
- alias: str, database connection alias
- executor: Django MigrationExecutor instance
"""Helper functions for time measurement and histogram bucket generation used internally by monitoring components.
def Time():
"""
Returns high-resolution timestamp representation for latency measurement.
Returns:
Opaque time object (should only be used with TimeSince)
"""
def TimeSince(t):
"""
Computes time elapsed since a timestamp from Time().
Parameters:
- t: Time object returned by Time()
Returns:
float: Elapsed time in fractional seconds
"""
def PowersOf(logbase: int, count: int, lower: int = 0, include_zero: bool = True):
"""
Generates histogram buckets as powers of a base number.
Parameters:
- logbase: int, base for power calculation
- count: int, number of powers to generate
- lower: int, starting power (default: 0)
- include_zero: bool, whether to include 0 as first bucket
Returns:
list: Bucket boundaries for histogram metrics
"""Django app configuration that automatically initializes metrics exports and migration monitoring on startup.
class DjangoPrometheusConfig(AppConfig):
"""
Django app configuration for django-prometheus.
Handles automatic initialization of metrics exports and migration monitoring.
"""
name = "django_prometheus"
verbose_name = "Django-Prometheus"
def ready(self):
"""
Initializes Prometheus exports and migration monitoring on Django startup.
Called automatically when Django loads the app, including for management commands.
"""Comprehensive test utilities for asserting Prometheus metric values and changes in test suites.
def assert_metric_equal(expected_value, metric_name: str, registry=REGISTRY, **labels):
"""
Asserts that metric equals expected value.
Parameters:
- expected_value: expected metric value
- metric_name: str, name of the metric
- registry: Prometheus registry (default: REGISTRY)
- **labels: metric label filters
"""
def save_registry(registry=REGISTRY):
"""
Freezes current state of metrics registry for comparison.
Parameters:
- registry: Prometheus registry to freeze
Returns:
Frozen registry state for later comparison
"""
def assert_metric_diff(frozen_registry, expected_diff, metric_name: str, registry=REGISTRY, **labels):
"""
Asserts that metric changed by expected_diff between frozen registry and now.
Parameters:
- frozen_registry: Previously frozen registry state
- expected_diff: Expected change in metric value
- metric_name: str, name of the metric
- registry: Prometheus registry (default: REGISTRY)
- **labels: metric label filters
"""
def get_metric(metric_name: str, registry=REGISTRY, **labels):
"""
Gets a single metric value.
Parameters:
- metric_name: str, name of the metric
- registry: Prometheus registry (default: REGISTRY)
- **labels: metric label filters
Returns:
Metric value or None if not found
"""from prometheus_client import Counter, Histogram, Gauge
from django.http import HttpRequest, HttpResponse
from django.utils.deprecation import MiddlewareMixin
from django.db.migrations.executor import MigrationExecutor