Add health check endpoint at /api/healthz (#128)

Author: somethingnew2-0

api/app.py

@@ -26,6 +26,7 @@
     bugs_views,
     exception_views,
     groups_views,
+    health_check_views,
     roles_views,
     tags_views,
     users_views,
@@ -128,6 +129,9 @@ def create_app(testing: Optional[bool] = False) -> Flask:
 
     @app.before_request
     def authenticate_request() -> Optional[ResponseReturnValue]:
+        # Skip authentication for health check endpoint
+        if request.path.startswith("/api/healthz"):
+            return None
         return AuthenticationHelpers.authenticate_user(request)
 
     @app.after_request
@@ -202,6 +206,7 @@ def add_headers(response: Response) -> ResponseReturnValue:
     warnings.filterwarnings("ignore", message="Only explicitly-declared fields will be included in the Schema Object")
 
     app.register_blueprint(exception_views.bp)
+    app.register_blueprint(health_check_views.bp)
     app.register_blueprint(access_requests_views.bp)
     access_requests_views.register_docs()
     app.register_blueprint(apps_views.bp)

api/views/health_check_views.py

@@ -0,0 +1,20 @@
+from typing import Any
+
+from flask import Blueprint
+from sqlalchemy import text
+
+from api.extensions import db
+
+bp_name = "api-health-check"
+bp_url_prefix = "/api/healthz"
+bp = Blueprint(bp_name, __name__, url_prefix=bp_url_prefix)
+
+
+@bp.route("", methods=["GET"])
+def health_check() -> Any:
+    try:
+        db.session.execute(text("SELECT 1"))
+    except Exception as e:
+        return {"status": "error", "error": str(e)}, 500
+
+    return {"status": "ok"}, 200

examples/kubernetes/deployment.yaml

@@ -56,4 +56,8 @@ spec:
           name: access
           ports:
             - containerPort: 3000
+          livenessProbe:
+            httpGet:
+              path: /api/healthz
+              port: 3000
       serviceAccountName: access

tests/test_health_check.py

@@ -0,0 +1,13 @@
+from flask import Flask, url_for
+from flask.testing import FlaskClient
+from flask_sqlalchemy import SQLAlchemy
+
+
+def test_health_check(app: Flask, client: FlaskClient, db: SQLAlchemy) -> None:
+    # test unauthenticated requests by setting the current user email to "Unauthenticated"
+    app.config["CURRENT_OKTA_USER_EMAIL"] = "Unauthenticated"
+
+    # test 200
+    health_check_url = url_for("api-health-check.health_check")
+    rep = client.get(health_check_url)
+    assert rep.status_code == 200