Coverage for notion_client/api_endpoints.py: 100%

1"""Notion API endpoints.""" # noqa: E501 

2 

3from typing import TYPE_CHECKING, Any 

4 

5from notion_client.helpers import pick 

6from notion_client.typing import SyncAsync 

7 

8if TYPE_CHECKING: # pragma: no cover 

9 from notion_client.client import BaseClient 

10 

11 

12class Endpoint: 

13 def __init__(self, parent: "BaseClient") -> None: 

14 self.parent = parent 

15 

16 

17class BlocksChildrenEndpoint(Endpoint): 

18 def append(self, block_id: str, **kwargs: Any) -> SyncAsync[Any]: 

19 """Create and append new children blocks to the block using the ID specified. 

20 

21 *[🔗 Endpoint documentation](https://developers.notion.com/reference/patch-block-children)* 

22 """ # noqa: E501 

23 return self.parent.request( 

24 path=f"blocks/{block_id}/children", 

25 method="PATCH", 

26 body=pick(kwargs, "children", "after"), 

27 auth=kwargs.get("auth"), 

28 ) 

29 

30 def list(self, block_id: str, **kwargs: Any) -> SyncAsync[Any]: 

31 """Return a paginated array of child [block objects](https://developers.notion.com/reference/block) contained in the block. 

32 

33 *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-block-children)* 

34 """ # noqa: E501 

35 return self.parent.request( 

36 path=f"blocks/{block_id}/children", 

37 method="GET", 

38 query=pick(kwargs, "start_cursor", "page_size"), 

39 auth=kwargs.get("auth"), 

40 ) 

41 

42 

43class BlocksEndpoint(Endpoint): 

44 def __init__(self, *args: Any, **kwargs: Any) -> None: 

45 super().__init__(*args, **kwargs) 

46 self.children = BlocksChildrenEndpoint(*args, **kwargs) 

47 

48 def retrieve(self, block_id: str, **kwargs: Any) -> SyncAsync[Any]: 

49 """Retrieve a [Block object](https://developers.notion.com/reference/block) using the ID specified. 

50 

51 *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-block)* 

52 """ # noqa: E501 

53 return self.parent.request( 

54 path=f"blocks/{block_id}", method="GET", auth=kwargs.get("auth") 

55 ) 

56 

57 def update(self, block_id: str, **kwargs: Any) -> SyncAsync[Any]: 

58 """Update the content for the specified `block_id` based on the block type. 

59 

60 *[🔗 Endpoint documentation](https://developers.notion.com/reference/update-a-block)* 

61 """ # noqa: E501 

62 return self.parent.request( 

63 path=f"blocks/{block_id}", 

64 method="PATCH", 

65 body=pick( 

66 kwargs, 

67 "embed", 

68 "type", 

69 "archived", 

70 "in_trash", 

71 "bookmark", 

72 "image", 

73 "video", 

74 "pdf", 

75 "file", 

76 "audio", 

77 "code", 

78 "equation", 

79 "divider", 

80 "breadcrumb", 

81 "table_of_contents", 

82 "link_to_page", 

83 "table_row", 

84 "heading_1", 

85 "heading_2", 

86 "heading_3", 

87 "paragraph", 

88 "bulleted_list_item", 

89 "numbered_list_item", 

90 "quote", 

91 "to_do", 

92 "toggle", 

93 "template", 

94 "callout", 

95 "synced_block", 

96 "table", 

97 "column", 

98 ), 

99 auth=kwargs.get("auth"), 

100 ) 

101 

102 def delete(self, block_id: str, **kwargs: Any) -> SyncAsync[Any]: 

103 """Set a [Block object](https://developers.notion.com/reference/block), including page blocks, to `archived: true`. 

104 

105 *[🔗 Endpoint documentation](https://developers.notion.com/reference/delete-a-block)* 

106 """ # noqa: E501 

107 return self.parent.request( 

108 path=f"blocks/{block_id}", 

109 method="DELETE", 

110 auth=kwargs.get("auth"), 

111 ) 

112 

113 

114class DatabasesEndpoint(Endpoint): 

