generic_client_interceptor.py 2.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556
  1. # Copyright 2017 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. """Base class for interceptors that operate on all RPC types."""
  15. import grpc
  16. class _GenericClientInterceptor(grpc.UnaryUnaryClientInterceptor,
  17. grpc.UnaryStreamClientInterceptor,
  18. grpc.StreamUnaryClientInterceptor,
  19. grpc.StreamStreamClientInterceptor):
  20. def __init__(self, interceptor_function):
  21. self._fn = interceptor_function
  22. def intercept_unary_unary(self, continuation, client_call_details, request):
  23. new_details, new_request_iterator, postprocess = self._fn(
  24. client_call_details, iter((request,)), False, False)
  25. response = continuation(new_details, next(new_request_iterator))
  26. return postprocess(response) if postprocess else response
  27. def intercept_unary_stream(self, continuation, client_call_details,
  28. request):
  29. new_details, new_request_iterator, postprocess = self._fn(
  30. client_call_details, iter((request,)), False, True)
  31. response_it = continuation(new_details, next(new_request_iterator))
  32. return postprocess(response_it) if postprocess else response_it
  33. def intercept_stream_unary(self, continuation, client_call_details,
  34. request_iterator):
  35. new_details, new_request_iterator, postprocess = self._fn(
  36. client_call_details, request_iterator, True, False)
  37. response = continuation(new_details, new_request_iterator)
  38. return postprocess(response) if postprocess else response
  39. def intercept_stream_stream(self, continuation, client_call_details,
  40. request_iterator):
  41. new_details, new_request_iterator, postprocess = self._fn(
  42. client_call_details, request_iterator, True, True)
  43. response_it = continuation(new_details, new_request_iterator)
  44. return postprocess(response_it) if postprocess else response_it
  45. def create(intercept_call):
  46. return _GenericClientInterceptor(intercept_call)