Coverage for tallipoika/api.py: 68.57%

1"""JSON Canonicalization Scheme (JCS) serializer API.""" 

2 

3import math 

4from typing import Any, no_type_check 

5 

6from tallipoika._factory import make_iterencode as _make_iterencode 

7from tallipoika.speedup import accelerated_make_encoder, encode_basestring, encode_basestring_ascii 

8 

9COLON = ':' 

10COMMA = ',' 

11SPACE = ' ' 

12 

13INFINITY = float('inf') 

14REPR_NAN = 'NaN' 

15REPR_POS_INF = 'Infinity' 

16REPR_NEG_INF = '-Infinity' 

17 

18 

19@no_type_check 

20class JSONEncoder: 

21 """JSON encoder for typical Python data structures. 

22 

23 Encodes the following nine Python types as the following 7 JSON types: 

24 

25 - Python `dict` -> JSON object 

26 - Python `list` and `tuple` -> JSON array 

27 - Python `str` -> JSON string 

28 - Python `float` and `int` -> JSON number 

29 - Python `True` ->JSON true 

30 - Python `False` -> JSON false 

31 - Python `None` -> JSON null 

32 

33 To extend recognition to other objects, subclass and implement a `default` method that returns a serializable 

34 object for `obj` if possible, otherwise it should call the superclass implementation (to raise `TypeError`). 

35 """ 

36 

37 item_separator = f'{COMMA}{SPACE}' 

38 key_separator = f'{COLON}{SPACE}' 

39 

40 @no_type_check 

41 def __init__( 

42 self, 

43 *, 

44 skipkeys=False, 

45 ensure_ascii=False, 

46 check_circular=True, 

47 allow_nan=True, 

48 sort_keys=True, 

49 indent=None, 

50 separators=(COMMA, COLON), 

51 default=None, 

52 ): 

53 """Constructor for JSONEncoder, with sensible defaults for JCS. 

54 

55 All parameter defaults except the value for `sort_keys` are as per the standard Python implementation; 

56 for documentation cf. https://docs.python.org/3/library/json.html#json.JSONEncoder. 

57 

58 The default value for the `sort_keys` parameter is `True`, so the output of dictionaries will be sorted by key. 

59 """ 

60 self.skipkeys = skipkeys 

61 self.ensure_ascii = ensure_ascii 

62 self.check_circular = check_circular 

63 self.allow_nan = allow_nan 

64 self.sort_keys = sort_keys 

65 self.indent = indent 

66 if separators is not None: 66 ↛ 68line 66 didn't jump to line 68, because the condition on line 66 was never false

67 self.item_separator, self.key_separator = separators 

68 elif indent is not None: 

69 self.item_separator = COMMA 

70 if default is not None: 70 ↛ 71line 70 didn't jump to line 71, because the condition on line 70 was never true

71 self.default = default 

72 

73 @no_type_check 

74 def default(self, obj): 

75 """Implement this method in a subclass such that it returns a serializable object for obj. 

76 

77 For an example, cf. https://docs.python.org/3/library/json.html#json.JSONEncoder.default 

78 """ 

79 raise TypeError(f"Object of type '{obj.__class__.__name__}' is not JSON serializable") 

80 

81 @no_type_check 

82 def encode(self, obj): 

83 """Return a JSON string representation of a Python data structure.""" 

84 if isinstance(obj, str): # This is for extremely simple cases and benchmarks. 84 ↛ 85line 84 didn't jump to line 85, because the condition on line 84 was never true

85 return encode_basestring_ascii(obj) if self.ensure_ascii else encode_basestring(obj) 

86 # This doesn't pass the iterator directly to ''.join() because the exceptions aren't as detailed. 

87 # The list call should be roughly equivalent to the PySequence_Fast that ''.join() would do. 

88 chunks = self.iterencode(obj, _one_shot=False) 

89 if not isinstance(chunks, (list, tuple)): 89 ↛ 91line 89 didn't jump to line 91, because the condition on line 89 was never false

90 chunks = list(chunks) 

91 return ''.join(chunks) 

92 

93 @no_type_check 

94 def iterencode(self, obj, _one_shot=False): 

95 """Encode the given object and yield each string representation as available. 

96 

97 For example: 

98 

99 for chunk in JSONEncoder().iterencode(big_object): 

100 mysocket.write(chunk) 

101 

102 """ 

103 markers = {} if self.check_circular else None 

104 _encoder = encode_basestring_ascii if self.ensure_ascii else encode_basestring 

105 

106 def floatstr(obj: Any, allow_nan=self.allow_nan, _repr=float.__repr__, _inf=INFINITY, _neginf=-INFINITY): 

107 """Check for specials. 

108 

109 Note that this type of test is processor and platform-specific, so tests shall not rely on the internals. 

110 """ 

111 if math.isfinite(obj): 

112 return _repr(obj) # noqa 

113 elif not allow_nan: 

114 raise ValueError(f'out of range float value {repr(obj)} is not JSON compliant') 

115 

116 if math.isnan(obj): 

117 return REPR_NAN 

118 if obj == _inf: 

119 return REPR_POS_INF 

120 if obj == _neginf: 

121 return REPR_NEG_INF 

122 

123 if _one_shot and accelerated_make_encoder is not None and self.indent is None: 123 ↛ 124line 123 didn't jump to line 124, because the condition on line 123 was never true

124 return accelerated_make_encoder( 

125 markers, 

126 self.default, 

127 _encoder, 

128 self.indent, 

129 self.key_separator, 

130 self.item_separator, 

131 self.sort_keys, 

132 self.skipkeys, 

133 self.allow_nan, 

134 )(obj, 0) 

135 

136 return _make_iterencode( 

137 markers, 

138 self.default, 

139 _encoder, 

140 self.indent, 

141 floatstr, 

142 self.key_separator, 

143 self.item_separator, 

144 self.sort_keys, 

145 self.skipkeys, 

146 _one_shot, 

147 )(obj, 0) 

148 

149 

150@no_type_check 

151def ensure_encoding(text: str, utf8: bool) -> str: 

152 return text.encode() if utf8 else text 

153 

154 

155@no_type_check 

156def canonicalize(obj, utf8=True): 

157 return ensure_encoding(JSONEncoder(sort_keys=True).encode(obj), utf8=utf8) 

158 

159 

160@no_type_check 

161def serialize(obj, utf8=True): 

162 return ensure_encoding(JSONEncoder(sort_keys=False).encode(obj), utf8=utf8)