Coverage for notion_client / errors.py: 100%
128 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-02 12:05 +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
11import httpx
12from urllib.parse import unquote
15class APIErrorCode(str, Enum):
16 """Error codes returned in responses from the API."""
18 Unauthorized = "unauthorized"
19 """The bearer token is not valid."""
21 RestrictedResource = "restricted_resource"
22 """Given the bearer token used, the client doesn't have permission to
23 perform this operation."""
25 ObjectNotFound = "object_not_found"
26 """Given the bearer token used, the resource does not exist.
27 This error can also indicate that the resource has not been shared with owner
28 of the bearer token."""
30 RateLimited = "rate_limited"
31 """This request exceeds the number of requests allowed. Slow down and try again."""
33 InvalidJSON = "invalid_json"
34 """The request body could not be decoded as JSON."""
36 InvalidRequestURL = "invalid_request_url"
37 """The request URL is not valid."""
39 InvalidRequest = "invalid_request"
40 """This request is not supported."""
42 ValidationError = "validation_error"
43 """The request body does not match the schema for the expected parameters."""
45 ConflictError = "conflict_error"
46 """The transaction could not be completed, potentially due to a data collision.
47 Make sure the parameters are up to date and try again."""
49 InternalServerError = "internal_server_error"
50 """An unexpected error occurred. Reach out to Notion support."""
52 ServiceUnavailable = "service_unavailable"
53 """Notion is unavailable. Try again later.
54 This can occur when the time to respond to a request takes longer than 60 seconds,
55 the maximum request timeout."""
58class ClientErrorCode(str, Enum):
59 """Error codes generated for client errors."""
61 RequestTimeout = "notionhq_client_request_timeout"
62 ResponseError = "notionhq_client_response_error"
63 InvalidPathParameter = "notionhq_client_invalid_path_parameter"
66# Error codes on errors thrown by the `Client`.
67NotionErrorCode = Union[APIErrorCode, ClientErrorCode]
70class NotionClientErrorBase(Exception):
71 """Base error type for all Notion client errors."""
73 def __init__(self, message: str = "") -> None:
74 super().__init__(message)
77def is_notion_client_error(error: Any) -> bool:
78 return isinstance(error, NotionClientErrorBase)
81def _is_notion_client_error_with_code(error: Any, codes: Set[str]) -> bool:
82 if not is_notion_client_error(error):
83 return False
84 error_code = error.code
85 if isinstance(error_code, Enum):
86 error_code = error_code.value
88 return error_code in codes
91class RequestTimeoutError(NotionClientErrorBase):
92 """Error thrown by the client if a request times out."""
94 code: ClientErrorCode = ClientErrorCode.RequestTimeout
96 def __init__(self, message: str = "Request to Notion API has timed out") -> None:
97 super().__init__(message)
99 @staticmethod
100 def is_request_timeout_error(error: Any) -> bool:
101 return _is_notion_client_error_with_code(
102 error,
103 {ClientErrorCode.RequestTimeout.value},
104 )
106 @staticmethod
107 async def reject_after_timeout(coro: Any, timeout_ms: int) -> Any:
108 try:
109 return await asyncio.wait_for(coro, timeout=timeout_ms / 1000.0)
110 except asyncio.TimeoutError:
111 raise RequestTimeoutError()
114class InvalidPathParameterError(NotionClientErrorBase):
115 """Error thrown when a path parameter contains invalid characters such as
116 path traversal sequences (..) that could alter the intended API endpoint."""
118 code: ClientErrorCode = ClientErrorCode.InvalidPathParameter
120 def __init__(
121 self,
122 message: str = (
123 "Path parameter contains invalid characters that could alter the request path"
124 ),
125 ) -> None:
126 super().__init__(message)
128 @staticmethod
129 def is_invalid_path_parameter_error(error: object) -> bool:
130 return _is_notion_client_error_with_code(
131 error,
132 {ClientErrorCode.InvalidPathParameter.value},
133 )
136def validate_request_path(path: str) -> None:
137 """Validates that a request path does not contain path traversal sequences.
138 Raises InvalidPathParameterError if the path contains ".." segments,
139 including URL-encoded variants like %2e%2e."""
141 # Check for literal path traversal
142 if ".." in path:
143 raise InvalidPathParameterError(
144 f'Request path "{path}" contains path traversal sequence ".."'
145 )
147 # Check for URL-encoded path traversal (%2e = '.')
148 # Only decode if path contains potential encoded dots
149 if "%2e" in path.lower():
150 decoded = unquote(path)
151 if ".." in decoded:
152 raise InvalidPathParameterError(
153 f'Request path "{path}" contains encoded path traversal sequence'
154 )
157HTTPResponseErrorCode = Union[ClientErrorCode, APIErrorCode]
160class HTTPResponseError(NotionClientErrorBase):
161 code: Union[str, APIErrorCode]
162 status: int
163 headers: httpx.Headers
164 body: str
165 additional_data: Optional[Dict[str, Any]]
166 request_id: Optional[str]
168 def __init__(
169 self,
170 code: Union[str, APIErrorCode],
171 status: int,
172 message: str,
173 headers: httpx.Headers,
174 raw_body_text: str,
175 additional_data: Optional[Dict[str, Any]] = None,
176 request_id: Optional[str] = None,
177 ) -> None:
178 super().__init__(message)
179 self.code = code
180 self.status = status
181 self.headers = headers
182 self.body = raw_body_text
183 self.additional_data = additional_data
184 self.request_id = request_id
187_http_response_error_codes: Set[str] = {
188 ClientErrorCode.ResponseError.value,
189 APIErrorCode.Unauthorized.value,
190 APIErrorCode.RestrictedResource.value,
191 APIErrorCode.ObjectNotFound.value,
192 APIErrorCode.RateLimited.value,
193 APIErrorCode.InvalidJSON.value,
194 APIErrorCode.InvalidRequestURL.value,
195 APIErrorCode.InvalidRequest.value,
196 APIErrorCode.ValidationError.value,
197 APIErrorCode.ConflictError.value,
198 APIErrorCode.InternalServerError.value,
199 APIErrorCode.ServiceUnavailable.value,
200}
203def is_http_response_error(error: Any) -> bool:
204 return _is_notion_client_error_with_code(error, _http_response_error_codes)
207class UnknownHTTPResponseError(HTTPResponseError):
208 """Error thrown if an API call responds with an unknown error code, or does not respond with
209 a properly-formatted error.
210 """
212 def __init__(
213 self,
214 status: int,
215 message: Optional[str] = None,
216 headers: Optional[httpx.Headers] = None,
217 raw_body_text: str = "",
218 ) -> None:
219 if message is None:
220 message = f"Request to Notion API failed with status: {status}"
221 if headers is None:
222 headers = httpx.Headers()
224 super().__init__(
225 code=ClientErrorCode.ResponseError.value,
226 status=status,
227 message=message,
228 headers=headers,
229 raw_body_text=raw_body_text,
230 additional_data=None,
231 request_id=None,
232 )
234 @staticmethod
235 def is_unknown_http_response_error(error: Any) -> bool:
236 return _is_notion_client_error_with_code(
237 error,
238 {ClientErrorCode.ResponseError.value},
239 )
242_api_error_codes: Set[str] = {
243 APIErrorCode.Unauthorized.value,
244 APIErrorCode.RestrictedResource.value,
245 APIErrorCode.ObjectNotFound.value,
246 APIErrorCode.RateLimited.value,
247 APIErrorCode.InvalidJSON.value,
248 APIErrorCode.InvalidRequestURL.value,
249 APIErrorCode.InvalidRequest.value,
250 APIErrorCode.ValidationError.value,
251 APIErrorCode.ConflictError.value,
252 APIErrorCode.InternalServerError.value,
253 APIErrorCode.ServiceUnavailable.value,
254}
257class APIResponseError(HTTPResponseError):
258 code: APIErrorCode
259 request_id: Optional[str]
261 @staticmethod
262 def is_api_response_error(error: Any) -> bool:
263 return _is_notion_client_error_with_code(error, _api_error_codes)
266# Type alias for all Notion client errors
267NotionClientError = Union[
268 RequestTimeoutError,
269 UnknownHTTPResponseError,
270 APIResponseError,
271 InvalidPathParameterError,
272]
275def build_request_error(
276 response: httpx.Response,
277 body_text: str,
278) -> Union[APIResponseError, UnknownHTTPResponseError]:
279 api_error_response_body = _parse_api_error_response_body(body_text)
281 if api_error_response_body is not None:
282 return APIResponseError(
283 code=api_error_response_body["code"],
284 message=api_error_response_body["message"],
285 headers=response.headers,
286 status=response.status_code,
287 raw_body_text=body_text,
288 additional_data=api_error_response_body.get("additional_data"),
289 request_id=api_error_response_body.get("request_id"),
290 )
292 return UnknownHTTPResponseError(
293 message=None,
294 headers=response.headers,
295 status=response.status_code,
296 raw_body_text=body_text,
297 )
300def _parse_api_error_response_body(body: str) -> Optional[Dict[str, Any]]:
301 if not isinstance(body, str):
302 return None
304 try:
305 parsed = json.loads(body)
306 except (json.JSONDecodeError, ValueError):
307 return None
309 if not isinstance(parsed, dict):
310 return None
312 message = parsed.get("message")
313 code = parsed.get("code")
315 if not isinstance(message, str) or not _is_api_error_code(code):
316 return None
318 result: Dict[str, Any] = {
319 "code": APIErrorCode(code),
320 "message": message,
321 }
323 if "additional_data" in parsed:
324 result["additional_data"] = parsed["additional_data"]
326 if "request_id" in parsed:
327 result["request_id"] = parsed["request_id"]
329 return result
332def _is_api_error_code(code: Any) -> bool:
333 return isinstance(code, str) and code in _api_error_codes