Coverage for notion_client / errors.py: 99%
134 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-02 18:32 +0000
1"""Custom exceptions for notion-sdk-py.
3This module defines the exceptions that can be raised when an error occurs.
4"""
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
14if sys.version_info >= (3, 10):
15 from typing import TypeGuard
16else:
17 from typing_extensions import TypeGuard
20class APIErrorCode(str, Enum):
21 """Error codes returned in responses from the API."""
23 Unauthorized = "unauthorized"
24 """The bearer token is not valid."""
26 RestrictedResource = "restricted_resource"
27 """Given the bearer token used, the client doesn't have permission to
28 perform this operation."""
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."""
35 RateLimited = "rate_limited"
36 """This request exceeds the number of requests allowed. Slow down and try again."""
38 InvalidJSON = "invalid_json"
39 """The request body could not be decoded as JSON."""
41 InvalidRequestURL = "invalid_request_url"
42 """The request URL is not valid."""
44 InvalidRequest = "invalid_request"
45 """This request is not supported."""
47 ValidationError = "validation_error"
48 """The request body does not match the schema for the expected parameters."""
50 ConflictError = "conflict_error"
51 """The transaction could not be completed, potentially due to a data collision.
52 Make sure the parameters are up to date and try again."""
54 InternalServerError = "internal_server_error"
55 """An unexpected error occurred. Reach out to Notion support."""
57 ServiceUnavailable = "service_unavailable"
58 """Notion is unavailable. Try again later.
59 This can occur when the time to respond to a request takes longer than 60 seconds,
60 the maximum request timeout."""
62 GatewayTimeout = "gateway_timeout"
63 """The request timed out on the server side."""
66class ClientErrorCode(str, Enum):
67 """Error codes generated for client errors."""
69 RequestTimeout = "notionhq_client_request_timeout"
70 ResponseError = "notionhq_client_response_error"
71 InvalidPathParameter = "notionhq_client_invalid_path_parameter"
74# Error codes on errors thrown by the `Client`.
75NotionErrorCode = Union[APIErrorCode, ClientErrorCode]
78class NotionClientErrorBase(Exception):
79 """Base error type for all Notion client errors."""
81 code: Union[str, NotionErrorCode]
83 def __init__(self, message: str = "") -> None:
84 super().__init__(message)
87def is_notion_client_error(error: Any) -> TypeGuard["NotionClientError"]:
88 return isinstance(error, NotionClientErrorBase)
91def _is_notion_client_error_with_code(error: Any, codes: Set[str]) -> bool:
92 if not is_notion_client_error(error):
93 return False
94 error_code = error.code
95 if isinstance(error_code, Enum):
96 error_code = error_code.value
98 return error_code in codes
101class RequestTimeoutError(NotionClientErrorBase):
102 """Error thrown by the client if a request times out."""
104 code: ClientErrorCode = ClientErrorCode.RequestTimeout
106 def __init__(self, message: str = "Request to Notion API has timed out") -> None:
107 super().__init__(message)
109 @staticmethod
110 def is_request_timeout_error(error: Any) -> bool:
111 return _is_notion_client_error_with_code(
112 error,
113 {ClientErrorCode.RequestTimeout.value},
114 )
116 @staticmethod
117 async def reject_after_timeout(coro: Any, timeout_ms: int) -> Any:
118 try:
119 return await asyncio.wait_for(coro, timeout=timeout_ms / 1000.0)
120 except asyncio.TimeoutError:
121 raise RequestTimeoutError()
124class InvalidPathParameterError(NotionClientErrorBase):
125 """Error thrown when a path parameter contains invalid characters such as
126 path traversal sequences (..) that could alter the intended API endpoint."""
128 code: ClientErrorCode = ClientErrorCode.InvalidPathParameter
130 def __init__(
131 self,
132 message: str = (
133 "Path parameter contains invalid characters that could alter the request path"
134 ),
135 ) -> None:
136 super().__init__(message)
138 @staticmethod
139 def is_invalid_path_parameter_error(error: object) -> bool:
140 return _is_notion_client_error_with_code(
141 error,
142 {ClientErrorCode.InvalidPathParameter.value},
143 )
146def validate_request_path(path: str) -> None:
147 """Validates that a request path does not contain path traversal sequences.
148 Raises InvalidPathParameterError if the path contains ".." segments,
149 including URL-encoded variants like %2e%2e."""
151 # Check for literal path traversal
152 if ".." in path:
153 raise InvalidPathParameterError(
154 f'Request path "{path}" contains path traversal sequence ".."'
155 )
157 # Check for URL-encoded path traversal (%2e = '.')
158 # Only decode if path contains potential encoded dots
159 if "%2e" in path.lower():
160 decoded = unquote(path)
161 if ".." in decoded:
162 raise InvalidPathParameterError(
163 f'Request path "{path}" contains encoded path traversal sequence'
164 )
167HTTPResponseErrorCode = Union[ClientErrorCode, APIErrorCode]
170class HTTPResponseError(NotionClientErrorBase):
171 code: Union[str, APIErrorCode]
172 status: int
173 headers: httpx.Headers
174 body: str
175 additional_data: Optional[Dict[str, Any]]
176 request_id: Optional[str]
178 def __init__(
179 self,
180 code: Union[str, APIErrorCode],
181 status: int,
182 message: str,
183 headers: httpx.Headers,
184 raw_body_text: str,
185 additional_data: Optional[Dict[str, Any]] = None,
186 request_id: Optional[str] = None,
187 ) -> None:
188 super().__init__(message)
189 self.code = code
190 self.status = status
191 self.headers = headers
192 self.body = raw_body_text
193 self.additional_data = additional_data
194 self.request_id = request_id
197_http_response_error_codes: Set[str] = {
198 ClientErrorCode.ResponseError.value,
199 APIErrorCode.Unauthorized.value,
200 APIErrorCode.RestrictedResource.value,
201 APIErrorCode.ObjectNotFound.value,
202 APIErrorCode.RateLimited.value,
203 APIErrorCode.InvalidJSON.value,
204 APIErrorCode.InvalidRequestURL.value,
205 APIErrorCode.InvalidRequest.value,
206 APIErrorCode.ValidationError.value,
207 APIErrorCode.ConflictError.value,
208 APIErrorCode.InternalServerError.value,
209 APIErrorCode.ServiceUnavailable.value,
210 APIErrorCode.GatewayTimeout.value,
211}
214def is_http_response_error(error: Any) -> TypeGuard[HTTPResponseError]:
215 return _is_notion_client_error_with_code(error, _http_response_error_codes)
218class UnknownHTTPResponseError(HTTPResponseError):
219 """Error thrown if an API call responds with an unknown error code, or does not respond with
220 a properly-formatted error.
221 """
223 def __init__(
224 self,
225 status: int,
226 message: Optional[str] = None,
227 headers: Optional[httpx.Headers] = None,
228 raw_body_text: str = "",
229 ) -> None:
230 if message is None:
231 message = f"Request to Notion API failed with status: {status}"
232 if headers is None:
233 headers = httpx.Headers()
235 super().__init__(
236 code=ClientErrorCode.ResponseError.value,
237 status=status,
238 message=message,
239 headers=headers,
240 raw_body_text=raw_body_text,
241 additional_data=None,
242 request_id=None,
243 )
245 @staticmethod
246 def is_unknown_http_response_error(error: Any) -> bool:
247 return _is_notion_client_error_with_code(
248 error,
249 {ClientErrorCode.ResponseError.value},
250 )
253_api_error_codes: Set[str] = {
254 APIErrorCode.Unauthorized.value,
255 APIErrorCode.RestrictedResource.value,
256 APIErrorCode.ObjectNotFound.value,
257 APIErrorCode.RateLimited.value,
258 APIErrorCode.InvalidJSON.value,
259 APIErrorCode.InvalidRequestURL.value,
260 APIErrorCode.InvalidRequest.value,
261 APIErrorCode.ValidationError.value,
262 APIErrorCode.ConflictError.value,
263 APIErrorCode.InternalServerError.value,
264 APIErrorCode.ServiceUnavailable.value,
265 APIErrorCode.GatewayTimeout.value,
266}
269class APIResponseError(HTTPResponseError):
270 code: APIErrorCode
271 request_id: Optional[str]
273 @staticmethod
274 def is_api_response_error(error: Any) -> TypeGuard["APIResponseError"]:
275 return _is_notion_client_error_with_code(error, _api_error_codes)
278# Type alias for all Notion client errors
279NotionClientError = Union[
280 RequestTimeoutError,
281 UnknownHTTPResponseError,
282 APIResponseError,
283 InvalidPathParameterError,
284]
287def build_request_error(
288 response: httpx.Response,
289 body_text: str,
290) -> Union[APIResponseError, UnknownHTTPResponseError]:
291 api_error_response_body = _parse_api_error_response_body(body_text)
293 if api_error_response_body is not None:
294 return APIResponseError(
295 code=api_error_response_body["code"],
296 message=api_error_response_body["message"],
297 headers=response.headers,
298 status=response.status_code,
299 raw_body_text=body_text,
300 additional_data=api_error_response_body.get("additional_data"),
301 request_id=api_error_response_body.get("request_id"),
302 )
304 return UnknownHTTPResponseError(
305 message=None,
306 headers=response.headers,
307 status=response.status_code,
308 raw_body_text=body_text,
309 )
312def _parse_api_error_response_body(body: str) -> Optional[Dict[str, Any]]:
313 if not isinstance(body, str):
314 return None
316 try:
317 parsed = json.loads(body)
318 except (json.JSONDecodeError, ValueError):
319 return None
321 if not isinstance(parsed, dict):
322 return None
324 message = parsed.get("message")
325 code = parsed.get("code")
327 if not isinstance(message, str) or not _is_api_error_code(code):
328 return None
330 result: Dict[str, Any] = {
331 "code": APIErrorCode(code),
332 "message": message,
333 }
335 if "additional_data" in parsed:
336 result["additional_data"] = parsed["additional_data"]
338 if "request_id" in parsed:
339 result["request_id"] = parsed["request_id"]
341 return result
344def _is_api_error_code(code: Any) -> bool:
345 return isinstance(code, str) and code in _api_error_codes