grpc_stubs_typed_server_setup_with_interceptor.py

1import grpc
2from concurrent import futures
3from typing import Any, Iterable, Optional
4
5# grpc-stubs allows type checkers to validate these components
6class MyInterceptor(grpc.ServerInterceptor):
7    def intercept_service(
8        self,
9        continuation: Any,
10        handler_call_details: grpc.HandlerCallDetails
11    ) -> grpc.RpcMethodHandler:
12        return continuation(handler_call_details)
13
14def serve() -> None:
15    # grpc-stubs provides type hints for the server instance and its methods
16    server: grpc.Server = grpc.server(
17        futures.ThreadPoolExecutor(max_workers=10),
18        interceptors=(MyInterceptor(),)
19    )
20    
21    # Example of a typed port binding
22    port: int = server.add_insecure_port('[::]:50051')
23    print(f"Server started on port {port}")
24    
25    server.start()
26    server.wait_for_termination()
27
28if __name__ == '__main__':
29    serve()