Coverage for notion_client/errors.py: 100%

1"""Custom exceptions for notion-sdk-py. 

2 

3This module defines the exceptions that can be raised when an error occurs. 

4""" 

5 

6import asyncio 

7import json 

8from enum import Enum 

9from typing import Any, Dict, Optional, Union, Set 

10import sys 

11import httpx 

12from urllib.parse import unquote 

13 

14if sys.version_info >= (3, 10): 

15 from typing import TypeGuard 

16else: 

17 from typing_extensions import TypeGuard # pragma: no cover 

18 

19 

20class APIErrorCode(str, Enum): 

21 """Error codes returned in responses from the API.""" 

22 

23 Unauthorized = "unauthorized" 

24 """The bearer token is not valid.""" 

25 

26 RestrictedResource = "restricted_resource" 

27 """Given the bearer token used, the client doesn't have permission to 

28 perform this operation.""" 

29 

30 ObjectNotFound = "object_not_found" 

31 """Given the bearer token used, the resource does not exist. 

32 This error can also indicate that the resource has not been shared with owner 

33 of the bearer token.""" 

34 

35 RateLimited = "rate_limited" 

36 """This request exceeds the number of requests allowed. Slow down and try again.""" 

37 

38 InvalidJSON = "invalid_json" 

39 """The request body could not be decoded as JSON.""" 

40 

41 InvalidRequestURL = "invalid_request_url" 

42 """The request URL is not valid.""" 

43 

44 InvalidRequest = "invalid_request" 

45 """This request is not supported.""" 

46 

47 InvalidBeta = "invalid_beta" 

48 """The `Notion-Beta` header is malformed, unknown, duplicated, or references a 

49 stale revision.""" 

50 

51 ValidationError = "validation_error" 

52 """The request body does not match the schema for the expected parameters.""" 

53 

54 ConflictError = "conflict_error" 

55 """The transaction could not be completed, potentially due to a data collision. 

56 Make sure the parameters are up to date and try again.""" 

57 

58 InternalServerError = "internal_server_error" 

59 """An unexpected error occurred. Reach out to Notion support.""" 

60 

61 ServiceOverload = "service_overload" 

62 """Notion is overloaded. Slow down and try again later.""" 

63 

64 ServiceUnavailable = "service_unavailable" 

65 """Notion is unavailable. Try again later. 

66 This can occur when the time to respond to a request takes longer than 60 seconds, 

67 the maximum request timeout.""" 

68 

69 GatewayTimeout = "gateway_timeout" 

70 """The request timed out on the server side.""" 

71 

72 

73class ClientErrorCode(str, Enum): 

74 """Error codes generated for client errors.""" 

75 

76 RequestTimeout = "notionhq_client_request_timeout" 

77 ResponseError = "notionhq_client_response_error" 

78 InvalidPathParameter = "notionhq_client_invalid_path_parameter" 

79 

80 

81# Error codes on errors thrown by the `Client`. 

82NotionErrorCode = Union[APIErrorCode, ClientErrorCode] 

83 

84 

85class NotionClientErrorBase(Exception): 

86 """Base error type for all Notion client errors.""" 

87 

88 code: Union[str, NotionErrorCode] 

89 

90 def __init__(self, message: str = "") -> None: 

91 super().__init__(message) 

92 

93 

94def is_notion_client_error(error: Any) -> TypeGuard["NotionClientError"]: 

95 return isinstance(error, NotionClientErrorBase) 

96 

97 

98def _is_notion_client_error_with_code(error: Any, codes: Set[str]) -> bool: 

99 if not is_notion_client_error(error): 

100 return False 

101 error_code = error.code 

102 if isinstance(error_code, Enum): 

103 error_code = error_code.value 

104 

105 return error_code in codes 

106 

107 

108class RequestTimeoutError(NotionClientErrorBase): 

109 """Error thrown by the client if a request times out.""" 

110 

111 code: ClientErrorCode = ClientErrorCode.RequestTimeout 

112 

113 def __init__(self, message: str = "Request to Notion API has timed out") -> None: 

114 super().__init__(message) 

115 

116 @staticmethod 

117 def is_request_timeout_error(error: Any) -> bool: 

118 return _is_notion_client_error_with_code( 

119 error, 

120 {ClientErrorCode.RequestTimeout.value}, 

121 ) 

122 

123 @staticmethod 

124 async def reject_after_timeout(coro: Any, timeout_ms: int) -> Any: 

125 try: 

126 return await asyncio.wait_for(coro, timeout=timeout_ms / 1000.0) 

127 except asyncio.TimeoutError: 

128 raise RequestTimeoutError() 

129 

130 

131class InvalidPathParameterError(NotionClientErrorBase): 

132 """Error thrown when a path parameter contains invalid characters such as 

133 path traversal sequences (..) that could alter the intended API endpoint.""" 

