codekingpro/portable-devtools

1# Copyright 2016 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"""Translates gRPC's client-side API into gRPC's client-side Beta API."""
15
16import grpc
17from grpc import _common
18from grpc.beta import _metadata
19from grpc.beta import interfaces
20from grpc.framework.common import cardinality
21from grpc.framework.foundation import future
22from grpc.framework.interfaces.face import face
23
24# pylint: disable=too-many-arguments,too-many-locals,unused-argument
25
26_STATUS_CODE_TO_ABORTION_KIND_AND_ABORTION_ERROR_CLASS = {
27    grpc.StatusCode.CANCELLED: (
28        face.Abortion.Kind.CANCELLED,
29        face.CancellationError,
30    ),
31    grpc.StatusCode.UNKNOWN: (
32        face.Abortion.Kind.REMOTE_FAILURE,
33        face.RemoteError,
34    ),
35    grpc.StatusCode.DEADLINE_EXCEEDED: (
36        face.Abortion.Kind.EXPIRED,
37        face.ExpirationError,
38    ),
39    grpc.StatusCode.UNIMPLEMENTED: (
40        face.Abortion.Kind.LOCAL_FAILURE,
41        face.LocalError,
42    ),
43}
44
45
46def _effective_metadata(metadata, metadata_transformer):
47    non_none_metadata = () if metadata is None else metadata
48    if metadata_transformer is None:
49        return non_none_metadata
50    return metadata_transformer(non_none_metadata)
51
52
53def _credentials(grpc_call_options):
54    return None if grpc_call_options is None else grpc_call_options.credentials
55
56
57def _abortion(rpc_error_call):
58    code = rpc_error_call.code()
59    pair = _STATUS_CODE_TO_ABORTION_KIND_AND_ABORTION_ERROR_CLASS.get(code)
60    error_kind = face.Abortion.Kind.LOCAL_FAILURE if pair is None else pair[0]
61    return face.Abortion(
62        error_kind,
63        rpc_error_call.initial_metadata(),
64        rpc_error_call.trailing_metadata(),
65        code,
66        rpc_error_call.details(),
67    )
68
69
70def _abortion_error(rpc_error_call):
71    code = rpc_error_call.code()
72    pair = _STATUS_CODE_TO_ABORTION_KIND_AND_ABORTION_ERROR_CLASS.get(code)
73    exception_class = face.AbortionError if pair is None else pair[1]
74    return exception_class(
75        rpc_error_call.initial_metadata(),
76        rpc_error_call.trailing_metadata(),
77        code,
78        rpc_error_call.details(),
79    )
80
81
82class _InvocationProtocolContext(interfaces.GRPCInvocationContext):
83    def disable_next_request_compression(self):
84        pass  # TODO(https://github.com/grpc/grpc/issues/4078): design, implement.
85
86
87class _Rendezvous(future.Future, face.Call):
88    def __init__(self, response_future, response_iterator, call):
89        self._future = response_future
90        self._iterator = response_iterator
91        self._call = call
92
93    def cancel(self):
94        return self._call.cancel()
95
96    def cancelled(self):
97        return self._future.cancelled()
98
99    def running(self):
100        return self._future.running()
101
102    def done(self):
103        return self._future.done()
104
105    def result(self, timeout=None):
106        try:
107            return self._future.result(timeout=timeout)
108        except grpc.RpcError as rpc_error_call:
109            raise _abortion_error(rpc_error_call)
110        except grpc.FutureTimeoutError:
111            raise future.TimeoutError()
112        except grpc.FutureCancelledError:
113            raise future.CancelledError()
114
115    def exception(self, timeout=None):
116        try:
117            rpc_error_call = self._future.exception(timeout=timeout)
118            if rpc_error_call is None:
119                return None
120            return _abortion_error(rpc_error_call)
121        except grpc.FutureTimeoutError:
122            raise future.TimeoutError()
123        except grpc.FutureCancelledError:
124            raise future.CancelledError()
125
126    def traceback(self, timeout=None):
127        try:
128            return self._future.traceback(timeout=timeout)
129        except grpc.FutureTimeoutError:
130            raise future.TimeoutError()
131        except grpc.FutureCancelledError:
132            raise future.CancelledError()
133
134    def add_done_callback(self, fn):
135        self._future.add_done_callback(lambda _ignored_callback: fn(self))
136
137    def __iter__(self):
138        return self
139
140    def _next(self):
141        try:
142            return next(self._iterator)
143        except grpc.RpcError as rpc_error_call:
144            raise _abortion_error(rpc_error_call)
145
146    def __next__(self):
147        return self._next()
148
149    def next(self):
150        return self._next()
151
152    def is_active(self):
153        return self._call.is_active()
154
155    def time_remaining(self):
156        return self._call.time_remaining()
157
158    def add_abortion_callback(self, abortion_callback):
159        def done_callback():
160            if self.code() is not grpc.StatusCode.OK:
161                abortion_callback(_abortion(self._call))
162
163        registered = self._call.add_callback(done_callback)
164        return None if registered else done_callback()
165
166    def protocol_context(self):
167        return _InvocationProtocolContext()
168
169    def initial_metadata(self):
170        return _metadata.beta(self._call.initial_metadata())
171
172    def terminal_metadata(self):
173        return _metadata.beta(self._call.terminal_metadata())
174
175    def code(self):
176        return self._call.code()
177
178    def details(self):
179        return self._call.details()
180
181
182def _blocking_unary_unary(
183    channel,
184    group,
185    method,
186    timeout,
187    with_call,
188    protocol_options,
189    metadata,
190    metadata_transformer,
191    request,
192    request_serializer,
193    response_deserializer,
194):
195    try:
196        multi_callable = channel.unary_unary(
197            _common.fully_qualified_method(group, method),
198            request_serializer=request_serializer,
199            response_deserializer=response_deserializer,
200        )
201        effective_metadata = _effective_metadata(metadata, metadata_transformer)
202        if with_call:
203            response, call = multi_callable.with_call(
204                request,
205                timeout=timeout,
206                metadata=_metadata.unbeta(effective_metadata),
207                credentials=_credentials(protocol_options),
208            )
209            return response, _Rendezvous(None, None, call)
210        return multi_callable(
211            request,
212            timeout=timeout,
213            metadata=_metadata.unbeta(effective_metadata),
214            credentials=_credentials(protocol_options),
215        )
216    except grpc.RpcError as rpc_error_call:
217        raise _abortion_error(rpc_error_call)
218
219
220def _future_unary_unary(
221    channel,
222    group,
223    method,
224    timeout,
225    protocol_options,
226    metadata,
227    metadata_transformer,
228    request,
229    request_serializer,
230    response_deserializer,
231):
232    multi_callable = channel.unary_unary(
233        _common.fully_qualified_method(group, method),
234        request_serializer=request_serializer,
235        response_deserializer=response_deserializer,
236    )
237    effective_metadata = _effective_metadata(metadata, metadata_transformer)
238    response_future = multi_callable.future(
239        request,
240        timeout=timeout,
241        metadata=_metadata.unbeta(effective_metadata),
242        credentials=_credentials(protocol_options),
243    )
244    return _Rendezvous(response_future, None, response_future)
245
246
247def _unary_stream(
248    channel,
249    group,
250    method,
251    timeout,
252    protocol_options,
253    metadata,
254    metadata_transformer,
255    request,
256    request_serializer,
257    response_deserializer,
258):
259    multi_callable = channel.unary_stream(
260        _common.fully_qualified_method(group, method),
261        request_serializer=request_serializer,
262        response_deserializer=response_deserializer,
263    )
264    effective_metadata = _effective_metadata(metadata, metadata_transformer)
265    response_iterator = multi_callable(
266        request,
267        timeout=timeout,
268        metadata=_metadata.unbeta(effective_metadata),
269        credentials=_credentials(protocol_options),
270    )
271    return _Rendezvous(None, response_iterator, response_iterator)
272
273
274def _blocking_stream_unary(
275    channel,
276    group,
277    method,
278    timeout,
279    with_call,
280    protocol_options,
281    metadata,
282    metadata_transformer,
283    request_iterator,
284    request_serializer,
285    response_deserializer,
286):
287    try:
288        multi_callable = channel.stream_unary(
289            _common.fully_qualified_method(group, method),
290            request_serializer=request_serializer,
291            response_deserializer=response_deserializer,
292        )
293        effective_metadata = _effective_metadata(metadata, metadata_transformer)
294        if with_call:
295            response, call = multi_callable.with_call(
296                request_iterator,
297                timeout=timeout,
298                metadata=_metadata.unbeta(effective_metadata),
299                credentials=_credentials(protocol_options),
300            )
301            return response, _Rendezvous(None, None, call)
302        return multi_callable(
303            request_iterator,
304            timeout=timeout,
305            metadata=_metadata.unbeta(effective_metadata),
306            credentials=_credentials(protocol_options),
307        )
308    except grpc.RpcError as rpc_error_call:
309        raise _abortion_error(rpc_error_call)
310
311
312def _future_stream_unary(
313    channel,
314    group,
315    method,
316    timeout,
317    protocol_options,
318    metadata,
319    metadata_transformer,
320    request_iterator,
321    request_serializer,
322    response_deserializer,
323):
324    multi_callable = channel.stream_unary(
325        _common.fully_qualified_method(group, method),
326        request_serializer=request_serializer,
327        response_deserializer=response_deserializer,
328    )
329    effective_metadata = _effective_metadata(metadata, metadata_transformer)
330    response_future = multi_callable.future(
331        request_iterator,
332        timeout=timeout,
333        metadata=_metadata.unbeta(effective_metadata),
334        credentials=_credentials(protocol_options),
335    )
336    return _Rendezvous(response_future, None, response_future)
337
338
339def _stream_stream(
340    channel,
341    group,
342    method,
343    timeout,
344    protocol_options,
345    metadata,
346    metadata_transformer,
347    request_iterator,
348    request_serializer,
349    response_deserializer,
350):
351    multi_callable = channel.stream_stream(
352        _common.fully_qualified_method(group, method),
353        request_serializer=request_serializer,
354        response_deserializer=response_deserializer,
355    )
356    effective_metadata = _effective_metadata(metadata, metadata_transformer)
357    response_iterator = multi_callable(
358        request_iterator,
359        timeout=timeout,
360        metadata=_metadata.unbeta(effective_metadata),
361        credentials=_credentials(protocol_options),
362    )
363    return _Rendezvous(None, response_iterator, response_iterator)
364
365
366class _UnaryUnaryMultiCallable(face.UnaryUnaryMultiCallable):
367    def __init__(
368        self,
369        channel,
370        group,
371        method,
372        metadata_transformer,
373        request_serializer,
374        response_deserializer,
375    ):
376        self._channel = channel
377        self._group = group
378        self._method = method
379        self._metadata_transformer = metadata_transformer
380        self._request_serializer = request_serializer
381        self._response_deserializer = response_deserializer
382
383    def __call__(
384        self,
385        request,
386        timeout,
387        metadata=None,
388        with_call=False,
389        protocol_options=None,
390    ):
391        return _blocking_unary_unary(
392            self._channel,
393            self._group,
394            self._method,
395            timeout,
396            with_call,
397            protocol_options,
398            metadata,
399            self._metadata_transformer,
400            request,
401            self._request_serializer,
402            self._response_deserializer,
403        )
404
405    def future(self, request, timeout, metadata=None, protocol_options=None):
406        return _future_unary_unary(
407            self._channel,
408            self._group,
409            self._method,
410            timeout,
411            protocol_options,
412            metadata,
413            self._metadata_transformer,
414            request,
415            self._request_serializer,
416            self._response_deserializer,
417        )
418
419    def event(
420        self,
421        request,
422        receiver,
423        abortion_callback,
424        timeout,
425        metadata=None,
426        protocol_options=None,
427    ):
428        raise NotImplementedError()
429
430
431class _UnaryStreamMultiCallable(face.UnaryStreamMultiCallable):
432    def __init__(
433        self,
434        channel,
435        group,
436        method,
437        metadata_transformer,
438        request_serializer,
439        response_deserializer,
440    ):
441        self._channel = channel
442        self._group = group
443        self._method = method
444        self._metadata_transformer = metadata_transformer
445        self._request_serializer = request_serializer
446        self._response_deserializer = response_deserializer
447
448    def __call__(self, request, timeout, metadata=None, protocol_options=None):
449        return _unary_stream(
450            self._channel,
451            self._group,
452            self._method,
453            timeout,
454            protocol_options,
455            metadata,
456            self._metadata_transformer,
457            request,
458            self._request_serializer,
459            self._response_deserializer,
460        )
461
462    def event(
463        self,
464        request,
465        receiver,
466        abortion_callback,
467        timeout,
468        metadata=None,
469        protocol_options=None,
470    ):
471        raise NotImplementedError()
472
473
474class _StreamUnaryMultiCallable(face.StreamUnaryMultiCallable):
475    def __init__(
476        self,
477        channel,
478        group,
479        method,
480        metadata_transformer,
481        request_serializer,
482        response_deserializer,
483    ):
484        self._channel = channel
485        self._group = group
486        self._method = method
487        self._metadata_transformer = metadata_transformer
488        self._request_serializer = request_serializer
489        self._response_deserializer = response_deserializer
490
491    def __call__(
492        self,
493        request_iterator,
494        timeout,
495        metadata=None,
496        with_call=False,
497        protocol_options=None,
498    ):
499        return _blocking_stream_unary(
500            self._channel,
501            self._group,
502            self._method,
503            timeout,
504            with_call,
505            protocol_options,
506            metadata,
507            self._metadata_transformer,
508            request_iterator,
509            self._request_serializer,
510            self._response_deserializer,
511        )
512
513    def future(
514        self, request_iterator, timeout, metadata=None, protocol_options=None
515    ):
516        return _future_stream_unary(
517            self._channel,
518            self._group,
519            self._method,
520            timeout,
521            protocol_options,
522            metadata,
523            self._metadata_transformer,
524            request_iterator,
525            self._request_serializer,
526            self._response_deserializer,
527        )
528
529    def event(
530        self,
531        receiver,
532        abortion_callback,
533        timeout,
534        metadata=None,
535        protocol_options=None,
536    ):
537        raise NotImplementedError()
538
539
540class _StreamStreamMultiCallable(face.StreamStreamMultiCallable):
541    def __init__(
542        self,
543        channel,
544        group,
545        method,
546        metadata_transformer,
547        request_serializer,
548        response_deserializer,
549    ):
550        self._channel = channel
551        self._group = group
552        self._method = method
553        self._metadata_transformer = metadata_transformer
554        self._request_serializer = request_serializer
555        self._response_deserializer = response_deserializer
556
557    def __call__(
558        self, request_iterator, timeout, metadata=None, protocol_options=None
559    ):
560        return _stream_stream(
561            self._channel,
562            self._group,
563            self._method,
564            timeout,
565            protocol_options,
566            metadata,
567            self._metadata_transformer,
568            request_iterator,
569            self._request_serializer,
570            self._response_deserializer,
571        )
572
573    def event(
574        self,
575        receiver,
576        abortion_callback,
577        timeout,
578        metadata=None,
579        protocol_options=None,
580    ):
581        raise NotImplementedError()
582
583
584class _GenericStub(face.GenericStub):
585    def __init__(
586        self,
587        channel,
588        metadata_transformer,
589        request_serializers,
590        response_deserializers,
591    ):
592        self._channel = channel
593        self._metadata_transformer = metadata_transformer
594        self._request_serializers = request_serializers or {}
595        self._response_deserializers = response_deserializers or {}
596
597    def blocking_unary_unary(
598        self,
599        group,
600        method,
601        request,
602        timeout,
603        metadata=None,
604        with_call=None,
605        protocol_options=None,
606    ):
607        request_serializer = self._request_serializers.get(
608            (
609                group,
610                method,
611            )
612        )
613        response_deserializer = self._response_deserializers.get(
614            (
615                group,
616                method,
617            )
618        )
619        return _blocking_unary_unary(
620            self._channel,
621            group,
622            method,
623            timeout,
624            with_call,
625            protocol_options,
626            metadata,
627            self._metadata_transformer,
628            request,
629            request_serializer,
630            response_deserializer,
631        )
632
633    def future_unary_unary(
634        self,
635        group,
636        method,
637        request,
638        timeout,
639        metadata=None,
640        protocol_options=None,
641    ):
642        request_serializer = self._request_serializers.get(
643            (
644                group,
645                method,
646            )
647        )
648        response_deserializer = self._response_deserializers.get(
649            (
650                group,
651                method,
652            )
653        )
654        return _future_unary_unary(
655            self._channel,
656            group,
657            method,
658            timeout,
659            protocol_options,
660            metadata,
661            self._metadata_transformer,
662            request,
663            request_serializer,
664            response_deserializer,
665        )
666
667    def inline_unary_stream(
668        self,
669        group,
670        method,
671        request,
672        timeout,
673        metadata=None,
674        protocol_options=None,
675    ):
676        request_serializer = self._request_serializers.get(
677            (
678                group,
679                method,
680            )
681        )
682        response_deserializer = self._response_deserializers.get(
683            (
684                group,
685                method,
686            )
687        )
688        return _unary_stream(
689            self._channel,
690            group,
691            method,
692            timeout,
693            protocol_options,
694            metadata,
695            self._metadata_transformer,
696            request,
697            request_serializer,
698            response_deserializer,
699        )
700
701    def blocking_stream_unary(
702        self,
703        group,
704        method,
705        request_iterator,
706        timeout,
707        metadata=None,
708        with_call=None,
709        protocol_options=None,
710    ):
711        request_serializer = self._request_serializers.get(
712            (
713                group,
714                method,
715            )
716        )
717        response_deserializer = self._response_deserializers.get(
718            (
719                group,
720                method,
721            )
722        )
723        return _blocking_stream_unary(
724            self._channel,
725            group,
726            method,
727            timeout,
728            with_call,
729            protocol_options,
730            metadata,
731            self._metadata_transformer,
732            request_iterator,
733            request_serializer,
734            response_deserializer,
735        )
736
737    def future_stream_unary(
738        self,
739        group,
740        method,
741        request_iterator,
742        timeout,
743        metadata=None,
744        protocol_options=None,
745    ):
746        request_serializer = self._request_serializers.get(
747            (
748                group,
749                method,
750            )
751        )
752        response_deserializer = self._response_deserializers.get(
753            (
754                group,
755                method,
756            )
757        )
758        return _future_stream_unary(
759            self._channel,
760            group,
761            method,
762            timeout,
763            protocol_options,
764            metadata,
765            self._metadata_transformer,
766            request_iterator,
767            request_serializer,
768            response_deserializer,
769        )
770
771    def inline_stream_stream(
772        self,
773        group,
774        method,
775        request_iterator,
776        timeout,
777        metadata=None,
778        protocol_options=None,
779    ):
780        request_serializer = self._request_serializers.get(
781            (
782                group,
783                method,
784            )
785        )
786        response_deserializer = self._response_deserializers.get(
787            (
788                group,
789                method,
790            )
791        )
792        return _stream_stream(
793            self._channel,
794            group,
795            method,
796            timeout,
797            protocol_options,
798            metadata,
799            self._metadata_transformer,
800            request_iterator,
801            request_serializer,
802            response_deserializer,
803        )
804
805    def event_unary_unary(
806        self,
807        group,
808        method,
809        request,
810        receiver,
811        abortion_callback,
812        timeout,
813        metadata=None,
814        protocol_options=None,
815    ):
816        raise NotImplementedError()
817
818    def event_unary_stream(
819        self,
820        group,
821        method,
822        request,
823        receiver,
824        abortion_callback,
825        timeout,
826        metadata=None,
827        protocol_options=None,
828    ):
829        raise NotImplementedError()
830
831    def event_stream_unary(
832        self,
833        group,
834        method,
835        receiver,
836        abortion_callback,
837        timeout,
838        metadata=None,
839        protocol_options=None,
840    ):
841        raise NotImplementedError()
842
843    def event_stream_stream(
844        self,
845        group,
846        method,
847        receiver,
848        abortion_callback,
849        timeout,
850        metadata=None,
851        protocol_options=None,
852    ):
853        raise NotImplementedError()
854
855    def unary_unary(self, group, method):
856        request_serializer = self._request_serializers.get(
857            (
858                group,
859                method,
860            )
861        )
862        response_deserializer = self._response_deserializers.get(
863            (
864                group,
865                method,
866            )
867        )
868        return _UnaryUnaryMultiCallable(
869            self._channel,
870            group,
871            method,
872            self._metadata_transformer,
873            request_serializer,
874            response_deserializer,
875        )
876
877    def unary_stream(self, group, method):
878        request_serializer = self._request_serializers.get(
879            (
880                group,
881                method,
882            )
883        )
884        response_deserializer = self._response_deserializers.get(
885            (
886                group,
887                method,
888            )
889        )
890        return _UnaryStreamMultiCallable(
891            self._channel,
892            group,
893            method,
894            self._metadata_transformer,
895            request_serializer,
896            response_deserializer,
897        )
898
899    def stream_unary(self, group, method):
900        request_serializer = self._request_serializers.get(
901            (
902                group,
903                method,
904            )
905        )
906        response_deserializer = self._response_deserializers.get(
907            (
908                group,
909                method,
910            )
911        )
912        return _StreamUnaryMultiCallable(
913            self._channel,
914            group,
915            method,
916            self._metadata_transformer,
917            request_serializer,
918            response_deserializer,
919        )
920
921    def stream_stream(self, group, method):
922        request_serializer = self._request_serializers.get(
923            (
924                group,
925                method,
926            )
927        )
928        response_deserializer = self._response_deserializers.get(
929            (
930                group,
931                method,
932            )
933        )
934        return _StreamStreamMultiCallable(
935            self._channel,
936            group,
937            method,
938            self._metadata_transformer,
939            request_serializer,
940            response_deserializer,
941        )
942
943    def __enter__(self):
944        return self
945
946    def __exit__(self, exc_type, exc_val, exc_tb):
947        return False
948
949
950class _DynamicStub(face.DynamicStub):
951    def __init__(self, backing_generic_stub, group, cardinalities):
952        self._generic_stub = backing_generic_stub
953        self._group = group
954        self._cardinalities = cardinalities
955
956    def __getattr__(self, attr):
957        method_cardinality = self._cardinalities.get(attr)
958        if method_cardinality is cardinality.Cardinality.UNARY_UNARY:
959            return self._generic_stub.unary_unary(self._group, attr)
960        if method_cardinality is cardinality.Cardinality.UNARY_STREAM:
961            return self._generic_stub.unary_stream(self._group, attr)
962        if method_cardinality is cardinality.Cardinality.STREAM_UNARY:
963            return self._generic_stub.stream_unary(self._group, attr)
964        if method_cardinality is cardinality.Cardinality.STREAM_STREAM:
965            return self._generic_stub.stream_stream(self._group, attr)
966        raise AttributeError(
967            '_DynamicStub object has no attribute "%s"!' % attr
968        )
969
970    def __enter__(self):
971        return self
972
973    def __exit__(self, exc_type, exc_val, exc_tb):
974        return False
975
976
977def generic_stub(
978    channel,
979    host,
980    metadata_transformer,
981    request_serializers,
982    response_deserializers,
983):
984    return _GenericStub(
985        channel,
986        metadata_transformer,
987        request_serializers,
988        response_deserializers,
989    )
990
991
992def dynamic_stub(
993    channel,
994    service,
995    cardinalities,
996    host,
997    metadata_transformer,
998    request_serializers,
999    response_deserializers,
1000):
1001    return _DynamicStub(
1002        _GenericStub(
1003            channel,
1004            metadata_transformer,
1005            request_serializers,
1006            response_deserializers,
1007        ),
1008        service,
1009        cardinalities,
1010    )
1011