Coverage for notion_client / client.py: 100%
127 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-02 12:05 +0000
1"""Synchronous and asynchronous clients for Notion's API."""
3import base64
4import logging
5from abc import abstractmethod
6from dataclasses import dataclass
7from types import TracebackType
8from typing import Any, Dict, List, Optional, Type, Union
10import httpx
11from httpx import Request, Response
13from notion_client.api_endpoints import (
14 BlocksEndpoint,
15 CommentsEndpoint,
16 DatabasesEndpoint,
17 DataSourcesEndpoint,
18 PagesEndpoint,
19 SearchEndpoint,
20 UsersEndpoint,
21 FileUploadsEndpoint,
22 OAuthEndpoint,
23)
24from notion_client.errors import (
25 RequestTimeoutError,
26 build_request_error,
27)
28from notion_client.logging import make_console_logger
29from notion_client.typing import SyncAsync
32@dataclass
33class ClientOptions:
34 """Options to configure the client.
36 Attributes:
37 auth: Bearer token for authentication. If left undefined, the `auth` parameter
38 should be set on each request.
39 timeout_ms: Number of milliseconds to wait before emitting a
40 `RequestTimeoutError`.
41 base_url: The root URL for sending API requests. This can be changed to test
42 with a mock server.
43 log_level: Verbosity of logs the instance will produce. By default, logs are
44 written to `stdout`.
45 logger: A custom logger.
46 notion_version: Notion version to use.
47 """
49 auth: Optional[str] = None
50 timeout_ms: int = 60_000
51 base_url: str = "https://api.notion.com"
52 log_level: int = logging.WARNING
53 logger: Optional[logging.Logger] = None
54 notion_version: str = "2025-09-03"
57class BaseClient:
58 def __init__(
59 self,
60 client: Union[httpx.Client, httpx.AsyncClient],
61 options: Optional[Union[Dict[str, Any], ClientOptions]] = None,
62 **kwargs: Any,
63 ) -> None:
64 if options is None:
65 options = ClientOptions(**kwargs)
66 elif isinstance(options, dict):
67 options = ClientOptions(**options)
69 self.logger = options.logger or make_console_logger()
70 self.logger.setLevel(options.log_level)
71 self.options = options
73 self._clients: List[Union[httpx.Client, httpx.AsyncClient]] = []
74 self.client = client
76 self.blocks = BlocksEndpoint(self)
77 self.databases = DatabasesEndpoint(self)
78 self.data_sources = DataSourcesEndpoint(self)
79 self.users = UsersEndpoint(self)
80 self.pages = PagesEndpoint(self)
81 self.search = SearchEndpoint(self)
82 self.comments = CommentsEndpoint(self)
83 self.file_uploads = FileUploadsEndpoint(self)
84 self.oauth = OAuthEndpoint(self)
86 @property
87 def client(self) -> Union[httpx.Client, httpx.AsyncClient]:
88 return self._clients[-1]
90 @client.setter
91 def client(self, client: Union[httpx.Client, httpx.AsyncClient]) -> None:
92 client.base_url = httpx.URL(f"{self.options.base_url}/v1/")
93 client.timeout = httpx.Timeout(timeout=self.options.timeout_ms / 1_000)
94 client.headers = httpx.Headers(
95 {
96 "Notion-Version": self.options.notion_version,
97 "User-Agent": "ramnes/notion-sdk-py@2.7.0",
98 }
99 )
100 if self.options.auth:
101 client.headers["Authorization"] = f"Bearer {self.options.auth}"
102 self._clients.append(client)
104 def _build_request(
105 self,
106 method: str,
107 path: str,
108 query: Optional[Dict[Any, Any]] = None,
109 body: Optional[Dict[Any, Any]] = None,
110 form_data: Optional[Dict[Any, Any]] = None,
111 auth: Optional[Union[str, Dict[str, str]]] = None,
112 ) -> Request:
113 headers = httpx.Headers()
114 if auth:
115 if isinstance(auth, dict):
116 client_id = auth.get("client_id", "")
117 client_secret = auth.get("client_secret", "")
118 credentials = f"{client_id}:{client_secret}"
119 encoded_credentials = base64.b64encode(credentials.encode()).decode()
120 headers["Authorization"] = f"Basic {encoded_credentials}"
121 else:
122 headers["Authorization"] = f"Bearer {auth}"
123 self.logger.info(f"{method} {self.client.base_url}{path}")
124 self.logger.debug(f"=> {query} -- {body} -- {form_data}")
126 if not form_data:
127 return self.client.build_request(
128 method,
129 path,
130 params=query,
131 json=body,
132 headers=headers,
133 )
135 files: Dict[str, Any] = {}
136 data: Dict[str, Any] = {}
137 for key, value in form_data.items():
138 if isinstance(value, tuple) and len(value) >= 2:
139 files[key] = value
140 elif hasattr(value, "read"):
141 files[key] = value
142 elif isinstance(value, str):
143 data[key] = value
144 else:
145 data[key] = str(value)
147 return self.client.build_request(
148 method,
149 path,
150 params=query,
151 files=files,
152 data=data,
153 headers=headers,
154 )
156 def _parse_response(self, response: Response) -> Any:
157 try:
158 response.raise_for_status()
159 except httpx.HTTPStatusError as error:
160 body_text = error.response.text
161 raise build_request_error(error.response, body_text)
163 body = response.json()
164 self.logger.debug(f"=> {body}")
166 return body
168 @abstractmethod
169 def request(
170 self,
171 path: str,
172 method: str,
173 query: Optional[Dict[Any, Any]] = None,
174 body: Optional[Dict[Any, Any]] = None,
175 form_data: Optional[Dict[Any, Any]] = None,
176 auth: Optional[Union[str, Dict[str, str]]] = None,
177 ) -> SyncAsync[Any]:
178 # noqa
179 pass
182class Client(BaseClient):
183 """Synchronous client for Notion's API."""
185 client: httpx.Client
187 def __init__(
188 self,
189 options: Optional[Union[Dict[Any, Any], ClientOptions]] = None,
190 client: Optional[httpx.Client] = None,
191 **kwargs: Any,
192 ) -> None:
193 if client is None:
194 client = httpx.Client()
195 super().__init__(client, options, **kwargs)
197 def __enter__(self) -> "Client":
198 self.client = httpx.Client()
199 self.client.__enter__()
200 return self
202 def __exit__(
203 self,
204 exc_type: Type[BaseException],
205 exc_value: BaseException,
206 traceback: TracebackType,
207 ) -> None:
208 self.client.__exit__(exc_type, exc_value, traceback)
209 del self._clients[-1]
211 def close(self) -> None:
212 """Close the connection pool of the current inner client."""
213 self.client.close()
215 def request(
216 self,
217 path: str,
218 method: str,
219 query: Optional[Dict[Any, Any]] = None,
220 body: Optional[Dict[Any, Any]] = None,
221 form_data: Optional[Dict[Any, Any]] = None,
222 auth: Optional[Union[str, Dict[str, str]]] = None,
223 ) -> Any:
224 """Send an HTTP request."""
225 request = self._build_request(method, path, query, body, form_data, auth)
226 try:
227 response = self.client.send(request)
228 except httpx.TimeoutException:
229 raise RequestTimeoutError()
230 return self._parse_response(response)
233class AsyncClient(BaseClient):
234 """Asynchronous client for Notion's API."""
236 client: httpx.AsyncClient
238 def __init__(
239 self,
240 options: Optional[Union[Dict[str, Any], ClientOptions]] = None,
241 client: Optional[httpx.AsyncClient] = None,
242 **kwargs: Any,
243 ) -> None:
244 if client is None:
245 client = httpx.AsyncClient()
246 super().__init__(client, options, **kwargs)
248 async def __aenter__(self) -> "AsyncClient":
249 self.client = httpx.AsyncClient()
250 await self.client.__aenter__()
251 return self
253 async def __aexit__(
254 self,
255 exc_type: Type[BaseException],
256 exc_value: BaseException,
257 traceback: TracebackType,
258 ) -> None:
259 await self.client.__aexit__(exc_type, exc_value, traceback)
260 del self._clients[-1]
262 async def aclose(self) -> None:
263 """Close the connection pool of the current inner client."""
264 await self.client.aclose()
266 async def request(
267 self,
268 path: str,
269 method: str,
270 query: Optional[Dict[Any, Any]] = None,
271 body: Optional[Dict[Any, Any]] = None,
272 form_data: Optional[Dict[Any, Any]] = None,
273 auth: Optional[Union[str, Dict[str, str]]] = None,
274 ) -> Any:
275 """Send an HTTP request asynchronously."""
276 request = self._build_request(method, path, query, body, form_data, auth)
277 try:
278 response = await self.client.send(request)
279 except httpx.TimeoutException:
280 raise RequestTimeoutError()
281 return self._parse_response(response)