from django.conf import settings
from moj_irat.healthchecks import HealthcheckResponse, UrlHealthcheck, registry


def get_stats():
    from celery import Celery

    app = Celery("laalaa")
    app.config_from_object("django.conf:settings")
    return app.control.inspect().stats()


class CeleryWorkersHealthcheck(object):
    def __init__(self, name):
        self.name = name

    def __call__(self, *args, **kwargs):

        try:
            stats = get_stats()

            if not stats:
                return self.error_response("No running workers were found.")

            workers = stats.values()
            if not workers:
                return self.error_response("No workers running.")

        except IOError as e:
            msg = str(e)
            msg += ". Check that the message broker is running."
            return self.error_response(msg)

        except ImportError as e:
            return self.error_response(str(e))

        return self.success_response()

    def error_response(self, error):
        return HealthcheckResponse(self.name, False, error=error)

    def success_response(self):
        return HealthcheckResponse(self.name, True)


registry.register_healthcheck(
    UrlHealthcheck(name="postcodes.io", url="{}/postcodes/SW1A1AA".format(settings.POSTCODES_IO_URL))
)

registry.register_healthcheck(CeleryWorkersHealthcheck(name="workers"))

from django.conf import settings
from moj_irat.healthchecks import HealthcheckResponse, UrlHealthcheck, registry


def get_stats():
    from celery import Celery

    app = Celery("laalaa")
    app.config_from_object("django.conf:settings", namespace="CELERY")
    return app.control.inspect().stats()


class CeleryWorkersHealthcheck(object):
    def __init__(self, name):
        self.name = name

    def __call__(self, *args, **kwargs):

        try:
            stats = get_stats()

            if not stats:
                return self.error_response("No running workers were found.")

            workers = stats.values()
            if not workers:
                return self.error_response("No workers running.")

        except IOError as e:
            msg = str(e)
            msg += ". Check that the message broker is running."
            return self.error_response(msg)

        except ImportError as e:
            return self.error_response(str(e))

        return self.success_response()

    def error_response(self, error):
        return HealthcheckResponse(self.name, False, error=error)

    def success_response(self):
        return HealthcheckResponse(self.name, True)


registry.register_healthcheck(
    UrlHealthcheck(name="postcodes.io", url="{}/postcodes/SW1A1AA".format(settings.POSTCODES_IO_URL))
)

registry.register_healthcheck(CeleryWorkersHealthcheck(name="workers"))

Load namespaced Celery configuration in healthcheck
Load namespaced Celery configuration in healthcheck

In bed52d9c60b00be751a6a9a6fc78b333fc5bccf6, I had to change the
configuration to be compatible with Django. I completely missed this
part.

Unfortunately, the test for this module starts with mocking the
`get_stats()` function, where this code exists, so I am at loss at how
to test this.

For now, I think we can expose `/healthcheck.json` on the staging
environment which would have helped detect this problem. In the
long-term, we will use a separate healthcheck for the worker processes.

There is still the question whether the "webapp" can connect to the
"broker" which I feel could be tackled by this app's healthcheck.

Thoughts for later, I suppose.

  from django.conf import settings
  from moj_irat.healthchecks import HealthcheckResponse, UrlHealthcheck, registry
  
  
  def get_stats():
      from celery import Celery
  
      app = Celery("laalaa")
-     app.config_from_object("django.conf:settings")
+     app.config_from_object("django.conf:settings", namespace="CELERY")
      return app.control.inspect().stats()
  
  
  class CeleryWorkersHealthcheck(object):
      def __init__(self, name):
          self.name = name
  
      def __call__(self, *args, **kwargs):
  
          try:
              stats = get_stats()
  
              if not stats:
                  return self.error_response("No running workers were found.")
  
              workers = stats.values()
              if not workers:
                  return self.error_response("No workers running.")
  
          except IOError as e:
              msg = str(e)
              msg += ". Check that the message broker is running."
              return self.error_response(msg)
  
          except ImportError as e:
              return self.error_response(str(e))
  
          return self.success_response()
  
      def error_response(self, error):
          return HealthcheckResponse(self.name, False, error=error)
  
      def success_response(self):
          return HealthcheckResponse(self.name, True)
  
  
  registry.register_healthcheck(
      UrlHealthcheck(name="postcodes.io", url="{}/postcodes/SW1A1AA".format(settings.POSTCODES_IO_URL))
  )
  
  registry.register_healthcheck(CeleryWorkersHealthcheck(name="workers"))
  
Load namespaced Celery configuration in healthcheck
## Code Before:
from django.conf import settings
from moj_irat.healthchecks import HealthcheckResponse, UrlHealthcheck, registry


def get_stats():
    from celery import Celery

    app = Celery("laalaa")
    app.config_from_object("django.conf:settings")
    return app.control.inspect().stats()


class CeleryWorkersHealthcheck(object):
    def __init__(self, name):
        self.name = name

    def __call__(self, *args, **kwargs):

        try:
            stats = get_stats()

            if not stats:
                return self.error_response("No running workers were found.")

            workers = stats.values()
            if not workers:
                return self.error_response("No workers running.")

        except IOError as e:
            msg = str(e)
            msg += ". Check that the message broker is running."
            return self.error_response(msg)

        except ImportError as e:
            return self.error_response(str(e))

        return self.success_response()

    def error_response(self, error):
        return HealthcheckResponse(self.name, False, error=error)

    def success_response(self):
        return HealthcheckResponse(self.name, True)


registry.register_healthcheck(
    UrlHealthcheck(name="postcodes.io", url="{}/postcodes/SW1A1AA".format(settings.POSTCODES_IO_URL))
)

registry.register_healthcheck(CeleryWorkersHealthcheck(name="workers"))

## Instruction:
Load namespaced Celery configuration in healthcheck
## Code After:
from django.conf import settings
from moj_irat.healthchecks import HealthcheckResponse, UrlHealthcheck, registry


def get_stats():
    from celery import Celery

    app = Celery("laalaa")
    app.config_from_object("django.conf:settings", namespace="CELERY")
    return app.control.inspect().stats()


class CeleryWorkersHealthcheck(object):
    def __init__(self, name):
        self.name = name

    def __call__(self, *args, **kwargs):

        try:
            stats = get_stats()

            if not stats:
                return self.error_response("No running workers were found.")

            workers = stats.values()
            if not workers:
                return self.error_response("No workers running.")

        except IOError as e:
            msg = str(e)
            msg += ". Check that the message broker is running."
            return self.error_response(msg)

        except ImportError as e:
            return self.error_response(str(e))

        return self.success_response()

    def error_response(self, error):
        return HealthcheckResponse(self.name, False, error=error)

    def success_response(self):
        return HealthcheckResponse(self.name, True)


registry.register_healthcheck(
    UrlHealthcheck(name="postcodes.io", url="{}/postcodes/SW1A1AA".format(settings.POSTCODES_IO_URL))
)

registry.register_healthcheck(CeleryWorkersHealthcheck(name="workers"))