134 

135 code: ClientErrorCode = ClientErrorCode.InvalidPathParameter 

136 

137 def __init__( 

138 self, 

139 message: str = ( 

140 "Path parameter contains invalid characters that could alter the request path" 

141 ), 

142 ) -> None: 

143 super().__init__(message) 

144 

145 @staticmethod 

146 def is_invalid_path_parameter_error(error: object) -> bool: 

147 return _is_notion_client_error_with_code( 

148 error, 

149 {ClientErrorCode.InvalidPathParameter.value}, 

150 ) 

151 

152 

153def validate_request_path(path: str) -> None: 

154 """Validates that a request path does not contain path traversal sequences. 

155 Raises InvalidPathParameterError if the path contains ".." segments, 

156 including URL-encoded variants like %2e%2e.""" 

157 

158 # Check for literal path traversal 

159 if ".." in path: 

160 raise InvalidPathParameterError( 

161 f'Request path "{path}" contains path traversal sequence ".."' 

162 ) 

163 

164 # Check for URL-encoded path traversal (%2e = '.') 

165 # Only decode if path contains potential encoded dots 

166 if "%2e" in path.lower(): 

167 decoded = unquote(path) 

168 if ".." in decoded: 

169 raise InvalidPathParameterError( 

170 f'Request path "{path}" contains encoded path traversal sequence' 

171 ) 

172 

173 

174HTTPResponseErrorCode = Union[ClientErrorCode, APIErrorCode] 

175 

176 

177def _get_response_header(headers: httpx.Headers, name: str) -> Optional[str]: 

178 """Reads a response header by name, or returns None when it is absent. 

179 

180 Header names are case-insensitive per RFC 9110, which `httpx.Headers` 

181 already honors, so this only narrows the value to an optional string. 

182 """ 

183 value = headers.get(name) 

184 return value if isinstance(value, str) else None 

185 

186 

187class HTTPResponseError(NotionClientErrorBase): 

188 code: Union[str, APIErrorCode] 

189 status: int 

190 headers: httpx.Headers 

191 body: str 

192 additional_data: Optional[Dict[str, Any]] 

193 request_id: Optional[str] 

194 ray_id: Optional[str] # Value of the `cf-ray` response header, when present. 

195 

196 def __init__( 

197 self, 

198 code: Union[str, APIErrorCode], 

199 status: int, 

200 message: str, 

201 headers: httpx.Headers, 

202 raw_body_text: str, 

203 additional_data: Optional[Dict[str, Any]] = None, 

204 request_id: Optional[str] = None, 

205 ) -> None: 

206 super().__init__(message) 

207 self.code = code 

208 self.status = status 

209 self.headers = headers 

210 self.body = raw_body_text 

211 self.additional_data = additional_data 

212 self.request_id = request_id or _get_response_header( 

213 self.headers, "x-notion-request-id" 

214 ) 

215 self.ray_id = _get_response_header(self.headers, "cf-ray") 

216 

217 

218_http_response_error_codes: Set[str] = { 

219 ClientErrorCode.ResponseError.value, 

220 APIErrorCode.Unauthorized.value, 

221 APIErrorCode.RestrictedResource.value, 

222 APIErrorCode.ObjectNotFound.value, 

223 APIErrorCode.RateLimited.value, 

224 APIErrorCode.InvalidJSON.value, 

225 APIErrorCode.InvalidRequestURL.value, 

226 APIErrorCode.InvalidRequest.value, 

227 APIErrorCode.InvalidBeta.value, 

228 APIErrorCode.ValidationError.value, 

229 APIErrorCode.ConflictError.value, 

230 APIErrorCode.InternalServerError.value, 

231 APIErrorCode.ServiceOverload.value, 

232 APIErrorCode.ServiceUnavailable.value, 

233 APIErrorCode.GatewayTimeout.value, 

234} 

235 

236 

237def is_http_response_error(error: Any) -> TypeGuard[HTTPResponseError]: 

238 return _is_notion_client_error_with_code(error, _http_response_error_codes) 

239 

240 

241class UnknownHTTPResponseError(HTTPResponseError): 

242 """Error thrown if an API call responds with an unknown error code, or does not respond with 

243 a properly-formatted error. 

244 """ 

245 

246 def __init__( 

247 self, 

248 status: int, 

249 message: Optional[str] = None, 

250 headers: Optional[httpx.Headers] = None, 

251 raw_body_text: str = "", 

252 ) -> None: 

253 if headers is None: 

254 headers = httpx.Headers() 

255 if message is None: 

256 message = _build_unknown_response_message(status, headers) 

257 

258 super().__init__( 

259 code=ClientErrorCode.ResponseError.value, 

260 status=status, 

261 message=message, 

262 headers=headers, 

263 raw_body_text=raw_body_text, 

264 additional_data=None, 

265 ) 

