grpcio_health_checking_server_setup_with_service_status.py

1from concurrent import futures
2import grpc
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        experimental_non_blocking=True,
14        experimental_thread_pool=futures.ThreadPoolExecutor(max_workers=1)
15    )
16    
17    # Register the health checking service with the server
18    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)
19
20    # Set the status of a service (empty string "" is the default for the whole server)
21    health_servicer.set("", health_pb2.HealthCheckResponse.SERVING)
22    health_servicer.set("my_service_name", health_pb2.HealthCheckResponse.SERVING)
23
24    server.add_insecure_port('[::]:50051')
25    print("Starting server on port 50051...")
26    server.start()
27    server.wait_for_termination()
28
29if __name__ == '__main__':
30    serve()