Coverage for src/prisma/_raw_query.py: 100%

1from __future__ import annotations 

2 

3import json 

4from typing import ( 

5 Any, 

6 Callable, 

7 overload, 

8) 

9 

10from ._types import BaseModelT 

11from ._compat import model_parse 

12 

13# From: https://github.com/prisma/prisma/blob/main/packages/client/src/runtime/utils/deserializeRawResults.ts 

14# Last checked: 2022-12-04 

15""" 

16type PrismaType = 

17 | 'int' 

18 | 'bigint' 

19 | 'float' 

20 | 'double' 

21 | 'string' 

22 | 'enum' 

23 | 'bytes' 

24 | 'bool' 

25 | 'char' 

26 | 'decimal' 

27 | 'json' 

28 | 'xml' 

29 | 'uuid' 

30 | 'datetime' 

31 | 'date' 

32 | 'time' 

33 | 'array' 

34 | 'null' 

35""" 

36 

37 

38@overload 

39def deserialize_raw_results(raw_list: list[dict[str, Any]]) -> list[dict[str, Any]]: ... 

40 

41 

42@overload 

43def deserialize_raw_results( 

44 raw_list: list[dict[str, object]], 

45 model: type[BaseModelT], 

46) -> list[BaseModelT]: ... 

47 

48 

49def deserialize_raw_results( 

50 raw_list: list[dict[str, Any]], 

51 model: type[BaseModelT] | None = None, 

52) -> list[BaseModelT] | list[dict[str, Any]]: 

53 """Deserialize a list of raw query results into their rich Python types. 

54 

55 If `model` is given, convert each result into the corresponding model. 

56 Otherwise results are returned as a dictionary 

57 """ 

58 if model is not None: 

59 return [_deserialize_prisma_object(obj, model=model, for_model=True) for obj in raw_list] 

60 

61 return [_deserialize_prisma_object(obj, for_model=False) for obj in raw_list] 

62 

63 

64# NOTE: this very weird `for_model` API is simply here as a workaround for 

65# https://github.com/RobertCraigie/prisma-client-py/issues/638 

66# 

67# This should hopefully be removed soon. 

68 

69 

70@overload 

71def _deserialize_prisma_object( 

72 raw_obj: dict[str, Any], 

73 *, 

74 for_model: bool, 

75) -> dict[str, Any]: ... 

76 

77 

78@overload 

79def _deserialize_prisma_object( 

80 raw_obj: dict[str, object], 

81 *, 

82 for_model: bool, 

83 model: type[BaseModelT], 

84) -> BaseModelT: ... 

85 

86 

87def _deserialize_prisma_object( 

88 raw_obj: dict[Any, Any], 

89 *, 

90 for_model: bool, 

91 model: type[BaseModelT] | None = None, 

92) -> BaseModelT | dict[str, Any]: 

93 # create a local reference to avoid performance penalty of global 

94 # lookups on some python versions 

95 _deserializers = DESERIALIZERS 

96 

97 new_obj = {} 

98 for key, raw_value in raw_obj.items(): 

99 value = raw_value['prisma__value'] 

100 prisma_type = raw_value['prisma__type'] 

101 

102 new_obj[key] = _deserializers[prisma_type](value, for_model) if prisma_type in _deserializers else value 

103 

104 if model is not None: 

105 return model_parse(model, new_obj) 

106 

107 return new_obj 

108 

109 

110def _deserialize_bigint(value: str, _for_model: bool) -> int: 

111 return int(value) 

112 

113 

114def _deserialize_decimal(value: str, _for_model: bool) -> float: 

115 return float(value) 

116 

117 

118def _deserialize_array(value: list[Any], for_model: bool) -> list[Any]: 

119 # create a local reference to avoid performance penalty of global 

120 # lookups on some python versions 

121 _deserializers = DESERIALIZERS 

122 

123 arr = [] 

124 for entry in value: 

125 prisma_type = entry['prisma__type'] 

126 prisma_value = entry['prisma__value'] 

127 arr.append( 

128 (_deserializers[prisma_type](prisma_value, for_model) if prisma_type in _deserializers else prisma_value) 

129 ) 

130 

131 return arr 

132 

133 

134def _deserialize_json(value: object, for_model: bool) -> object: 

135 # TODO: this may break if someone inserts just a string into the database 

136 if not isinstance(value, str) and for_model: 

137 # TODO: this is very bad 

138 # 

139 # Pydantic expects Json fields to be a `str`, we should implement 

140 # an actual workaround for this validation instead of wasting compute 

141 # on re-serializing the data. 

142 return json.dumps(value) 

143 

144 # This may or may not have already been deserialized by the database 

145 return value 

146 

147 

148DESERIALIZERS: dict[str, Callable[[Any, bool], object]] = { 

149 'bigint': _deserialize_bigint, 

150 'decimal': _deserialize_decimal, 

151 'array': _deserialize_array, 

152 'json': _deserialize_json, 

153}