Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/looker_sdk/rtl/serialize.py: 54%

1# The MIT License (MIT) 

2# 

3# Copyright (c) 2019 Looker Data Sciences, Inc. 

4# 

5# Permission is hereby granted, free of charge, to any person obtaining a copy 

6# of this software and associated documentation files (the "Software"), to deal 

7# in the Software without restriction, including without limitation the rights 

8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 

9# copies of the Software, and to permit persons to whom the Software is 

10# furnished to do so, subject to the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included in 

13# all copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 

21# THE SOFTWARE. 

22 

23"""Deserialize API response into models 

24""" 

25import datetime 

26import enum 

27import functools 

28import json 

29import keyword 

30import sys 

31from typing import ( 

32 Callable, 

33 MutableMapping, 

34 Sequence, 

35 Type, 

36 Union, 

37) 

38 

39import cattr 

40 

41from looker_sdk.rtl import model, hooks 

42 

43 

44class DeserializeError(Exception): 

45 """Improperly formatted data to deserialize.""" 

46 

47 

48TModelOrSequence = Union[ 

49 MutableMapping[str, str], 

50 Sequence[int], 

51 Sequence[str], 

52 model.Model, 

53 Sequence[model.Model], 

54] 

55TDeserializeReturn = TModelOrSequence 

56TStructure = Union[Type[Sequence[int]], Type[Sequence[str]], Type[TDeserializeReturn]] 

57TDeserialize = Callable[[str, TStructure], TDeserializeReturn] 

58TSerialize = Callable[[TModelOrSequence], bytes] 

59 

60 

61def deserialize( 

62 *, data: str, structure: TStructure, converter: cattr.Converter 

63) -> TDeserializeReturn: 

64 """Translate API data into models.""" 

65 try: 

66 data = json.loads(data) 

67 except json.JSONDecodeError as ex: 

68 raise DeserializeError(f"Bad json {ex}") 

69 try: 

70 response: TDeserializeReturn = converter.structure( # type: ignore 

71 data, structure 

72 ) 

73 except (TypeError, AttributeError) as ex: 

74 raise DeserializeError(f"Bad data {ex}") 

75 return response 

76 

77 

78def serialize(*, api_model: TModelOrSequence, converter: cattr.Converter) -> bytes: 

79 """Translate api_model into formdata encoded json bytes""" 

80 data = converter.unstructure(api_model) # type: ignore 

81 return json.dumps(data,default=lambda o: o.__dict__).encode("utf-8") # type: ignore 

82 

83 

84def forward_ref_structure_hook(context, converter, data, forward_ref): 

85 """Applied to ForwardRef model and enum annotations 

86 

87 - Map reserved words in json keys to approriate (safe) names in model. 

88 - handle ForwardRef types until github.com/Tinche/cattrs/pull/42/ is fixed 

89 Note: this is the reason we need a "context" param and have to use a 

90 partial func to register the hook. Once the issue is resolved we can 

91 remove "context" and the partial. 

92 """ 

93 data = hooks.tr_data_keys(data) 

94 actual_type = eval(forward_ref.__forward_arg__, context, locals()) 

95 if issubclass(actual_type, enum.Enum): 

96 instance = converter.structure(data, actual_type) 

97 elif issubclass(actual_type, model.Model): 

98 # cannot use converter.structure - recursion error 

99 instance = converter.structure_attrs_fromdict(data, actual_type) 

100 else: 

101 raise DeserializeError(f"Unknown type to deserialize: {actual_type}") 

102 return instance 

103 

104 

105def translate_keys_structure_hook(converter, data, model_type): 

106 """Applied only to models.Model""" 

107 new_data = hooks.tr_data_keys(data) 

108 ret = converter.structure_attrs_fromdict(new_data, model_type) 

109 return ret 

110 

111converter40 = cattr.Converter() 

112deserialize40 = functools.partial(deserialize, converter=converter40) 

113serialize40 = functools.partial(serialize, converter=converter40) 

114 

115converter40.register_structure_hook(datetime.datetime, hooks.datetime_structure_hook) 

116unstructure_hook40 = functools.partial(hooks.unstructure_hook, converter40) # type: ignore 

117converter40.register_unstructure_hook(model.Model, unstructure_hook40) # type: ignore 

118converter40.register_unstructure_hook( 

119 datetime.datetime, hooks.datetime_unstructure_hook # type: ignore 

120)