Coverage for polar/exceptions.py: 76%

1from collections.abc import Sequence 2 ctx1ba

2from typing import Any, ClassVar, Literal, LiteralString, NotRequired, TypedDict 2 ctx1ba

3 

4from pydantic import BaseModel, Field, create_model 2 ctx1ba

5from pydantic_core import ErrorDetails, InitErrorDetails, PydanticCustomError 2 ctx1ba

6from pydantic_core import ValidationError as PydanticValidationError 2 ctx1ba

7 

8from polar.config import settings 2 ctx1ba

9 

10 

11class PolarError(Exception): 2 ctx1ba

12 """ 

13 Base exception class for all errors raised by Polar. 

14 

15 A custom exception handler for FastAPI takes care 

16 of catching and returning a proper HTTP error from them. 

17 

18 Args: 

19 message: The error message that'll be displayed to the user. 

20 status_code: The status code of the HTTP response. Defaults to 500. 

21 headers: Additional headers to be included in the response. 

22 """ 

23 

24 _schema: ClassVar[type[BaseModel] | None] = None 2 ctx1ba

25 

26 def __init__( 2 ctx1ba

27 self, 

28 message: str, 

29 status_code: int = 500, 

30 headers: dict[str, str] | None = None, 

31 ) -> None: 

32 super().__init__(message) 1 ctx1c

33 self.message = message 1 ctx1c

34 self.status_code = status_code 1 ctx1c

35 self.headers = headers 1 ctx1c

36 

37 @classmethod 2 ctx1ba

38 def schema(cls) -> type[BaseModel]: 2 ctx1ba

39 if cls._schema is not None: 1 ctx1a

40 return cls._schema 1 ctx1a

41 

42 error_literal = Literal[cls.__name__] # type: ignore 1 ctx1a

43 

44 model = create_model( 1 ctx1a

45 cls.__name__, 

46 error=(error_literal, Field(examples=[cls.__name__])), 

47 detail=(str, ...), 

48 ) 

49 cls._schema = model 1 ctx1a

50 return cls._schema 1 ctx1a

51 

52 

53class PolarTaskError(PolarError): 2 ctx1ba

54 """ 

55 Base exception class for errors raised by tasks. 

56 

57 Args: 

58 message: The error message. 

59 """ 

60 

61 def __init__(self, message: str) -> None: 2 ctx1ba

62 super().__init__(message) 

63 

64 

65class PolarRedirectionError(PolarError): 2 ctx1ba

66 """ 

67 Exception class for errors 

68 that should be displayed nicely to the user through our UI. 

69 

70 A specific exception handler will redirect to `/error` page in the client app. 

71 

72 Args: 

73 return_to: Target URL of the *Go back* button on the error page. 

74 """ 

75 

76 def __init__( 2 ctx1ba

77 self, message: str, status_code: int = 400, return_to: str | None = None 

78 ) -> None: 

79 self.return_to = return_to 

80 super().__init__(message, status_code) 

81 

82 

83class BadRequest(PolarError): 2 ctx1ba

84 def __init__(self, message: str = "Bad request", status_code: int = 400) -> None: 2 ctx1ba

85 super().__init__(message, status_code) 

86 

87 

88class NotPermitted(PolarError): 2 ctx1ba

89 def __init__(self, message: str = "Not permitted", status_code: int = 403) -> None: 2 ctx1ba

90 super().__init__(message, status_code) 

91 

92 

93class Unauthorized(PolarError): 2 ctx1ba

94 def __init__(self, message: str = "Unauthorized", status_code: int = 401) -> None: 2 ctx1ba

95 super().__init__( 1 ctx1c

96 message, 

97 status_code, 

98 headers={ 

99 "WWW-Authenticate": f'Bearer realm="{settings.WWW_AUTHENTICATE_REALM}"' 

100 }, 

101 ) 

102 

103 

104class InternalServerError(PolarError): 2 ctx1ba

105 def __init__( 2 ctx1ba

106 self, message: str = "Internal Server Error", status_code: int = 500 

107 ) -> None: 

108 super().__init__(message, status_code) 

109 

110 

111class ResourceNotFound(PolarError): 2 ctx1ba

112 def __init__(self, message: str = "Not found", status_code: int = 404) -> None: 2 ctx1ba

113 super().__init__(message, status_code) 1 ctx1c

114 

115 

116class ResourceNotModified(Exception): 2 ctx1ba

117 # Handled separately to avoid any content being returned 

118 """304 Not Modified.""" 

119 

120 def __init__(self) -> None: 2 ctx1ba

121 self.status_code = 304 

122 

123 

124class ResourceUnavailable(PolarError): 2 ctx1ba

125 def __init__(self, message: str = "Unavailable", status_code: int = 410) -> None: 2 ctx1ba

126 super().__init__(message, status_code) 

127 

128 

129class ResourceAlreadyExists(PolarError): 2 ctx1ba

130 def __init__(self, message: str = "Already exists", status_code: int = 409) -> None: 2 ctx1ba

131 super().__init__(message, status_code) 

132 

133 

134class PaymentNotReady(PolarError): 2 ctx1ba

135 def __init__( 2 ctx1ba

136 self, 

137 message: str = "Organization is not ready to accept payments", 

138 status_code: int = 403, 

139 ) -> None: 

140 super().__init__(message, status_code) 

141 

142 

143class ValidationError(TypedDict): 2 ctx1ba

144 loc: tuple[int | str, ...] 2 ctx1ba

145 msg: LiteralString 2 ctx1ba

146 type: LiteralString 2 ctx1ba

147 input: Any 2 ctx1ba

148 ctx: NotRequired[dict[str, Any]] 2 ctx1ba

149 url: NotRequired[str] 2 ctx1ba

150 

151 

152class PolarRequestValidationError(PolarError): 2 ctx1ba

153 def __init__(self, errors: Sequence[ValidationError]) -> None: 2 ctx1ba

154 self._errors = errors 

155 

156 def errors(self) -> list[ErrorDetails]: 2 ctx1ba

157 pydantic_errors: list[InitErrorDetails] = [] 

158 for error in self._errors: 

159 pydantic_errors.append( 

160 { 

161 "type": PydanticCustomError(error["type"], error["msg"]), 

162 "loc": error["loc"], 

163 "input": error["input"], 

164 } 

165 ) 

166 pydantic_error = PydanticValidationError.from_exception_data( 

167 self.__class__.__name__, pydantic_errors 

168 ) 

169 return pydantic_error.errors()