Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/grpc/_compression.py: 32%

1# Copyright 2019 The 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 

15from __future__ import annotations 

16 

17from typing import Optional 

18 

19import grpc 

20from grpc._cython import cygrpc 

21from grpc._typing import MetadataType 

22 

23NoCompression = cygrpc.CompressionAlgorithm.none 

24Deflate = cygrpc.CompressionAlgorithm.deflate 

25Gzip = cygrpc.CompressionAlgorithm.gzip 

26 

27_METADATA_STRING_MAPPING = { 

28 NoCompression: "identity", 

29 Deflate: "deflate", 

30 Gzip: "gzip", 

31} 

32 

33 

34def _compression_algorithm_to_metadata_value( 

35 compression: grpc.Compression, 

36) -> str: 

37 return _METADATA_STRING_MAPPING[compression] 

38 

39 

40def compression_algorithm_to_metadata(compression: grpc.Compression): 

41 return ( 

42 cygrpc.GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY, 

43 _compression_algorithm_to_metadata_value(compression), 

44 ) 

45 

46 

47def create_channel_option(compression: Optional[grpc.Compression]): 

48 return ( 

49 ((cygrpc.GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM, int(compression)),) 

50 if compression 

51 else () 

52 ) 

53 

54 

55def augment_metadata( 

56 metadata: Optional[MetadataType], compression: Optional[grpc.Compression] 

57): 

58 if not metadata and not compression: 

59 return None 

60 base_metadata = tuple(metadata) if metadata else () 

61 compression_metadata = ( 

62 (compression_algorithm_to_metadata(compression),) if compression else () 

63 ) 

64 return base_metadata + compression_metadata 

65 

66 

67__all__ = ( 

68 "NoCompression", 

69 "Deflate", 

70 "Gzip", 

71)