grpc_health_checking_servicer_setup_and_status_management.py

1from concurrent import futures
2import logging
3
4import grpc
5from grpc_health.v1 import health
6from grpc_health.v1 import health_pb2
7from grpc_health.v1 import health_pb2_grpc
8
9
10def serve():
11    # Create a gRPC server
12    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
13
14    # Create a health-checking servant
15    health_servant = health.HealthServicer()
16
17    # Register the health-checking servant with the server
18    health_pb2_grpc.add_HealthServicer_to_server(health_servant, server)
19
20    # Set the initial status for the overall server or specific services
21    # Empty string represents the default health status of the whole server
22    health_servant.set("", health_pb2.HealthCheckResponse.SERVING)
23    health_servant.set("v1.MyService", health_pb2.HealthCheckResponse.SERVING)
24
25    # (Optional) You can change status later as needed
26    # health_servant.set("v1.MyService", health_pb2.HealthCheckResponse.NOT_SERVING)
27
28    server.add_insecure_port('[::]:50051')
29    server.start()
30    print("Server started on port 50051 with health checking enabled.")
31    server.wait_for_termination()
32
33
34if __name__ == '__main__':
35    logging.basicConfig()
36    serve()