move stats views to django_rq.stats_views module

Author: terencehonles

django_rq/stats_views.py

@@ -0,0 +1,56 @@
+from django.contrib import admin
+from django.contrib.admin.views.decorators import staff_member_required
+from django.http import Http404, HttpResponse, JsonResponse
+from django.shortcuts import render
+from django.views.decorators.cache import never_cache
+
+from .settings import API_TOKEN
+from .utils import get_scheduler_statistics, get_statistics
+
+try:
+    import prometheus_client
+
+    from .metrics_collector import RQCollector
+except ImportError:
+    prometheus_client = RQCollector = None  # type: ignore[assignment, misc]
+
+registry = None
+
+
+@never_cache
+@staff_member_required
+def prometheus_metrics(request):
+    global registry
+
+    if not RQCollector:  # type: ignore[truthy-function]
+        raise Http404
+
+    if not registry:
+        registry = prometheus_client.CollectorRegistry(auto_describe=True)
+        registry.register(RQCollector())
+
+    encoder, content_type = prometheus_client.exposition.choose_encoder(request.META.get('HTTP_ACCEPT', ''))
+    if 'name[]' in request.GET:
+        registry = registry.restricted_registry(request.GET.getlist('name[]'))
+
+    return HttpResponse(encoder(registry), headers={'Content-Type': content_type})
+
+
+@never_cache
+@staff_member_required
+def stats(request):
+    context_data = {
+        **admin.site.each_context(request),
+        **get_statistics(run_maintenance_tasks=True),
+        **get_scheduler_statistics(),
+    }
+    return render(request, 'django_rq/stats.html', context_data)
+
+
+def stats_json(request, token=None):
+    if request.user.is_staff or (token and token == API_TOKEN):
+        return JsonResponse(get_statistics())
+
+    return JsonResponse(
+        {"error": True, "description": "Please configure API_TOKEN in settings.py before accessing this view."}
+    )

django_rq/urls.py

@@ -1,15 +1,15 @@
 from django.urls import re_path
 
-from . import views
+from . import stats_views, views
 from .metrics_collector import RQCollector
 
 metrics_view = [
-    re_path(r'^metrics/?$', views.prometheus_metrics, name='rq_metrics'),
+    re_path(r'^metrics/?$', stats_views.prometheus_metrics, name='rq_metrics'),
 ] if RQCollector else []  # type: ignore[truthy-function]
 
 urlpatterns = [
-    re_path(r'^$', views.stats, name='rq_home'),
-    re_path(r'^stats.json/(?P<token>[\w]+)?/?$', views.stats_json, name='rq_home_json'),
+    re_path(r'^$', stats_views.stats, name='rq_home'),
+    re_path(r'^stats.json/(?P<token>[\w]+)?/?$', stats_views.stats_json, name='rq_home_json'),
     *metrics_view,
     re_path(r'^queues/(?P<queue_index>[\d]+)/$', views.jobs, name='rq_jobs'),
     re_path(r'^workers/(?P<queue_index>[\d]+)/$', views.workers, name='rq_workers'),

django_rq/views.py

@@ -28,54 +28,6 @@
 from .settings import API_TOKEN, QUEUES_MAP
 from .utils import get_executions, get_jobs, get_scheduler_statistics, get_statistics, stop_jobs
 
-try:
-    import prometheus_client
-
-    from .metrics_collector import RQCollector
-except ImportError:
-    prometheus_client = RQCollector = None  # type: ignore[assignment, misc]
-
-registry = None
-
-
-@never_cache
-@staff_member_required
-def stats(request):
-    context_data = {
-        **admin.site.each_context(request),
-        **get_statistics(run_maintenance_tasks=True),
-        **get_scheduler_statistics(),
-    }
-    return render(request, 'django_rq/stats.html', context_data)
-
-
-def stats_json(request, token=None):
-    if request.user.is_staff or (token and token == API_TOKEN):
-        return JsonResponse(get_statistics())
-
-    return JsonResponse(
-        {"error": True, "description": "Please configure API_TOKEN in settings.py before accessing this view."}
-    )
-
-
-@never_cache
-@staff_member_required
-def prometheus_metrics(request):
-    global registry
-
-    if not RQCollector:  # type: ignore[truthy-function]
-        raise Http404
-
-    if not registry:
-        registry = prometheus_client.CollectorRegistry(auto_describe=True)
-        registry.register(RQCollector())
-
-    encoder, content_type = prometheus_client.exposition.choose_encoder(request.META.get('HTTP_ACCEPT', ''))
-    if 'name[]' in request.GET:
-        registry = registry.restricted_registry(request.GET.getlist('name[]'))
-
-    return HttpResponse(encoder(registry), headers={'Content-Type': content_type})
-
 
 @never_cache
 @staff_member_required