Add health_check django command (#217)

README.rst

@@ -266,6 +266,26 @@ and customizing the ``template_name``, ``get``, ``render_to_response`` and ``ren
         url(r'^ht/$', views.HealthCheckCustomView.as_view(), name='health_check_custom'),
     ]
 
+Django command
+--------------
+
+You can run the Django command `health_check` to perform your health checks via the command line,
+or periodically with a cron, as follow:
+
+.. code::
+
+    django-admin health_check
+
+This should yield the following output:
+
+.. code::
+
+    DatabaseHealthCheck      ... working
+    CustomHealthCheck        ... unavailable: Something went wrong!
+
+Similar to the http version, a critical error will case the command to quit with the exit code `1`.
+
+
 Other resources
 ---------------
 

health_check/management/commands/health_check.py

@@ -0,0 +1,25 @@
+import sys
+
+from django.core.management.base import BaseCommand
+
+from health_check.mixins import CheckMixin
+
+
+class Command(CheckMixin, BaseCommand):
+    help = "Run health checks and exit 0 if everything went well."
+
+    def handle(self, *args, **options):
+        # perform all checks
+        errors = self.errors
+
+        for plugin in self.plugins:
+            style_func = self.style.SUCCESS if not plugin.errors else self.style.ERROR
+            self.stdout.write(
+                "{:<24} ... {}\n".format(
+                    plugin.identifier(),
+                    style_func(plugin.pretty_status())
+                )
+            )
+
+        if errors:
+            sys.exit(1)

health_check/mixins.py

@@ -0,0 +1,51 @@
+import copy
+
+from concurrent.futures import ThreadPoolExecutor
+
+from health_check.conf import HEALTH_CHECK
+from health_check.exceptions import ServiceWarning
+from health_check.plugins import plugin_dir
+
+
+class CheckMixin:
+    _errors = None
+    _plugins = None
+
+    @property
+    def errors(self):
+        if not self._errors:
+            self._errors = self.run_check()
+        return self._errors
+
+    @property
+    def plugins(self):
+        if not self._plugins:
+            self._plugins = sorted((
+                plugin_class(**copy.deepcopy(options))
+                for plugin_class, options in plugin_dir._registry
+            ), key=lambda plugin: plugin.identifier())
+        return self._plugins
+
+    def run_check(self):
+        errors = []
+
+        def _run(plugin):
+            plugin.run_check()
+            try:
+                return plugin
+            finally:
+                from django.db import connection
+                connection.close()
+
+        with ThreadPoolExecutor(max_workers=len(self.plugins) or 1) as executor:
+            for plugin in executor.map(_run, self.plugins):
+                if plugin.critical_service:
+                    if not HEALTH_CHECK['WARNINGS_AS_ERRORS']:
+                        errors.extend(
+                            e for e in plugin.errors
+                            if not isinstance(e, ServiceWarning)
+                        )
+                    else:
+                        errors.extend(plugin.errors)
+
+        return errors

health_check/views.py

@@ -1,14 +1,10 @@
-import copy
 import re
-from concurrent.futures import ThreadPoolExecutor
 
 from django.http import HttpResponse, JsonResponse
 from django.views.decorators.cache import never_cache
 from django.views.generic import TemplateView
 
-from health_check.conf import HEALTH_CHECK
-from health_check.exceptions import ServiceWarning
-from health_check.plugins import plugin_dir
+from health_check.mixins import CheckMixin
 
 
 class MediaType:
@@ -81,51 +77,25 @@ def __lt__(self, other):
         return self.weight.__lt__(other.weight)
 
 
-class MainView(TemplateView):
+class MainView(CheckMixin, TemplateView):
     template_name = 'health_check/index.html'
 
     @never_cache
     def get(self, request, *args, **kwargs):
-        errors = []
-
-        plugins = sorted((
-            plugin_class(**copy.deepcopy(options))
-            for plugin_class, options in plugin_dir._registry
-        ), key=lambda plugin: plugin.identifier())
-
-        def _run(plugin):
-            plugin.run_check()
-            try:
-                return plugin
-            finally:
-                from django.db import connection
-                connection.close()
-
-        with ThreadPoolExecutor(max_workers=len(plugins) or 1) as executor:
-            for plugin in executor.map(_run, plugins):
-                if plugin.critical_service:
-                    if not HEALTH_CHECK['WARNINGS_AS_ERRORS']:
-                        errors.extend(
-                            e for e in plugin.errors
-                            if not isinstance(e, ServiceWarning)
-                        )
-                    else:
-                        errors.extend(plugin.errors)
-
-        status_code = 500 if errors else 200
+        status_code = 500 if self.errors else 200
 
         format_override = request.GET.get('format')
 
         if format_override == 'json':
-            return self.render_to_response_json(plugins, status_code)
+            return self.render_to_response_json(self.plugins, status_code)
 
         accept_header = request.META.get('HTTP_ACCEPT', '*/*')
-        context = {'plugins': plugins, 'status_code': status_code}
         for media in MediaType.parse_header(accept_header):
             if media.mime_type in ('text/html', 'application/xhtml+xml', 'text/*', '*/*'):
+                context = {'plugins': self.plugins, 'status_code': status_code}
                 return self.render_to_response(context, status=status_code)
             elif media.mime_type in ('application/json', 'application/*'):
-                return self.render_to_response_json(plugins, status_code)
+                return self.render_to_response_json(self.plugins, status_code)
         return HttpResponse(
             'Not Acceptable: Supported content types: text/html, application/json',
             status=406,

tests/test_commands.py

@@ -0,0 +1,38 @@
+from io import StringIO
+
+import pytest
+
+from health_check.backends import BaseHealthCheckBackend
+from health_check.plugins import plugin_dir
+
+from django.core.management import call_command
+
+
+class FailPlugin(BaseHealthCheckBackend):
+    def check_status(self):
+        self.add_error('Oops')
+
+
+class OkPlugin(BaseHealthCheckBackend):
+    def check_status(self):
+        pass
+
+
+class TestCommand:
+    @pytest.yield_fixture(autouse=True)
+    def setup(self):
+        plugin_dir.reset()
+        plugin_dir.register(FailPlugin)
+        plugin_dir.register(OkPlugin)
+        yield
+        plugin_dir.reset()
+
+    def test_command(self):
+        stdout = StringIO()
+        with pytest.raises(SystemExit):
+            call_command("health_check", stdout=stdout)
+        stdout.seek(0)
+        assert stdout.read() == (
+            "FailPlugin               ... unknown error: Oops\n"
+            "OkPlugin                 ... working\n"
+        )

tests/test_mixins.py

@@ -0,0 +1,38 @@
+import pytest
+
+from health_check.backends import BaseHealthCheckBackend
+from health_check.mixins import CheckMixin
+from health_check.plugins import plugin_dir
+
+
+class FailPlugin(BaseHealthCheckBackend):
+    def check_status(self):
+        self.add_error('Oops')
+
+
+class OkPlugin(BaseHealthCheckBackend):
+    def check_status(self):
+        pass
+
+
+class Checker(CheckMixin):
+    pass
+
+
+class TestCheckMixin:
+    @pytest.yield_fixture(autouse=True)
+    def setup(self):
+        plugin_dir.reset()
+        plugin_dir.register(FailPlugin)
+        plugin_dir.register(OkPlugin)
+        yield
+        plugin_dir.reset()
+
+    def test_plugins(self):
+        assert len(Checker().plugins) == 2
+
+    def test_errors(self):
+        assert len(Checker().errors) == 1
+
+    def test_run_check(self):
+        assert len(Checker().run_check()) == 1