Coverage for notion_client / errors.py: 99%
132 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-17 21:35 +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."""
63class ClientErrorCode(str, Enum):
64 """Error codes generated for client errors."""
66 RequestTimeout = "notionhq_client_request_timeout"
67 ResponseError = "notionhq_client_response_error"
68 InvalidPathParameter = "notionhq_client_invalid_path_parameter"
71# Error codes on errors thrown by the `Client`.
72NotionErrorCode = Union[APIErrorCode, ClientErrorCode]
75class NotionClientErrorBase(Exception):
76 """Base error type for all Notion client errors."""
78 code: Union[str, NotionErrorCode]
80 def __init__(self, message: str = "") -> None:
81 super().__init__(message)
84def is_notion_client_error(error: Any) -> TypeGuard["NotionClientError"]:
85 return isinstance(error, NotionClientErrorBase)
88def _is_notion_client_error_with_code(error: Any, codes: Set[str]) -> bool:
89 if not is_notion_client_error(error):
90 return False
91 error_code = error.code
92 if isinstance(error_code, Enum):
93 error_code = error_code.value
95 return error_code in codes
98class RequestTimeoutError(NotionClientErrorBase):
99 """Error thrown by the client if a request times out."""
101 code: ClientErrorCode = ClientErrorCode.RequestTimeout
103 def __init__(self, message: str = "Request to Notion API has timed out") -> None:
104 super().__init__(message)
106 @staticmethod
107 def is_request_timeout_error(error: Any) -> bool:
108 return _is_notion_client_error_with_code(
109 error,
110 {ClientErrorCode.RequestTimeout.value},
111 )
113 @staticmethod
114 async def reject_after_timeout(coro: Any, timeout_ms: int) -> Any:
115 try:
116 return await asyncio.wait_for(coro, timeout=timeout_ms / 1000.0)
117 except asyncio.TimeoutError:
118 raise RequestTimeoutError()
121class InvalidPathParameterError(NotionClientErrorBase):
122 """Error thrown when a path parameter contains invalid characters such as
123 path traversal sequences (..) that could alter the intended API endpoint."""
125 code: ClientErrorCode = ClientErrorCode.InvalidPathParameter
127 def __init__(
128 self,
129 message: str = (
130 "Path parameter contains invalid characters that could alter the request path"
131 ),
132 ) -> None:
133 super().__init__(message)
135 @staticmethod
136 def is_invalid_path_parameter_error(error: object) -> bool:
137 return _is_notion_client_error_with_code(
138 error,
139 {ClientErrorCode.InvalidPathParameter.value},
140 )
143def validate_request_path(path: str) -> None:
144 """Validates that a request path does not contain path traversal sequences.
145 Raises InvalidPathParameterError if the path contains ".." segments,
146 including URL-encoded variants like %2e%2e."""
148 # Check for literal path traversal
149 if ".." in path:
150 raise InvalidPathParameterError(
151 f'Request path "{path}" contains path traversal sequence ".."'
152 )
154 # Check for URL-encoded path traversal (%2e = '.')
155 # Only decode if path contains potential encoded dots
156 if "%2e" in path.lower():
157 decoded = unquote(path)
158 if ".." in decoded:
159 raise InvalidPathParameterError(
160 f'Request path "{path}" contains encoded path traversal sequence'
161 )
164HTTPResponseErrorCode = Union[ClientErrorCode, APIErrorCode]
167class HTTPResponseError(NotionClientErrorBase):
168 code: Union[str, APIErrorCode]
169 status: int
170 headers: httpx.Headers
171 body: str
172 additional_data: Optional[Dict[str, Any]]
173 request_id: Optional[str]
175 def __init__(
176 self,
177 code: Union[str, APIErrorCode],
178 status: int,
179 message: str,
180 headers: httpx.Headers,
181 raw_body_text: str,
182 additional_data: Optional[Dict[str, Any]] = None,
183 request_id: Optional[str] = None,
184 ) -> None:
185 super().__init__(message)
186 self.code = code
187 self.status = status
188 self.headers = headers
189 self.body = raw_body_text
190 self.additional_data = additional_data
191 self.request_id = request_id
194_http_response_error_codes: Set[str] = {
195 ClientErrorCode.ResponseError.value,
196 APIErrorCode.Unauthorized.value,
197 APIErrorCode.RestrictedResource.value,
198 APIErrorCode.ObjectNotFound.value,
199 APIErrorCode.RateLimited.value,
200 APIErrorCode.InvalidJSON.value,
201 APIErrorCode.InvalidRequestURL.value,
202 APIErrorCode.InvalidRequest.value,
203 APIErrorCode.ValidationError.value,
204 APIErrorCode.ConflictError.value,
205 APIErrorCode.InternalServerError.value,
206 APIErrorCode.ServiceUnavailable.value,
207}
210def is_http_response_error(error: Any) -> TypeGuard[HTTPResponseError]:
211 return _is_notion_client_error_with_code(error, _http_response_error_codes)
214class UnknownHTTPResponseError(HTTPResponseError):
215 """Error thrown if an API call responds with an unknown error code, or does not respond with
216 a properly-formatted error.
217 """
219 def __init__(
220 self,
221 status: int,
222 message: Optional[str] = None,
223 headers: Optional[httpx.Headers] = None,
224 raw_body_text: str = "",
225 ) -> None:
226 if message is None:
227 message = f"Request to Notion API failed with status: {status}"
228 if headers is None:
229 headers = httpx.Headers()
231 super().__init__(
232 code=ClientErrorCode.ResponseError.value,
233 status=status,
234 message=message,
235 headers=headers,
236 raw_body_text=raw_body_text,
237 additional_data=None,
238 request_id=None,
239 )
241 @staticmethod
242 def is_unknown_http_response_error(error: Any) -> bool:
243 return _is_notion_client_error_with_code(
244 error,
245 {ClientErrorCode.ResponseError.value},
246 )
249_api_error_codes: Set[str] = {
250 APIErrorCode.Unauthorized.value,
251 APIErrorCode.RestrictedResource.value,
252 APIErrorCode.ObjectNotFound.value,
253 APIErrorCode.RateLimited.value,
254 APIErrorCode.InvalidJSON.value,
255 APIErrorCode.InvalidRequestURL.value,
256 APIErrorCode.InvalidRequest.value,
257 APIErrorCode.ValidationError.value,
258 APIErrorCode.ConflictError.value,
259 APIErrorCode.InternalServerError.value,
260 APIErrorCode.ServiceUnavailable.value,
261}
264class APIResponseError(HTTPResponseError):
265 code: APIErrorCode
266 request_id: Optional[str]
268 @staticmethod
269 def is_api_response_error(error: Any) -> TypeGuard["APIResponseError"]:
270 return _is_notion_client_error_with_code(error, _api_error_codes)
273# Type alias for all Notion client errors
274NotionClientError = Union[
275 RequestTimeoutError,
276 UnknownHTTPResponseError,
277 APIResponseError,
278 InvalidPathParameterError,
279]
282def build_request_error(
283 response: httpx.Response,
284 body_text: str,
285) -> Union[APIResponseError, UnknownHTTPResponseError]:
286 api_error_response_body = _parse_api_error_response_body(body_text)
288 if api_error_response_body is not None:
289 return APIResponseError(
290 code=api_error_response_body["code"],
291 message=api_error_response_body["message"],
292 headers=response.headers,
293 status=response.status_code,
294 raw_body_text=body_text,
295 additional_data=api_error_response_body.get("additional_data"),
296 request_id=api_error_response_body.get("request_id"),
297 )
299 return UnknownHTTPResponseError(
300 message=None,
301 headers=response.headers,
302 status=response.status_code,
303 raw_body_text=body_text,
304 )
307def _parse_api_error_response_body(body: str) -> Optional[Dict[str, Any]]:
308 if not isinstance(body, str):
309 return None
311 try:
312 parsed = json.loads(body)
313 except (json.JSONDecodeError, ValueError):
314 return None
316 if not isinstance(parsed, dict):
317 return None
319 message = parsed.get("message")
320 code = parsed.get("code")
322 if not isinstance(message, str) or not _is_api_error_code(code):
323 return None
325 result: Dict[str, Any] = {
326 "code": APIErrorCode(code),
327 "message": message,
328 }
330 if "additional_data" in parsed:
331 result["additional_data"] = parsed["additional_data"]
333 if "request_id" in parsed:
334 result["request_id"] = parsed["request_id"]
336 return result
339def _is_api_error_code(code: Any) -> bool:
340 return isinstance(code, str) and code in _api_error_codes