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 ), 

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

99 ) 

100 

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

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

103 

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

105 """ # noqa: E501 

106 return self.parent.request( 

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

108 method="DELETE", 

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

110 ) 

111 

112 

113class DatabasesEndpoint(Endpoint): 

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

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

116 

117 > ⚠️ **Deprecated endpoint** 

118 

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

120 """ # noqa: E501 

121 return self.parent.request( 

122 path="databases", 

123 method="GET", 

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

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

126 ) 

127 

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

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

130 

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

132 """ # noqa: E501 

133 return self.parent.request( 

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

135 method="POST", 

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

137 body=pick( 

138 kwargs, 

139 "filter", 

140 "sorts", 

141 "start_cursor", 

142 "page_size", 

143 "archived", 

144 "in_trash", 

145 ), 

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

147 ) 

148 

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

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

151 

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

153 """ # noqa: E501 

154 return self.parent.request( 

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

156 ) 

157 

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

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

160 

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

162 """ # noqa: E501 

163 return self.parent.request( 

164 path="databases", 

165 method="POST", 

166 body=pick( 

167 kwargs, 

168 "parent", 

169 "title", 

170 "description", 

171 "properties", 

172 "icon", 

173 "cover", 

174 "is_inline", 

175 ), 

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

177 ) 

178 

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

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

181 

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

183 """ # noqa: E501 

184 return self.parent.request( 

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

186 method="PATCH", 

187 body=pick( 

188 kwargs, 

189 "properties", 

190 "title", 

191 "description", 

192 "icon", 

193 "cover", 

194 "is_inline", 

195 "archived", 

196 "in_trash", 

197 ), 

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

199 ) 

200 

201 

202class PagesPropertiesEndpoint(Endpoint): 

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

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

205 

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

207 """ # noqa: E501 

208 return self.parent.request( 

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

210 method="GET", 

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

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

213 ) 

214 

215 

216class PagesEndpoint(Endpoint): 

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

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

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

220 

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

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

223 

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

225 """ # noqa: E501 

226 return self.parent.request( 

227 path="pages", 

228 method="POST", 

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

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

231 ) 

232 

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

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

235 

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

237 """ # noqa: E501 

238 return self.parent.request( 

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

240 method="GET", 

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

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

243 ) 

244 

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

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

247 

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

249 """ # noqa: E501 

250 return self.parent.request( 

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

252 method="PATCH", 

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

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

255 ) 

256 

257 

258class UsersEndpoint(Endpoint): 

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

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

261 

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

263 """ # noqa: E501 

264 return self.parent.request( 

265 path="users", 

266 method="GET", 

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

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

269 ) 

270 

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

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

273 

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

275 """ # noqa: E501 

276 return self.parent.request( 

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

278 ) 

279 

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

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

282 

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

284 """ # noqa: E501 

285 return self.parent.request( 

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

287 ) 

288 

289 

290class SearchEndpoint(Endpoint): 

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

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

293 

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

295 """ # noqa: E501 

296 return self.parent.request( 

297 path="search", 

298 method="POST", 

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

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

301 ) 

302 

303 

304class CommentsEndpoint(Endpoint): 

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

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

307 

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

309 """ # noqa: E501 

310 return self.parent.request( 

311 path="comments", 

312 method="POST", 

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

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

315 ) 

316 

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

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

319 

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

321 """ # noqa: E501 

322 return self.parent.request( 

323 path="comments", 

324 method="GET", 

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

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

327 )