Coverage for src/prisma/errors.py: 92%

1from __future__ import annotations 

2 

3from typing import Any, Optional 

4 

5__all__ = ( 

6 'PrismaError', 

7 'DataError', 

8 'UniqueViolationError', 

9 'ForeignKeyViolationError', 

10 'MissingRequiredValueError', 

11 'RawQueryError', 

12 'TableNotFoundError', 

13 'RecordNotFoundError', 

14 'HTTPClientClosedError', 

15 'ClientNotConnectedError', 

16 'PrismaWarning', 

17 'UnsupportedSubclassWarning', 

18) 

19 

20 

21class PrismaError(Exception): 

22 pass 

23 

24 

25class ClientNotRegisteredError(PrismaError): 

26 def __init__(self) -> None: 

27 super().__init__('No client instance registered; You must call prisma.register(prisma.Prisma())') 

28 

29 

30class ClientAlreadyRegisteredError(PrismaError): 

31 def __init__(self) -> None: 

32 super().__init__('A client has already been registered.') 

33 

34 

35class ClientNotConnectedError(PrismaError): 

36 def __init__(self) -> None: 

37 super().__init__( 

38 'Client is not connected to the query engine, ' 'you must call `connect()` before attempting to query data.' 

39 ) 

40 

41 

42class HTTPClientClosedError(PrismaError): 

43 def __init__(self) -> None: 

44 super().__init__('Cannot make a request from a closed client.') 

45 

46 

47class UnsupportedDatabaseError(PrismaError): 

48 context: str 

49 database: str 

50 

51 def __init__(self, database: str, context: str) -> None: 

52 super().__init__(f'{context} is not supported by {database}') 

53 self.database = database 

54 self.context = context 

55 

56 

57class DataError(PrismaError): 

58 data: Any 

59 code: Any 

60 meta: Any 

61 

62 def __init__(self, data: Any, *, message: Optional[str] = None) -> None: 

63 self.data = data 

64 

65 user_facing_error = data.get('user_facing_error', {}) 

66 self.code = user_facing_error.get('error_code') 

67 self.meta = user_facing_error.get('meta') 

68 

69 message = message or user_facing_error.get('message') 

70 super().__init__(message or 'An error occurred while processing data.') 

71 

72 

73class UniqueViolationError(DataError): 

74 pass 

75 

76 

77class ForeignKeyViolationError(DataError): 

78 pass 

79 

80 

81class MissingRequiredValueError(DataError): 

82 pass 

83 

84 

85class RawQueryError(DataError): 

86 def __init__(self, data: Any) -> None: 

87 try: 

88 super().__init__(data, message=data['user_facing_error']['meta']['message']) 

89 except KeyError: 

90 super().__init__(data) 

91 

92 

93class TableNotFoundError(DataError): 

94 def __init__(self, data: Any) -> None: 

95 super().__init__(data) 

96 self.table: Optional[str] = self.meta.get('table') 

97 

98 

99class FieldNotFoundError(DataError): 

100 # currently we cannot easily resolve the erroneous field as Prisma 

101 # returns different results for unknown fields in different situations 

102 # e.g. root query, nested query and mutation queries 

103 def __init__(self, data: Any, *, message: str | None = None) -> None: 

104 if message is None: 104 ↛ 117line 104 didn't jump to line 117, because the condition on line 104 was never false

105 meta = data.get('user_facing_error', {}).get('meta', {}) 

106 if meta.get('kind') == 'Union': 

107 error = _pick_union_error(meta.get('errors', [])) 

108 else: 

109 error = meta 

110 

111 argument_path = error.get('argumentPath') 

112 selection_path = error.get('selectionPath') 

113 

114 if argument_path: 

115 message = f'Could not find field at `{".".join(selection_path)}.{".".join(argument_path)}`' 

116 

117 super().__init__(data, message=message) 

118 

119 

120class RecordNotFoundError(DataError): 

121 pass 

122 

123 

124class InputError(DataError): 

125 pass 

126 

127 

128class TransactionError(PrismaError): 

129 pass 

130 

131 

132class TransactionExpiredError(TransactionError): 

133 pass 

134 

135 

136class TransactionNotStartedError(TransactionError): 

137 def __init__(self) -> None: 

138 super().__init__( 

139 'Transaction has not been started yet.\n' 

140 'Transactions must be used within a context manager or started manually.' 

141 ) 

142 

143 

144class BuilderError(PrismaError): 

145 pass 

146 

147 

148class InvalidModelError(BuilderError): 

149 def __init__(self, model: type) -> None: 

150 super().__init__(f'Expected the {model} type to have a `__prisma_model__` class variable set') 

151 

152 

153class UnknownModelError(BuilderError): 

154 def __init__(self, model: str) -> None: 

155 super().__init__(f'Model: "{model}" does not exist.') 

156 

157 

158class UnknownRelationalFieldError(BuilderError): 

159 def __init__(self, model: str, field: str) -> None: 

160 super().__init__(f'Field: "{field}" either does not exist or is not a relational field on the {model} model') 

161 

162 

163class GeneratorError(PrismaError): 

164 pass 

165 

166 

167class UnsupportedListTypeError(GeneratorError): 

168 type: str 

169 

170 def __init__(self, typ: str) -> None: 

171 super().__init__( 

172 f'Cannot use {typ} as a list yet; Please create a ' 

173 'feature request at https://github.com/RobertCraigie/prisma-client-py/issues/new' 

174 ) 

175 self.type = typ 

176 

177 

178class PrismaWarning(Warning): 

179 pass 

180 

181 

182# Note: this is currently unused but not worth removing 

183class UnsupportedSubclassWarning(PrismaWarning): 

184 pass 

185 

186 

187# TODO: proper types 

188def _pick_union_error(errors: list[Any]) -> Any: 

189 # Note: uses the same heuristic as the TS client 

190 return max( 

191 errors, 

192 key=lambda e: (len(e.get('argumentPath', [])) + len(e.get('selectionPath'))), 

193 )