115 def list(self, **kwargs: Any) -> SyncAsync[Any]: # pragma: no cover 

116 """List all [Databases](https://developers.notion.com/reference/database) shared with the authenticated integration. 

117 

118 > ⚠️ **Deprecated endpoint** 

119 

120 *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-databases)* 

121 """ # noqa: E501 

122 return self.parent.request( 

123 path="databases", 

124 method="GET", 

125 query=pick(kwargs, "start_cursor", "page_size"), 

126 auth=kwargs.get("auth"), 

127 ) 

128 

129 def query(self, database_id: str, **kwargs: Any) -> SyncAsync[Any]: 

130 """Get a list of [Pages](https://developers.notion.com/reference/page) contained in the database. 

131 

132 *[🔗 Endpoint documentation](https://developers.notion.com/reference/post-database-query)* 

133 """ # noqa: E501 

134 return self.parent.request( 

135 path=f"databases/{database_id}/query", 

136 method="POST", 

137 query=pick(kwargs, "filter_properties"), 

138 body=pick( 

139 kwargs, 

140 "filter", 

141 "sorts", 

142 "start_cursor", 

143 "page_size", 

144 "archived", 

145 "in_trash", 

146 ), 

147 auth=kwargs.get("auth"), 

148 ) 

149 

150 def retrieve(self, database_id: str, **kwargs: Any) -> SyncAsync[Any]: 

151 """Retrieve a [Database object](https://developers.notion.com/reference/database) using the ID specified. 

152 

153 *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-database)* 

154 """ # noqa: E501 

155 return self.parent.request( 

156 path=f"databases/{database_id}", method="GET", auth=kwargs.get("auth") 

157 ) 

158 

159 def create(self, **kwargs: Any) -> SyncAsync[Any]: 

160 """Create a database as a subpage in the specified parent page. 

161 

162 *[🔗 Endpoint documentation](https://developers.notion.com/reference/create-a-database)* 

163 """ # noqa: E501 

164 return self.parent.request( 

165 path="databases", 

166 method="POST", 

167 body=pick( 

168 kwargs, 

169 "parent", 

170 "title", 

171 "description", 

172 "properties", 

173 "icon", 

174 "cover", 

175 "is_inline", 

176 ), 

177 auth=kwargs.get("auth"), 

178 ) 

179 

180 def update(self, database_id: str, **kwargs: Any) -> SyncAsync[Any]: 

181 """Update an existing database as specified by the parameters. 

182 

183 *[🔗 Endpoint documentation](https://developers.notion.com/reference/update-a-database)* 

184 """ # noqa: E501 

185 return self.parent.request( 

186 path=f"databases/{database_id}", 

187 method="PATCH", 

188 body=pick( 

189 kwargs, 

190 "properties", 

191 "title", 

192 "description", 

193 "icon", 

194 "cover", 

195 "is_inline", 

196 "archived", 

197 "in_trash", 

198 ), 

199 auth=kwargs.get("auth"), 

200 ) 

201 

202 

203class PagesPropertiesEndpoint(Endpoint): 

204 def retrieve(self, page_id: str, property_id: str, **kwargs: Any) -> SyncAsync[Any]: 

205 """Retrieve a `property_item` object for a given `page_id` and `property_id`. 

206 

207 *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-page-property)* 

208 """ # noqa: E501 

209 return self.parent.request( 

210 path=f"pages/{page_id}/properties/{property_id}", 

211 method="GET", 

212 auth=kwargs.get("auth"), 

213 query=pick(kwargs, "start_cursor", "page_size"), 

214 ) 

215 

216 

217class PagesEndpoint(Endpoint): 

218 def __init__(self, *args: Any, **kwargs: Any) -> None: 

219 super().__init__(*args, **kwargs) 

220 self.properties = PagesPropertiesEndpoint(*args, **kwargs) 

221 

222 def create(self, **kwargs: Any) -> SyncAsync[Any]: 

223 """Create a new page in the specified database or as a child of an existing page. 

224 

225 *[🔗 Endpoint documentation](https://developers.notion.com/reference/post-page)* 

226 """ # noqa: E501 

