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

1from __future__ import annotations 

2 

3import json 

4from typing import Any, Callable, overload 

5from typing_extensions import Literal 

6 

7from ._types import BaseModelT 

8from ._compat import model_parse 

9 

10# from https://github.com/prisma/prisma/blob/7da6f030350931eff8574e805acb9c0de9087e8e/packages/client/src/runtime/utils/deserializeRawResults.ts 

11PrismaType = Literal[ 

12 'int', 

13 'bigint', 

14 'float', 

15 'double', 

16 'string', 

17 'enum', 

18 'bytes', 

19 'bool', 

20 'char', 

21 'decimal', 

22 'json', 

23 'xml', 

24 'uuid', 

25 'datetime', 

26 'date', 

27 'time', 

28 'int-array', 

29 'bigint-array', 

30 'float-array', 

31 'double-array', 

32 'string-array', 

33 'enum-array', 

34 'bytes-array', 

35 'bool-array', 

36 'char-array', 

37 'decimal-array', 

38 'json-array', 

39 'xml-array', 

40 'uuid-array', 

41 'datetime-array', 

42 'date-array', 

43 'time-array', 

44 'unknown-array', 

45 'unknown', 

46] 

47 

48 

49class RawQueryResult: 

50 columns: list[str] 

51 types: list[PrismaType] 

52 rows: list[list[object]] 

53 

54 def __init__( 

55 self, 

56 *, 

57 columns: list[str], 

58 types: list[PrismaType], 

59 rows: list[list[object]], 

60 ) -> None: 

61 self.columns = columns 

62 self.types = types 

63 self.rows = rows 

64 

65 

66@overload 

67def deserialize_raw_results(raw_result: dict[str, Any]) -> list[dict[str, Any]]: ... 

68 

69 

70@overload 

71def deserialize_raw_results( 

72 raw_result: dict[str, Any], 

73 model: type[BaseModelT], 

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

75 

76 

77def deserialize_raw_results( 

78 raw_result: dict[str, Any], 

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

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

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

82 

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

84 Otherwise results are returned as a dictionary 

85 """ 

86 result = RawQueryResult( 

87 columns=raw_result['columns'], 

88 types=raw_result['types'], 

89 rows=raw_result['rows'], 

90 ) 

91 if model is not None: 

92 return [_deserialize_prisma_object(obj, result=result, model=model, for_model=True) for obj in result.rows] 

93 

94 return [_deserialize_prisma_object(obj, result=result, for_model=False) for obj in result.rows] 

95 

96 

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

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

99# 

100# This should hopefully be removed soon. 

101 

102 

103@overload 

104def _deserialize_prisma_object( 

105 fields: list[object], 

106 *, 

107 result: RawQueryResult, 

108 for_model: bool, 

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

110 

111 

112@overload 

113def _deserialize_prisma_object( 

114 fields: list[object], 

115 *, 

116 result: RawQueryResult, 

117 for_model: bool, 

118 model: type[BaseModelT], 

119) -> BaseModelT: ... 

120 

121 

122def _deserialize_prisma_object( 

123 fields: list[object], 

124 *, 

125 result: RawQueryResult, 

126 for_model: bool, 

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

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

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

130 # lookups on some python versions 

131 _deserializers = DESERIALIZERS 

132 

133 new_obj: dict[str, Any] = {} 

134 for i, field in enumerate(fields): 

135 key = result.columns[i] 

136 prisma_type = result.types[i] 

137 

138 if field is None: 

139 new_obj[key] = None 

140 continue 

141 

142 if prisma_type.endswith('-array'): 

143 if not isinstance(field, list): 143 ↛ 144line 143 didn't jump to line 144, because the condition on line 143 was never true

144 raise TypeError( 

145 f'Expected array data for {key} column with internal type {prisma_type}', 

146 ) 

147 

148 item_type, _ = prisma_type.split('-') 

149 

150 new_obj[key] = [ 

151 _deserializers[item_type](value, for_model) 

152 # 

153 if item_type in _deserializers 

154 else value 

155 for value in field 

156 ] 

157 else: 

158 value = field 

159 

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

161 

162 if model is not None: 

163 return model_parse(model, new_obj) 

164 

165 return new_obj 

166 

167 

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

169 return int(value) 

170 

171 

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

173 return float(value) 

174 

175 

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

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

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

179 # TODO: this is very bad 

180 # 

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

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

183 # on re-serializing the data. 

184 return json.dumps(value) 

185 

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

187 return value 

188 

189 

190DESERIALIZERS: dict[PrismaType, Callable[[Any, bool], object]] = { 

191 'bigint': _deserialize_bigint, 

192 'decimal': _deserialize_decimal, 

193 'json': _deserialize_json, 

194}