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

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) -> str: 

36 return _METADATA_STRING_MAPPING[compression] 

37 

38 

39def compression_algorithm_to_metadata(compression: grpc.Compression): 

40 return (cygrpc.GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY, 

41 _compression_algorithm_to_metadata_value(compression)) 

42 

43 

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

45 return ((cygrpc.GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM, 

46 int(compression)),) if compression else () 

47 

48 

49def augment_metadata(metadata: Optional[MetadataType], 

50 compression: Optional[grpc.Compression]): 

51 if not metadata and not compression: 

52 return None 

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

54 compression_metadata = ( 

55 compression_algorithm_to_metadata(compression),) if compression else () 

56 return base_metadata + compression_metadata 

57 

58 

59__all__ = ( 

60 "NoCompression", 

61 "Deflate", 

62 "Gzip", 

63)