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 "bookmark", 

71 "image", 

72 "video", 

73 "pdf", 

74 "file", 

75 "audio", 

76 "code", 

77 "equation", 

78 "divider", 

79 "breadcrumb", 

80 "table_of_contents", 

81 "link_to_page", 

82 "table_row", 

83 "heading_1", 

84 "heading_2", 

85 "heading_3", 

86 "paragraph", 

87 "bulleted_list_item", 

88 "numbered_list_item", 

89 "quote", 

90 "to_do", 

91 "toggle", 

92 "template", 

93 "callout", 

94 "synced_block", 

95 "table", 

96 ), 

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

98 ) 

99 

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

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

102 

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

104 """ # noqa: E501 

105 return self.parent.request( 

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

107 method="DELETE", 

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

109 ) 

110 

111 

112class DatabasesEndpoint(Endpoint): 

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

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

115 

116 > ⚠️ **Deprecated endpoint** 

117 

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

119 """ # noqa: E501 

120 return self.parent.request( 

121 path="databases", 

122 method="GET", 

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

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

125 ) 

126 

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

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

129 

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

131 """ # noqa: E501 

132 return self.parent.request( 

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

134 method="POST", 

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

136 body=pick(kwargs, "filter", "sorts", "start_cursor", "page_size"), 

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

138 ) 

139 

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

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

142 

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

144 """ # noqa: E501 

145 return self.parent.request( 

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

147 ) 

148 

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

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

151 

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

153 """ # noqa: E501 

154 return self.parent.request( 

155 path="databases", 

156 method="POST", 

157 body=pick( 

158 kwargs, "parent", "title", "properties", "icon", "cover", "is_inline" 

159 ), 

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

161 ) 

162 

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

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

165 

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

167 """ # noqa: E501 

168 return self.parent.request( 

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

170 method="PATCH", 

171 body=pick( 

172 kwargs, 

173 "properties", 

174 "title", 

175 "description", 

176 "icon", 

177 "cover", 

178 "is_inline", 

179 ), 

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

181 ) 

182 

183 

184class PagesPropertiesEndpoint(Endpoint): 

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

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

187 

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

189 """ # noqa: E501 

190 return self.parent.request( 

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

192 method="GET", 

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

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

195 ) 

196 

197 

198class PagesEndpoint(Endpoint): 

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

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

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

202 

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

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

205 

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

207 """ # noqa: E501 

208 return self.parent.request( 

209 path="pages", 

210 method="POST", 

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

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

213 ) 

214 

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

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

217 

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

219 """ # noqa: E501 

220 return self.parent.request( 

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

222 method="GET", 

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

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

225 ) 

226 

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

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

229 

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

231 """ # noqa: E501 

232 return self.parent.request( 

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

234 method="PATCH", 

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

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

237 ) 

238 

239 

240class UsersEndpoint(Endpoint): 

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

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

243 

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

245 """ # noqa: E501 

246 return self.parent.request( 

247 path="users", 

248 method="GET", 

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

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

251 ) 

252 

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

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

255 

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

257 """ # noqa: E501 

258 return self.parent.request( 

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

260 ) 

261 

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

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

264 

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

266 """ # noqa: E501 

267 return self.parent.request( 

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

269 ) 

270 

271 

272class SearchEndpoint(Endpoint): 

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

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

275 

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

277 """ # noqa: E501 

278 return self.parent.request( 

279 path="search", 

280 method="POST", 

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

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

283 ) 

284 

285 

286class CommentsEndpoint(Endpoint): 

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

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

289 

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

291 """ # noqa: E501 

292 return self.parent.request( 

293 path="comments", 

294 method="POST", 

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

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

297 ) 

298 

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

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

301 

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

303 """ # noqa: E501 

304 return self.parent.request( 

305 path="comments", 

306 method="GET", 

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

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

309 )