227 return self.parent.request( 

228 path="pages", 

229 method="POST", 

230 body=pick(kwargs, "parent", "properties", "children", "icon", "cover"), 

231 auth=kwargs.get("auth"), 

232 ) 

233 

234 def retrieve(self, page_id: str, **kwargs: Any) -> SyncAsync[Any]: 

235 """Retrieve a [Page object](https://developers.notion.com/reference/page) using the ID specified. 

236 

237 *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-page)* 

238 """ # noqa: E501 

239 return self.parent.request( 

240 path=f"pages/{page_id}", 

241 method="GET", 

242 query=pick(kwargs, "filter_properties"), 

243 auth=kwargs.get("auth"), 

244 ) 

245 

246 def update(self, page_id: str, **kwargs: Any) -> SyncAsync[Any]: 

247 """Update [page property values](https://developers.notion.com/reference/page#property-value-object) for the specified page. 

248 

249 *[🔗 Endpoint documentation](https://developers.notion.com/reference/patch-page)* 

250 """ # noqa: E501 

251 return self.parent.request( 

252 path=f"pages/{page_id}", 

253 method="PATCH", 

254 body=pick(kwargs, "in_trash", "archived", "properties", "icon", "cover"), 

255 auth=kwargs.get("auth"), 

256 ) 

257 

258 

259class UsersEndpoint(Endpoint): 

260 def list(self, **kwargs: Any) -> SyncAsync[Any]: 

261 """Return a paginated list of [Users](https://developers.notion.com/reference/user) for the workspace. 

262 

263 *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-users)* 

264 """ # noqa: E501 

265 return self.parent.request( 

266 path="users", 

267 method="GET", 

268 query=pick(kwargs, "start_cursor", "page_size"), 

269 auth=kwargs.get("auth"), 

270 ) 

271 

272 def retrieve(self, user_id: str, **kwargs: Any) -> SyncAsync[Any]: 

273 """Retrieve a [User](https://developers.notion.com/reference/user) using the ID specified. 

274 

275 *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-user)* 

276 """ # noqa: E501 

277 return self.parent.request( 

278 path=f"users/{user_id}", method="GET", auth=kwargs.get("auth") 

279 ) 

280 

281 def me(self, **kwargs: Any) -> SyncAsync[Any]: 

282 """Retrieve the bot [User](https://developers.notion.com/reference/user) associated with the API token. 

283 

284 *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-self)* 

285 """ # noqa: E501 

286 return self.parent.request( 

287 path="users/me", method="GET", auth=kwargs.get("auth") 

288 ) 

289 

290 

291class SearchEndpoint(Endpoint): 

292 def __call__(self, **kwargs: Any) -> SyncAsync[Any]: 

293 """Search all pages and child pages that are shared with the integration. 

294 

295 *[🔗 Endpoint documentation](https://developers.notion.com/reference/post-search)* 

296 """ # noqa: E501 

297 return self.parent.request( 

298 path="search", 

299 method="POST", 

300 body=pick(kwargs, "query", "sort", "filter", "start_cursor", "page_size"), 

301 auth=kwargs.get("auth"), 

302 ) 

303 

304 

305class CommentsEndpoint(Endpoint): 

306 def create(self, **kwargs: Any) -> SyncAsync[Any]: 

307 """Create a new comment in the specified page or existing discussion thread. 

308 

309 *[🔗 Endpoint documentation](https://developers.notion.com/reference/create-a-comment)* 

310 """ # noqa: E501 

311 return self.parent.request( 

312 path="comments", 

313 method="POST", 

314 body=pick(kwargs, "parent", "discussion_id", "rich_text"), 

315 auth=kwargs.get("auth"), 

316 ) 

317 

318 def list(self, **kwargs: Any) -> SyncAsync[Any]: 

319 """Retrieve a list of un-resolved [Comment objects](https://developers.notion.com/reference/comment-object) from the specified block. 

320 

321 *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-comment)* 

322 """ # noqa: E501 

323 return self.parent.request( 

324 path="comments", 

325 method="GET", 

326 query=pick(kwargs, "block_id", "start_cursor", "page_size"), 

327 auth=kwargs.get("auth"), 

328 )