72 lines
1.8 KiB
Python
72 lines
1.8 KiB
Python
import json
|
|
import os
|
|
import socket
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
import pymysql
|
|
|
|
|
|
def get_env(name, default=None):
|
|
value = os.environ.get(name, default)
|
|
return value if value is not None else default
|
|
|
|
|
|
def db_check():
|
|
host = get_env("DB_HOST")
|
|
port = int(get_env("DB_PORT", "3306"))
|
|
user = get_env("DB_USER")
|
|
password = get_env("DB_PASSWORD")
|
|
dbname = get_env("DB_NAME", "postgres")
|
|
|
|
if not host or not user or not password:
|
|
return False, "missing DB_HOST/DB_USER/DB_PASSWORD"
|
|
|
|
conn = pymysql.connect(
|
|
host=host,
|
|
port=port,
|
|
user=user,
|
|
password=password,
|
|
database=dbname,
|
|
connect_timeout=5,
|
|
read_timeout=5,
|
|
write_timeout=5,
|
|
)
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT 1")
|
|
cur.fetchone()
|
|
return True, "ok"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
ok, msg = False, ""
|
|
status = 500
|
|
try:
|
|
ok, msg = db_check()
|
|
status = 200 if ok else 500
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
msg = str(exc)
|
|
|
|
body = {
|
|
"status": "ok" if ok else "error",
|
|
"message": msg,
|
|
"hostname": socket.gethostname(),
|
|
}
|
|
data = json.dumps(body).encode("utf-8")
|
|
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
port = int(get_env("PORT", "8080"))
|
|
server = HTTPServer(("0.0.0.0", port), Handler)
|
|
print(f"listening on :{port}")
|
|
server.serve_forever()
|