266 

267 @staticmethod 

268 def is_unknown_http_response_error(error: Any) -> bool: 

269 return _is_notion_client_error_with_code( 

270 error, 

271 {ClientErrorCode.ResponseError.value}, 

272 ) 

273 

274 

275_api_error_codes: Set[str] = { 

276 APIErrorCode.Unauthorized.value, 

277 APIErrorCode.RestrictedResource.value, 

278 APIErrorCode.ObjectNotFound.value, 

279 APIErrorCode.RateLimited.value, 

280 APIErrorCode.InvalidJSON.value, 

281 APIErrorCode.InvalidRequestURL.value, 

282 APIErrorCode.InvalidRequest.value, 

283 APIErrorCode.InvalidBeta.value, 

284 APIErrorCode.ValidationError.value, 

285 APIErrorCode.ConflictError.value, 

286 APIErrorCode.InternalServerError.value, 

287 APIErrorCode.ServiceOverload.value, 

288 APIErrorCode.ServiceUnavailable.value, 

289 APIErrorCode.GatewayTimeout.value, 

290} 

291 

292 

293class APIResponseError(HTTPResponseError): 

294 code: APIErrorCode 

295 request_id: Optional[str] 

296 

297 @staticmethod 

298 def is_api_response_error(error: Any) -> TypeGuard["APIResponseError"]: 

299 return _is_notion_client_error_with_code(error, _api_error_codes) 

300 

301 

302# Type alias for all Notion client errors 

303NotionClientError = Union[ 

304 RequestTimeoutError, 

305 UnknownHTTPResponseError, 

306 APIResponseError, 

307 InvalidPathParameterError, 

308] 

309 

310 

311def build_request_error( 

312 response: httpx.Response, 

313 body_text: str, 

314) -> Union[APIResponseError, UnknownHTTPResponseError]: 

315 api_error_response_body = _parse_api_error_response_body(body_text) 

316 

317 if api_error_response_body is not None: 

318 return APIResponseError( 

319 code=api_error_response_body["code"], 

320 message=api_error_response_body["message"], 

321 headers=response.headers, 

322 status=response.status_code, 

323 raw_body_text=body_text, 

324 additional_data=api_error_response_body.get("additional_data"), 

325 request_id=api_error_response_body.get("request_id"), 

326 ) 

327 

328 return UnknownHTTPResponseError( 

329 message=None, 

330 headers=response.headers, 

331 status=response.status_code, 

332 raw_body_text=body_text, 

333 ) 

334 

335 

336def _build_unknown_response_message(status: int, headers: httpx.Headers) -> str: 

337 """Builds the default message for an unrecognized HTTP response. 

338 

339 A response with a Cloudflare Ray ID but no Notion request ID was answered 

340 before it reached the API. The message surfaces the Ray ID rather than 

341 leaving callers to infer it from an HTML body. 

342 """ 

343 base = f"Request to Notion API failed with status: {status}" 

344 ray_id = _get_response_header(headers, "cf-ray") 

345 request_id = _get_response_header(headers, "x-notion-request-id") 

346 answered_by_edge = ray_id is not None and request_id is None 

347 if not answered_by_edge: 

348 return base 

349 

350 content_type = _get_response_header(headers, "content-type") 

351 content_type_note = ( 

352 f" (content-type: {content_type})" if content_type is not None else "" 

353 ) 

354 blocked_request_note = ( 

355 " This may mean the request was blocked by a network security rule." 

356 if status == 403 

357 else "" 

358 ) 

359 return ( 

360 f"{base}. The response was returned by Notion's edge proxy" 

361 f" before reaching the Notion API{content_type_note}." 

362 f"{blocked_request_note}" 

363 f" Cloudflare Ray ID: {ray_id}." 

364 " Include this ID when contacting Notion support." 

365 ) 

366 

367 

368def _parse_api_error_response_body(body: str) -> Optional[Dict[str, Any]]: 

369 if not isinstance(body, str): 

370 return None 

371 

372 try: 

373 parsed = json.loads(body) 

374 except (json.JSONDecodeError, ValueError): 

375 return None 

376 

377 if not isinstance(parsed, dict): 

378 return None 

379 

380 message = parsed.get("message") 

381 code = parsed.get("code") 

382 

383 if not isinstance(message, str) or not _is_api_error_code(code): 

384 return None 

385 

386 result: Dict[str, Any] = { 

387 "code": APIErrorCode(code), 

388 "message": message, 

389 } 

390 

391 if "additional_data" in parsed: 

392 result["additional_data"] = parsed["additional_data"] 

393 

394 if "request_id" in parsed: 

395 result["request_id"] = parsed["request_id"] 

396 

397 return result 

398 

399 

400def _is_api_error_code(code: Any) -> bool: 

401 return isinstance(code, str) and code in _api_error_codes