grpc_health_checking_service_registration_quickstart.py

1import grpc
2from concurrent import futures
3from grpc_health.v1 import health
4from grpc_health.v1 import health_pb2
5from grpc_health.v1 import health_pb2_grpc
6
7def serve():
8    # Create the gRPC server
9    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
10
11    # Create the HealthServicer
12    health_servicer = health.HealthServicer()
13
14    # Register the HealthServicer with the server
15    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)
16
17    # Set the initial status for the server or specific services
18    # Empty string represents the overall server health
19    health_servicer.set("", health_pb2.HealthCheckResponse.SERVING)
20    # You can also set status for specific service names
21    health_servicer.set("my.service.v1", health_pb2.HealthCheckResponse.SERVING)
22
23    # Standard server setup
24    server.add_insecure_port('[::]:50051')
25    server.start()
26    print("Server started on port 50051 with health checking enabled.")
27    server.wait_for_termination()
28
29if __name__ == '__main__':
30    serve()