Implement health service and proper shutdown

main.py

@@ -1,5 +1,6 @@
 import concurrent.futures
 import os
+import signal
 import sys
 import time
 
@@ -21,6 +22,7 @@
 import tucana.generated.velorum.info_pb2_grpc as info_pb2_grpc
 import tucana.generated.velorum.info_pb2 as info_pb2
 from grpc_reflection.v1alpha import reflection
+from grpc_health.v1 import health, health_pb2, health_pb2_grpc
 
 from src.endpoint.info.info_endpoint import InfoService
 from src.endpoint.generation.generate_endpoint import GenerateService
@@ -34,6 +36,11 @@
     generate_pb2_grpc.add_GenerateServiceServicer_to_server(GenerateService(), server)
     info_pb2_grpc.add_InfoServiceServicer_to_server(InfoService(), server)
 
+    health_servicer = health.HealthServicer()
+    health_servicer.set('liveness', health_pb2.HealthCheckResponse.SERVING)
+    health_servicer.set('readiness', health_pb2.HealthCheckResponse.SERVING)
+    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)
+
     port = "0.0.0.0:50051"
     server.add_insecure_port(port)
     print(f"Velorum GenerateService läuft auf {port}...")
@@ -47,9 +54,12 @@
 
     server.start()
 
-    try:
-        while True:
-            time.sleep(86400)
-    except KeyboardInterrupt:
+    def shutdown(signum, frame):
         print("\nServer wird sauber heruntergefahren...")
-        server.stop(0)
+        health_servicer.enter_graceful_shutdown()
+        server.stop(5).wait()
+
+    signal.signal(signal.SIGTERM, shutdown)
+    signal.signal(signal.SIGINT, shutdown)
+
+    server.wait_for_termination()

pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">=3.10,<3.13"
 dependencies = [
     "apscheduler>=3.11.2",
     "grpcio>=1.80.0",
+    "grpcio-health-checking>=1.80.0",
     "grpcio-reflection>=1.80.0",
     "instructor>=1.15.1",
     "PyJWT>=2.4.0",

src/interceptor/auth_interceptor.py

@@ -8,6 +8,10 @@
 class AuthInterceptor(grpc.ServerInterceptor):
 
     def intercept_service(self, continuation, handler_call_details):
+        method = handler_call_details.method or ''
+        if method.startswith('/grpc.health.v1.Health/'):
+            return continuation(handler_call_details)
+
         metadata = dict(handler_call_details.invocation_metadata or [])
         auth_token = metadata.get('authorization', '')