-
Notifications
You must be signed in to change notification settings - Fork 45
/
healthcheck.py
74 lines (59 loc) · 1.88 KB
/
healthcheck.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import traceback
from flask import Flask
def healthcheck() -> tuple[dict, int]:
"""
This preforms a very basic check to make sure that the database
and the cache are reachable on the current instance. If the status
comes back as "Unhealthy", then there was some exception that was
raised when attempting to connect to it. It may only be a temporary
issue, or something specific to the instance it is running on.
response = {
"service": "anubis-api",
"api": "Healthy",
"db": "Healthy",
"cache": "Healthy",
"commit": "abcdef123456"
}
"""
from anubis.models import Config
from anubis.utils.cache import cache_health
from anubis.utils.logging import logger
from anubis.env import env
# Construct basic status & status code
status_code = 200
status = {
"service": env.LOGGER_NAME,
"api": "Healthy",
"db": "Healthy",
"cache": "Healthy",
"commit": env.GIT_TAG
}
# Attempt to connect to db
try:
Config.query.all()
# If there is any issue, mark the db
# connection as Unhealthy and log the error
except Exception as e:
status["db"] = "Unhealthy"
status_code = 500
logger.error(traceback.format_exc())
# Attempt to connect to cache
try:
cache_health()
# If there is any issue, mark the cache
# connection as Unhealthy and log the error
except Exception as e:
status["cache"] = "Unhealthy"
status_code = 500
logger.error(traceback.format_exc())
return status, status_code
def add_healthcheck(app: Flask):
from anubis.utils.http.decorators import json_response
@app.route("/")
@json_response
def index():
"""
Healthcheck endpoint.
"""
# Pass back status and status_code
return healthcheck()