1# Copyright 2023 gRPC authors. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14"""gRPC Python helloworld.Greeter client with health checking.""" 15 16import logging 17from time import sleep 18 19import grpc 20from grpc_health.v1 import health_pb2 21from grpc_health.v1 import health_pb2_grpc 22import helloworld_pb2 23import helloworld_pb2_grpc 24 25 26def unary_call(stub: helloworld_pb2_grpc.GreeterStub, message: str): 27 response = stub.SayHello( 28 helloworld_pb2.HelloRequest(name=message), timeout=3 29 ) 30 print(f"Greeter client received: {response.message}") 31 32 33def health_check_call(stub: health_pb2_grpc.HealthStub): 34 request = health_pb2.HealthCheckRequest(service="helloworld.Greeter") 35 resp = stub.Check(request) 36 if resp.status == health_pb2.HealthCheckResponse.SERVING: 37 print("server is serving") 38 elif resp.status == health_pb2.HealthCheckResponse.NOT_SERVING: 39 print("server stopped serving") 40 41 42def run(): 43 with grpc.insecure_channel("localhost:50051") as channel: 44 stub = helloworld_pb2_grpc.GreeterStub(channel) 45 health_stub = health_pb2_grpc.HealthStub(channel) 46 # Should succeed 47 unary_call(stub, "you") 48 49 # Check health status every 1 second for 30 seconds 50 for _ in range(30): 51 health_check_call(health_stub) 52 sleep(1) 53 54 55if __name__ == "__main__": 56 logging.basicConfig() 57 run() 58