healthchecks.py

"""
DJANGO
"""

from django.http import JsonResponse


def simple(request):
    return JsonResponse({'health': 'OK'})

"""
FLASK
"""

from flask import Blueprint, current_app, jsonify
from pykafka import KafkaClient
from pykafka.exceptions import NoBrokersAvailableError

healthchecks = Blueprint('healthchecks', __name__)


@healthchecks.route('/memcached')
def memcached_healthcheck():
    current_app.cache.set('test_key', 'test_value', timeout=20)
    if current_app.cache.get('test_key') == 'test_value':
        return jsonify(health='OK', cache='OK')
    else:
        return jsonify(health='OK', cache='Error'), 400


@healthchecks.route('/simple')
def simple_healthcheck():
    return jsonify(health='OK')
    
@healthchecks.route('/kafka')
def kafka_healthcheck():
    try:
        KafkaClient(NEW_KAFKA_NETLOC, socket_timeout_ms=500)
    except NoBrokersAvailableError:
        return jsonify(health='OK', kafka='Error'), 400
    return jsonify(health='OK', kafka='OK')