Coverage for polar/exceptions.py: 93%

1from collections.abc import Sequence 

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

3 

4from pydantic import BaseModel, Field, create_model 

5from pydantic_core import ErrorDetails, InitErrorDetails, PydanticCustomError 

6from pydantic_core import ValidationError as PydanticValidationError 

7 

8from polar.config import settings 

9 

10 

11class PolarError(Exception): 

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 

25 

26 def __init__( 

27 self, 

28 message: str, 

29 status_code: int = 500, 

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

31 ) -> None: 

32 super().__init__(message) 

33 self.message = message 

34 self.status_code = status_code 

35 self.headers = headers 

36 

37 @classmethod 

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

39 if cls._schema is not None: 

40 return cls._schema 

41 

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

43 

44 model = create_model( 

45 cls.__name__, 

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

47 detail=(str, ...), 

48 ) 

49 cls._schema = model 

50 return cls._schema 

51 

52 

53class PolarTaskError(PolarError): 

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: 

62 super().__init__(message) 

63 

64 

65class PolarRedirectionError(PolarError): 

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__( 

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): 

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

85 super().__init__(message, status_code) 

86 

87 

88class NotPermitted(PolarError): 

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

90 super().__init__(message, status_code) 

91 

92 

93class Unauthorized(PolarError): 

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

95 super().__init__( 

96 message, 

97 status_code, 

98 headers={ 

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

100 }, 

101 ) 

102 

103 

104class InternalServerError(PolarError): 

105 def __init__( 

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

107 ) -> None: 

108 super().__init__(message, status_code) 

109 

110 

111class ResourceNotFound(PolarError): 

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

113 super().__init__(message, status_code) 

114 

115 

116class ResourceNotModified(Exception): 

117 # Handled separately to avoid any content being returned 

118 """304 Not Modified.""" 

119 

120 def __init__(self) -> None: 

121 self.status_code = 304 

122 

123 

124class ResourceUnavailable(PolarError): 

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

126 super().__init__(message, status_code) 

127 

128 

129class ResourceAlreadyExists(PolarError): 

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

131 super().__init__(message, status_code) 

132 

133 

134class PaymentNotReady(PolarError): 

135 def __init__( 

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): 

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

145 msg: LiteralString 

146 type: LiteralString 

147 input: Any 

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

149 url: NotRequired[str] 

150 

151 

152class PolarRequestValidationError(PolarError): 

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

154 self._errors = errors 

155 

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

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()