Skip to content

Async

AsyncNotionAPI

Main class for Notion API wrapper.

Parameters:

Name Type Description Default
access_token str

Notion access token

 required
api_version str

Version of the notion API

 '2022-06-28'
rate_limit tuple[int, int]

(number_of_requests, number of seconds). Default

 (500, 200)
Source code in python_notion_api/async_api/api.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
class AsyncNotionAPI:
    """Main class for Notion API wrapper.

    Args:
        access_token: Notion access token
        api_version: Version of the notion API
        rate_limit: (number_of_requests, number of seconds). Default
        is set at the rate limit of Notion (3 per second), with a longer
        interval to allow bursts.
    """

    def __init__(
        self,
        access_token: str,
        api_version: str = "2022-06-28",
        page_limit: int = 20,
        rate_limit: tuple[int, int] = (500, 200),
    ):
        self._access_token = access_token
        self._base_url = "https://api.notion.com/v1/"
        self._api_version = api_version
        self._default_retry_strategy = RetryStrategy(
            total=3,
            backoff_factor=0.1,
            status_forcelist=[429, 500, 502, 503, 504, 409],
        )
        self._page_limit = page_limit
        self.limiter = AsyncLimiter(*rate_limit)

    @property
    def request_headers(self):
        """Gets request headers for making requests."""
        return {
            "Authorization": f"Bearer {self._access_token}",
            "Notion-Version": f"{self._api_version}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }

    async def get_database(self, database_id: str) -> NotionDatabase:
        """Gets Notion database.

        Args:
            database_id: Id of the database to fetch.

        Returns:
            A Notion database with the given id.
        """
        database = NotionDatabase(self, database_id)
        await database.reload()

        return database

    async def get_page(
        self, page_id: str, page_cast: type[NotionPage] = NotionPage
    ) -> NotionPage:
        """Gets Notion page.

        Args:
            page_id: Id of the database to fetch.
            page_cast: A subclass of a NotionPage. Allows custom
                property retrieval.
        Returns:
            A Notion page with the given id.
        """
        page = page_cast(api=self, page_id=page_id)
        await page.reload()

        return page

    async def get_block(self, block_id: str) -> NotionBlock:
        """Gets Notion block.

        Args:
            block_id: Id of the block to fetch.
        """
        block = NotionBlock(self, block_id)
        await block.reload()

        return block

    async def me(self) -> User:
        return await self._get("users/me")

    async def _request_attempt(
        self,
        session: aiohttp.ClientSession,
        request_type: Literal["get", "post", "patch"],
        url: str = "",
        params: Dict[str, Any] = {},
        data: Optional[str] = None,
    ):
        """Attempts a request to url.

        Args:
            url: URL for the request.
            request_type: Type of the http request to make.
            data: Data to pass to the request.
            params: Params to pass to the request.
        """
        async with self.limiter:
            return await session.request(
                method=request_type,
                url=url,
                headers=self.request_headers,
                params=params,
                data=data,
            )

    async def _request(
        self,
        request_type: Literal["get", "post", "patch"],
        endpoint: str = "",
        params: Dict[str, Any] = {},
        data: Optional[str] = None,
        cast_cls: Type[NotionObjectBase] = NotionObject,
        retry_strategy: Optional[RetryStrategy] = None,
    ) -> NotionObject:
        """Main request handler.

        Should not be called directly, for internal use only.

        Args:
            request_type: Type of the http request to make.
            endpoint: Endpoint of the request. Will be prefixed with the
                notion API base url.
            params: Params to pass to the request.
            data: Data to pass to the request.
            params: Params to pass to the request.
            cast_cls: A NotionObjectBase class to auto-cast the response of the
                request to.
        """
        retry_strategy = retry_strategy or self._default_retry_strategy

        url = self._base_url + endpoint

        logger.debug(f"Sending {request_type} request to {url}")

        response = None

        async with aiohttp.ClientSession() as session:
            for i in range(retry_strategy.total):
                response = await self._request_attempt(
                    request_type=request_type,
                    session=session,
                    url=url,
                    params=params,
                    data=data,
                )

                response_data = await response.read()
                decoded_data = response_data.decode("utf-8")

                if response.status == 200:
                    return cast_cls.from_obj(json.loads(decoded_data))

                elif response.status not in retry_strategy.status_forcelist:
                    logger.error(
                        f"Request to {url} failed:"
                        f"\n{response.status}\n{decoded_data}"
                    )
                    raise Exception("Request failed")

                if response.status == 429:
                    delay = int(response.headers["Retry-After"])
                    logger.warning(
                        f"Request to {url} failed:"
                        f"\n{response.status}"
                        f"\nRetry-After: {delay}"
                        f"\n{decoded_data}"
                    )
                else:
                    delay = min(
                        retry_strategy.backoff_factor * (2 ** (i)),
                        retry_strategy.max_backoff,
                    )

                logger.warning(
                    f"Notion is busy ({response.status})."
                    f"Retrying ({i+1}) in {delay}s"
                )
                await asyncio.sleep(delay)

        logger.warning(
            f"Request failed after {retry_strategy.total}" " attempts."
        )
        raise MaxRetryError("Request failed")

    async def _post(
        self,
        endpoint: str,
        data: Optional[str] = None,
        cast_cls: Type[NotionObjectBase] = NotionObject,
        retry_strategy: Any = None,
    ) -> NotionObject:
        """Wrapper for post requests.

        Should not be called directly, for internal use only.

        Args:
            endpoint: Endpoint of the request. Will be prepened with the
                notion API base url.
            data: Data to pass to the request.
            cast_cls: A NotionObjectBase class to auto-cast the response of the
                request to.
        """
        return await self._request(
            request_type="post",
            endpoint=endpoint,
            data=data,
            cast_cls=cast_cls,
            retry_strategy=retry_strategy,
        )

    async def _get(
        self,
        endpoint: str,
        params: Dict[str, str] = {},
        cast_cls: Type[NotionObjectBase] = NotionObject,
    ) -> NotionObject:
        """Wrapper for post requests.

        Should not be called directly, for internal use only.

        Args:
            endpoint: Endpoint of the request. Will be prepened with the
                notion API base url.
            params: Params to pass to the request.
            cast_cls: A NotionObjectBase class to auto-cast the response of the
                request to.
        """
        return await self._request(
            request_type="get",
            endpoint=endpoint,
            params=params,
            cast_cls=cast_cls,
        )

    async def _patch(
        self,
        endpoint: str,
        params: Dict[str, str] = {},
        data: Optional[str] = None,
        cast_cls=NotionObject,
    ) -> NotionObject:
        """Wrapper for patch requests.

        Should not be called directly, for internal use only.

        Args:
            endpoint: Endpoint of the request. Will be prepened with the
                notion API base url.
            params: Params to pass to the request.
            data: Data to pass to the request.
            cast_cls: A NotionObjectBase class to auto-cast the response of the
                request to.
        """
        return await self._request(
            request_type="patch",
            endpoint=endpoint,
            params=params,
            data=data,
            cast_cls=cast_cls,
        )

    async def _post_iterate(
        self,
        endpoint: str,
        data: Dict[str, Any] = {},
        page_limit: Optional[int] = None,
    ) -> NotionObjectGenerator:
        """Wrapper for post requests where expected return type is Pagination.

        Should not be called directly, for internal use only.

        Args:
            endpoint: Endpoint of the request. Will be prefixed with the
                notion API base url.
            data: Data to pass to the request.
        """
        has_more = True
        cursor = None
        page_size = page_limit or self._page_limit

        while has_more:
            data.update({"start_cursor": cursor, "page_size": page_size})

            if cursor is None:
                data.pop("start_cursor")

            while page_size > 0:
                try:
                    response = await self._post(
                        endpoint=endpoint, data=json.dumps(data)
                    )

                    for item in response.results:
                        yield item

                    has_more = response.has_more
                    cursor = response.next_cursor

                    break
                except MaxRetryError as e:
                    page_size = floor(page_size / 2)
                    logger.warning(
                        f"Retrying request with smaller page size({page_size})"
                    )
                    if page_size == 0:
                        raise e
                    data.update({"page_size": page_size})

    async def _get_iterate(
        self,
        endpoint: str,
        params: Dict[str, Any] = {},
        page_limit: Optional[int] = None,
    ) -> NotionObjectGenerator:
        """Wrapper for get requests where expected return type is Pagination.

        Should not be called directly, for internal use only.

        Args:
            endpoint: Endpoint of the request. Will be prepened with the
                notion API base url.
            params: Params to pass to the request.
        """
        has_more = True
        cursor = None
        page_size = page_limit or self._page_limit

        while has_more:
            params.update({"start_cursor": cursor, "page_size": page_size})

            if cursor is None:
                params.pop("start_cursor")

            while page_size > 0:
                try:
                    response = await self._get(
                        endpoint=endpoint, params=params
                    )

                    if hasattr(response, "property_item"):
                        # Required for rollups
                        property_item = response.property_item
                    else:
                        # property doesn't exist for Blocks
                        property_item = None

                    for item in response.results:
                        yield item, property_item

                    has_more = response.has_more
                    cursor = response.next_cursor

                    break
                except MaxRetryError as e:
                    page_size = floor(page_size / 2)
                    logger.warning(
                        f"Retrying request with smaller page size({page_size})"
                    )
                    if page_size == 0:
                        raise e
                    params.update({"page_size": page_size})

request_headers property

Gets request headers for making requests.

get_block(block_id) async

Gets Notion block.

Parameters:

Name Type Description Default
block_id str

Id of the block to fetch.

 required
Source code in python_notion_api/async_api/api.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
async def get_block(self, block_id: str) -> NotionBlock:
    """Gets Notion block.

    Args:
        block_id: Id of the block to fetch.
    """
    block = NotionBlock(self, block_id)
    await block.reload()

    return block

get_database(database_id) async

Gets Notion database.

Parameters:

Name Type Description Default
database_id str

Id of the database to fetch.

 required

Returns:

Type Description
NotionDatabase

A Notion database with the given id.
Source code in python_notion_api/async_api/api.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
async def get_database(self, database_id: str) -> NotionDatabase:
    """Gets Notion database.

    Args:
        database_id: Id of the database to fetch.

    Returns:
        A Notion database with the given id.
    """
    database = NotionDatabase(self, database_id)
    await database.reload()

    return database

get_page(page_id, page_cast=NotionPage) async

Gets Notion page.

Parameters:

Name Type Description Default
page_id str

Id of the database to fetch.

 required
page_cast type[NotionPage]

A subclass of a NotionPage. Allows custom property retrieval.

 NotionPage

Returns: A Notion page with the given id.

Source code in python_notion_api/async_api/api.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def get_page(
    self, page_id: str, page_cast: type[NotionPage] = NotionPage
) -> NotionPage:
    """Gets Notion page.

    Args:
        page_id: Id of the database to fetch.
        page_cast: A subclass of a NotionPage. Allows custom
            property retrieval.
    Returns:
        A Notion page with the given id.
    """
    page = page_cast(api=self, page_id=page_id)
    await page.reload()

    return page

NotionBlock

Wrapper for a Notion block object

Parameters:

Name Type Description Default
api AsyncNotionAPI

Instance of the NotionAPI.

 required
block_id str

Id of the block.

 required
Source code in python_notion_api/async_api/notion_block.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class NotionBlock:
    """Wrapper for a Notion block object

    Args:
        api: Instance of the NotionAPI.
        block_id: Id of the block.
    """

    def __init__(self, api: "AsyncNotionAPI", block_id: str):
        self._api = api
        self._block_id = block_id
        self._object = None

    async def reload(self):
        """Reloads the block from Notion."""
        self._object = await self._api._get(endpoint=f"blocks/{self.block_id}")

    class AddChildrenRequest(BaseModel):
        children: List[Block]

    @ensure_loaded
    def __getattr__(self, attr_key):
        return getattr(self._object, attr_key)

    @property
    def block_id(self) -> str:
        """Gets the block id."""
        return self._block_id.replace("-", "")

    async def get_child_blocks(self) -> AsyncBlockIterator:
        """Gets all children blocks.

        Returns:
            An iterator of all children blocks in the block.
        """
        generator = await self._api._get_iterate(
            endpoint=f"blocks/{self._block_id}/children"
        )
        return AsyncBlockIterator(generator)

    async def add_child_block(
        self, content: List[Block], reload_block: bool = False
    ) -> AsyncBlockIterator:
        """Adds new blocks as children.

        Args:
            content: Content of the new block.

        Returns:
            An iterator of the newly created blocks.
        """

        request = NotionBlock.AddChildrenRequest(children=content)

        data = request.json(
            by_alias=True, exclude_unset=True, exclude_none=True
        )

        new_blocks = await self._api._patch(
            endpoint=f"blocks/{self.block_id}/children", data=data
        )

        if reload_block:
            await self.reload()

        return AsyncBlockIterator(iter(new_blocks.results))

    async def set(self, block: Block, reload_block: bool = False) -> Block:
        """Updates the content of a Block.

        The entire content is replaced.

        Args:
            block: Block with the new values.

        Returns:
            The updated block.
        """

        data = block.patch_json()

        new_block = await self._api._patch(
            endpoint=f"blocks/{self.block_id}", data=data
        )

        if reload_block:
            await self.reload()

        return new_block

block_id property

Gets the block id.

add_child_block(content, reload_block=False) async

Adds new blocks as children.

Parameters:

Name Type Description Default
content List[Block]

Content of the new block.

 required

Returns:

Type Description
AsyncBlockIterator

An iterator of the newly created blocks.
Source code in python_notion_api/async_api/notion_block.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
async def add_child_block(
    self, content: List[Block], reload_block: bool = False
) -> AsyncBlockIterator:
    """Adds new blocks as children.

    Args:
        content: Content of the new block.

    Returns:
        An iterator of the newly created blocks.
    """

    request = NotionBlock.AddChildrenRequest(children=content)

    data = request.json(
        by_alias=True, exclude_unset=True, exclude_none=True
    )

    new_blocks = await self._api._patch(
        endpoint=f"blocks/{self.block_id}/children", data=data
    )

    if reload_block:
        await self.reload()

    return AsyncBlockIterator(iter(new_blocks.results))

get_child_blocks() async

Gets all children blocks.

Returns:

Type Description
AsyncBlockIterator

An iterator of all children blocks in the block.
Source code in python_notion_api/async_api/notion_block.py 
42
43
44
45
46
47
48
49
50
51
async def get_child_blocks(self) -> AsyncBlockIterator:
    """Gets all children blocks.

    Returns:
        An iterator of all children blocks in the block.
    """
    generator = await self._api._get_iterate(
        endpoint=f"blocks/{self._block_id}/children"
    )
    return AsyncBlockIterator(generator)

reload() async

Reloads the block from Notion.

Source code in python_notion_api/async_api/notion_block.py 
26
27
28
async def reload(self):
    """Reloads the block from Notion."""
    self._object = await self._api._get(endpoint=f"blocks/{self.block_id}")

set(block, reload_block=False) async

Updates the content of a Block.

The entire content is replaced.

Parameters:

Name Type Description Default
block Block

Block with the new values.

 required

Returns:

Type Description
Block

The updated block.
Source code in python_notion_api/async_api/notion_block.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def set(self, block: Block, reload_block: bool = False) -> Block:
    """Updates the content of a Block.

    The entire content is replaced.

    Args:
        block: Block with the new values.

    Returns:
        The updated block.
    """

    data = block.patch_json()

    new_block = await self._api._patch(
        endpoint=f"blocks/{self.block_id}", data=data
    )

    if reload_block:
        await self.reload()

    return new_block

NotionDatabase

Wrapper for a Notion database object.

Parameters:

Name Type Description Default
api AsyncNotionAPI

Instance of the NotionAPI.

 required
database_id str

Id of the database.

 required
Source code in python_notion_api/async_api/notion_database.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
class NotionDatabase:
    """Wrapper for a Notion database object.

    Args:
        api: Instance of the NotionAPI.
        database_id: Id of the database.
    """

    class CreatePageRequest(BaseModel):
        parent: ParentObject
        properties: Dict[str, PropertyValue]
        cover: Optional[FileObject]

    def __init__(self, api: "AsyncNotionAPI", database_id: str):
        self._api = api
        self._database_id = database_id
        self._object = None
        self._properties = None
        self._title = None

    @ensure_loaded
    def __getattr__(self, attr_key):
        return getattr(self._object, attr_key)

    @property
    def database_id(self) -> str:
        """Gets the database id."""
        return self._database_id.replace("-", "")

    async def reload(self):
        """Reloads the database from Notion."""
        self._object = await self._api._get(
            endpoint=f"databases/{self._database_id}", cast_cls=Database
        )

        if self._object is None:
            raise Exception(f"Error loading database {self._database_id}")

        self._properties = {
            key: NotionPropertyConfiguration.from_obj(val)
            for key, val in self._object.properties.items()
        }
        self._title = "".join(rt.plain_text for rt in self._object.title)

    async def query(
        self,
        filters: Optional[FilterItem] = None,
        sorts: Optional[List[Sort]] = None,
        page_limit: Optional[int] = None,
        cast_cls=NotionPage,
    ) -> AsyncGenerator[NotionPage, None]:
        """Queries the database.

        Retrieves all pages belonging to the database that satisfy the given filters
        in the order specified by the sorts.

        Args:
            filters: Filters to apply to the query.
            sorts: Sorts to apply to the query.
            cast_cls: A subclass of a NotionPage. Allows custom
            property retrieval.

        Returns:
            Generator of NotionPage objects.
        """
        data: dict[str, Any] = {}

        if filters is not None:
            filters = filters.dict(by_alias=True, exclude_unset=True)
            data["filter"] = filters

        if sorts is not None:
            data["sorts"] = [
                sort.dict(by_alias=True, exclude_unset=True) for sort in sorts
            ]

        async for item in self._api._post_iterate(
            endpoint=f"databases/{self._database_id}/query",
            data=data,
            page_limit=page_limit,
        ):
            yield cast_cls(
                api=self._api, database=self, page_id=item.page_id, obj=item
            )

    @property
    @ensure_loaded
    def title(self) -> str:
        """Gets title of the database."""
        assert self._title is not None
        return self._title

    @property
    @ensure_loaded
    def properties(self) -> Dict[str, NotionPropertyConfiguration]:
        """Gets all property configurations of the database."""
        assert self._properties is not None
        return self._properties

    @property
    @ensure_loaded
    def relations(self) -> Dict[str, RelationPropertyConfiguration]:
        """Gets all property configurations of the database that are
        relations.
        """
        assert self._properties is not None
        return {
            key: val
            for key, val in self._properties.items()
            if isinstance(val, RelationPropertyConfiguration)
        }

    async def create_page(
        self,
        properties: Dict[str, Any] = {},
        cover_url: Optional[str] = None,
    ) -> NotionPage:
        """Creates a new page in the Database and updates the new page with
        the properties.

        Args:
            properties: Dictionary of property names and values. Value types
            will depend on the property type. Can be the raw value
            (e.g. string, float) or an object (e.g. SelectValue,
            NumberPropertyItem)
            cover: URL of an image for the page cover.

        Returns:
            A new page.
        """

        validated_properties = {}
        for prop_name, prop_value in properties.items():
            prop = self.properties.get(prop_name, None)
            if prop is None:
                raise ValueError(f"Unknown property: {prop_name}")
            value = generate_value(prop.config_type, prop_value)
            validated_properties[prop_name] = value

        request = NotionDatabase.CreatePageRequest(
            parent=ParentObject(
                type="database_id", database_id=self.database_id
            ),
            properties=validated_properties,
            cover=(
                FileObject.from_url(cover_url)
                if cover_url is not None
                else None
            ),
        )

        data = request.json(by_alias=True, exclude_unset=True)

        new_page = await self._api._post("pages", data=data)

        return NotionPage(
            api=self._api,
            page_id=new_page.page_id,
            obj=new_page,
            database=self,
        )

database_id property

Gets the database id.

properties property

Gets all property configurations of the database.

relations property

Gets all property configurations of the database that are relations.

title property

Gets title of the database.

create_page(properties={}, cover_url=None) async

Creates a new page in the Database and updates the new page with the properties.

Parameters:

Name Type Description Default
properties Dict[str, Any]

Dictionary of property names and values. Value types

 {}
cover

URL of an image for the page cover.

 required

Returns:

Type Description
NotionPage

A new page.
Source code in python_notion_api/async_api/notion_database.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
async def create_page(
    self,
    properties: Dict[str, Any] = {},
    cover_url: Optional[str] = None,
) -> NotionPage:
    """Creates a new page in the Database and updates the new page with
    the properties.

    Args:
        properties: Dictionary of property names and values. Value types
        will depend on the property type. Can be the raw value
        (e.g. string, float) or an object (e.g. SelectValue,
        NumberPropertyItem)
        cover: URL of an image for the page cover.

    Returns:
        A new page.
    """

    validated_properties = {}
    for prop_name, prop_value in properties.items():
        prop = self.properties.get(prop_name, None)
        if prop is None:
            raise ValueError(f"Unknown property: {prop_name}")
        value = generate_value(prop.config_type, prop_value)
        validated_properties[prop_name] = value

    request = NotionDatabase.CreatePageRequest(
        parent=ParentObject(
            type="database_id", database_id=self.database_id
        ),
        properties=validated_properties,
        cover=(
            FileObject.from_url(cover_url)
            if cover_url is not None
            else None
        ),
    )

    data = request.json(by_alias=True, exclude_unset=True)

    new_page = await self._api._post("pages", data=data)

    return NotionPage(
        api=self._api,
        page_id=new_page.page_id,
        obj=new_page,
        database=self,
    )

query(filters=None, sorts=None, page_limit=None, cast_cls=NotionPage) async

Queries the database.

Retrieves all pages belonging to the database that satisfy the given filters in the order specified by the sorts.

Parameters:

Name Type Description Default
filters Optional[FilterItem]

Filters to apply to the query.

 None
sorts Optional[List[Sort]]

Sorts to apply to the query.

 None
cast_cls

A subclass of a NotionPage. Allows custom

 NotionPage

Returns:

Type Description
AsyncGenerator[NotionPage, None]

Generator of NotionPage objects.
Source code in python_notion_api/async_api/notion_database.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def query(
    self,
    filters: Optional[FilterItem] = None,
    sorts: Optional[List[Sort]] = None,
    page_limit: Optional[int] = None,
    cast_cls=NotionPage,
) -> AsyncGenerator[NotionPage, None]:
    """Queries the database.

    Retrieves all pages belonging to the database that satisfy the given filters
    in the order specified by the sorts.

    Args:
        filters: Filters to apply to the query.
        sorts: Sorts to apply to the query.
        cast_cls: A subclass of a NotionPage. Allows custom
        property retrieval.

    Returns:
        Generator of NotionPage objects.
    """
    data: dict[str, Any] = {}

    if filters is not None:
        filters = filters.dict(by_alias=True, exclude_unset=True)
        data["filter"] = filters

    if sorts is not None:
        data["sorts"] = [
            sort.dict(by_alias=True, exclude_unset=True) for sort in sorts
        ]

    async for item in self._api._post_iterate(
        endpoint=f"databases/{self._database_id}/query",
        data=data,
        page_limit=page_limit,
    ):
        yield cast_cls(
            api=self._api, database=self, page_id=item.page_id, obj=item
        )

reload() async

Reloads the database from Notion.

Source code in python_notion_api/async_api/notion_database.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
async def reload(self):
    """Reloads the database from Notion."""
    self._object = await self._api._get(
        endpoint=f"databases/{self._database_id}", cast_cls=Database
    )

    if self._object is None:
        raise Exception(f"Error loading database {self._database_id}")

    self._properties = {
        key: NotionPropertyConfiguration.from_obj(val)
        for key, val in self._object.properties.items()
    }
    self._title = "".join(rt.plain_text for rt in self._object.title)

NotionPage

Wrapper for a Notion page object.

Parameters:

Name Type Description Default
api AsyncNotionAPI

Instance of the NotionAPI.

 required
page_id str

Id of the page.

 required
Source code in python_notion_api/async_api/notion_page.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
class NotionPage:
    """Wrapper for a Notion page object.

    Args:
        api: Instance of the NotionAPI.
        page_id: Id of the page.
    """

    class PatchRequest(BaseModel):
        properties: dict[str, PropertyValue]

    class AddBlocksRequest(BaseModel):
        children: list[Block]

    # Map from property names to function names.
    # For use in subclasses
    special_properties: dict[str, str] = {}

    def __init__(
        self,
        api: "AsyncNotionAPI",
        page_id: str,
        obj: Optional[Page] = None,
        database: Optional[Database] = None,
    ):
        self._api = api
        self._page_id = page_id
        self._object = obj
        self.database = database

    async def reload(self):
        """Reloads page from Notion."""
        self._object = await self._api._get(endpoint=f"pages/{self._page_id}")
        if self._object is not None:
            parent_id = self.parent.database_id
            if parent_id is not None:
                self.database = await self._api.get_database(parent_id)

    @ensure_loaded
    def __getattr__(self, attr_key: str):
        return getattr(self._object, attr_key)

    @property
    def page_id(self) -> str:
        """Returns the page id."""
        return self._page_id.replace("-", "")

    @property
    @ensure_loaded
    def is_alive(self) -> bool:
        """Checks if the page is archived.

        Returns:
            `True` if the page is not archived, False otherwise.
        """
        assert self._object is not None
        return not self._object.archived

    async def archive(self):
        """Archives the page"""
        await self._archive(True)

    async def unarchive(self):
        """Unarchives the page"""
        await self._archive(False)

    async def _archive(self, archive_status=True) -> None:
        """Changes archive status of the page.

        Args:
            archive_status: Whether to archive or unarchive the page.
        """
        await self._api._patch(
            endpoint=f"pages/{self._page_id}",
            data=json.dumps({"archived": archive_status}),
        )

    @ensure_loaded
    async def set(
        self, prop_key: str, value: Any, reload_page: bool = False
    ) -> None:
        """Sets a single page property.

        Args:
            prop_key: Name or id of the property to update.
            value: A new value of the property.
            reload_page: Whether to reload the page after updating the property.
        """

        prop_name = self._get_prop_name(prop_key=prop_key)

        if prop_name is None:
            raise ValueError(f"Unknown property '{prop_name}'")

        assert self._object is not None

        prop_type = self._object.properties[prop_name]["type"]

        value = generate_value(prop_type, value)
        request = NotionPage.PatchRequest(properties={prop_name: value})

        data = request.json(by_alias=True, exclude_unset=True)

        await self._api._patch(endpoint=f"pages/{self._page_id}", data=data)

        if reload_page:
            await self.reload()

    @ensure_loaded
    async def update(
        self, properties: dict[str, Any], reload_page: bool = False
    ) -> None:
        """Updates the page with a dictionary of new values.

        Args:
            properties: A dictionary mapping property keys to new
                values.
            reload_page: Whether to reload the page after updating the properties.
        """
        values = {}
        for prop_key, value in properties.items():
            prop_name = self._get_prop_name(prop_key=prop_key)

            if prop_name is None:
                raise ValueError(f"Unknown property '{prop_name}'")

            assert self._object is not None

            prop_type = self._object.properties[prop_name]["type"]

            value = generate_value(prop_type, value)
            values[prop_name] = value

        request = NotionPage.PatchRequest(properties=values)

        data = request.json(by_alias=True, exclude_unset=True)

        await self._api._patch(endpoint=f"pages/{self._page_id}", data=data)

        if reload_page:
            await self.reload()

    @ensure_loaded
    async def get_properties(
        self, raw: bool = False
    ) -> dict[str, PropertyValue]:
        """Gets all properties of the page."""
        assert self._object is not None
        return {
            prop_name: await self.get(prop_name, raw=raw)
            for prop_name in self._object.properties
        }

    @ensure_loaded
    async def to_dict(
        self,
        include_rels: bool = True,
        rels_only=False,
        properties: Optional[dict] = None,
    ) -> dict[str, Union[str, list]]:
        """ "Returns all properties of the page as a dict of builtin type values.

        Args:
            include_rels: Include relations.
            rels_only: Return relations only.
            properties: List of properties to return. If `None`, will
                get values for all properties.
        """
        if properties is None:
            assert self._object is not None
            properties = self._object.properties
        vals = {}

        for prop_name in properties:
            prop = await self.get(prop_name, raw=True)

            if isinstance(prop, AsyncPropertyItemIterator):
                value = await prop.get_value()
            else:
                value = prop.value

            if prop.property_type == "relation":
                if include_rels:
                    vals[prop_name] = value
            else:
                if not rels_only:
                    vals[prop_name] = value
        return vals

    async def add_blocks(self, blocks: list[Block]) -> AsyncBlockIterator:
        """Adds new blocks to the page.

        Args:
            blocks: List of Blocks to add.

        Returns:
            Iterator of new blocks.
        """
        request = NotionPage.AddBlocksRequest(children=blocks)

        data = request.json(
            by_alias=True, exclude_unset=True, exclude_none=True
        )

        new_blocks = await self._api._patch(
            endpoint=f"blocks/{self.page_id}/children", data=data
        )
        return AsyncBlockIterator(iter(new_blocks.results))

    async def get_blocks(self) -> AsyncBlockIterator:
        """Gets all blocks in the page.

        Returns:
            Iterator of blocks is returned.
        """

        generator = self._api._get_iterate(
            endpoint=f"blocks/{self._page_id}/children"
        )
        return AsyncBlockIterator(generator)

    @ensure_loaded
    async def get(
        self,
        prop_key: str,
        cache: bool = True,
        safety_off: bool = False,
        raw: bool = False,
    ) -> Union[PropertyValue, AsyncPropertyItemIterator, None]:
        """Gets a single page property.

        First checks if the property is 'special', if so, will call the special
        function to get that property value.
        If not, gets the property through the api.

        Args:
            prop_key: Name or id of the property to retrieve.
            cache: If `True` and the property has been retrieved before, will return a cached value.
                Use `False` to force a new API call.
            safety_off: If `True` will use cached values of rollups and
                formulas.
        """
        if prop_key in self.special_properties:
            # For subclasses of NotionPage
            # Any special properties should have an associated function
            # in the subclass, and a mapping from the property name
            # to the function name in self.special_properties
            # Those functions must return PropertyItemIterator or PropertyItem
            attr = getattr(self, self.special_properties[prop_key])()
            assert isinstance(attr, PropertyValue)
            property_value = attr
        else:
            property_value = await self._direct_get(
                prop_key=prop_key, cache=cache, safety_off=safety_off
            )

        if raw:
            return property_value
        else:
            if isinstance(property_value, AsyncPropertyItemIterator):
                return await property_value.get_value()
            return property_value.value

    async def _direct_get(
        self, prop_key: str, cache: bool = True, safety_off: bool = False
    ) -> Union[PropertyValue, AsyncPropertyItemIterator, None]:
        """Wrapper for 'Retrieve a page property item' action.

        Will return whatever is retrieved from the API, no special cases.

        Args:
            prop_key: Name or id of the property to retrieve.
            cache: Boolean to decide whether to return the info from the page
                or query the API again.
            safety_off: If `True` will use cached values of rollups and
                formulas
        """
        prop_name = self._get_prop_name(prop_key)

        if prop_name is None:
            raise ValueError(f"Invalid property key '{prop_key}'")

        assert self._object is not None

        prop = self._object.properties[prop_name]

        obj = PropertyItem.from_obj(prop)

        prop_id = obj.property_id
        prop_type = obj.property_type

        # We need to always query the API for formulas and rollups as
        # otherwise we might get incorrect values.
        if not safety_off and prop_type in ("formula", "rollup"):
            cache = False

        if cache and not obj.has_more:
            return PropertyValue.from_property_item(obj)

        ret = await self._api._get(
            endpoint=f"pages/{self._page_id}/properties/{prop_id}",
            params={"page_size": 20},
        )

        if isinstance(ret, Pagination):
            generator = self._api._get_iterate(
                endpoint=f"pages/{self._page_id}/properties/{prop_id}"
            )
            return create_property_iterator(generator, obj)

        elif isinstance(ret, PropertyItem):
            return PropertyValue.from_property_item(ret)
        else:
            return None

    def _get_prop_name(self, prop_key: str) -> Optional[str]:
        """Gets propetry name from property key.

        Args:
            prop_key: Either a property name or property id.

        Returns:
            Property name or `None` if key is invalid.
        """
        assert self._object is not None
        _properties = self._object.properties
        prop_name = next(
            (
                key
                for key in _properties
                if key == prop_key or _properties[key]["id"] == prop_key
            ),
            None,
        )

        return prop_name

is_alive property

Checks if the page is archived.

Returns:

Type Description
bool

True if the page is not archived, False otherwise.

page_id property

Returns the page id.

add_blocks(blocks) async

Adds new blocks to the page.

Parameters:

Name Type Description Default
blocks list[Block]

List of Blocks to add.

 required

Returns:

Type Description
AsyncBlockIterator

Iterator of new blocks.
Source code in python_notion_api/async_api/notion_page.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
async def add_blocks(self, blocks: list[Block]) -> AsyncBlockIterator:
    """Adds new blocks to the page.

    Args:
        blocks: List of Blocks to add.

    Returns:
        Iterator of new blocks.
    """
    request = NotionPage.AddBlocksRequest(children=blocks)

    data = request.json(
        by_alias=True, exclude_unset=True, exclude_none=True
    )

    new_blocks = await self._api._patch(
        endpoint=f"blocks/{self.page_id}/children", data=data
    )
    return AsyncBlockIterator(iter(new_blocks.results))

archive() async

Archives the page

Source code in python_notion_api/async_api/notion_page.py 
78
79
80
async def archive(self):
    """Archives the page"""
    await self._archive(True)

get(prop_key, cache=True, safety_off=False, raw=False) async

Gets a single page property.

First checks if the property is 'special', if so, will call the special function to get that property value. If not, gets the property through the api.

Parameters:

Name Type Description Default
prop_key str

Name or id of the property to retrieve.

 required
cache bool

If True and the property has been retrieved before, will return a cached value. Use False to force a new API call.

 True
safety_off bool

If True will use cached values of rollups and formulas.

 False
Source code in python_notion_api/async_api/notion_page.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
@ensure_loaded
async def get(
    self,
    prop_key: str,
    cache: bool = True,
    safety_off: bool = False,
    raw: bool = False,
) -> Union[PropertyValue, AsyncPropertyItemIterator, None]:
    """Gets a single page property.

    First checks if the property is 'special', if so, will call the special
    function to get that property value.
    If not, gets the property through the api.

    Args:
        prop_key: Name or id of the property to retrieve.
        cache: If `True` and the property has been retrieved before, will return a cached value.
            Use `False` to force a new API call.
        safety_off: If `True` will use cached values of rollups and
            formulas.
    """
    if prop_key in self.special_properties:
        # For subclasses of NotionPage
        # Any special properties should have an associated function
        # in the subclass, and a mapping from the property name
        # to the function name in self.special_properties
        # Those functions must return PropertyItemIterator or PropertyItem
        attr = getattr(self, self.special_properties[prop_key])()
        assert isinstance(attr, PropertyValue)
        property_value = attr
    else:
        property_value = await self._direct_get(
            prop_key=prop_key, cache=cache, safety_off=safety_off
        )

    if raw:
        return property_value
    else:
        if isinstance(property_value, AsyncPropertyItemIterator):
            return await property_value.get_value()
        return property_value.value

get_blocks() async

Gets all blocks in the page.

Returns:

Type Description
AsyncBlockIterator

Iterator of blocks is returned.
Source code in python_notion_api/async_api/notion_page.py 
229
230
231
232
233
234
235
236
237
238
239
async def get_blocks(self) -> AsyncBlockIterator:
    """Gets all blocks in the page.

    Returns:
        Iterator of blocks is returned.
    """

    generator = self._api._get_iterate(
        endpoint=f"blocks/{self._page_id}/children"
    )
    return AsyncBlockIterator(generator)

get_properties(raw=False) async

Gets all properties of the page.

Source code in python_notion_api/async_api/notion_page.py 
162
163
164
165
166
167
168
169
170
171
@ensure_loaded
async def get_properties(
    self, raw: bool = False
) -> dict[str, PropertyValue]:
    """Gets all properties of the page."""
    assert self._object is not None
    return {
        prop_name: await self.get(prop_name, raw=raw)
        for prop_name in self._object.properties
    }

reload() async

Reloads page from Notion.

Source code in python_notion_api/async_api/notion_page.py 
50
51
52
53
54
55
56
async def reload(self):
    """Reloads page from Notion."""
    self._object = await self._api._get(endpoint=f"pages/{self._page_id}")
    if self._object is not None:
        parent_id = self.parent.database_id
        if parent_id is not None:
            self.database = await self._api.get_database(parent_id)

set(prop_key, value, reload_page=False) async

Sets a single page property.

Parameters:

Name Type Description Default
prop_key str

Name or id of the property to update.

 required
value Any

A new value of the property.

 required
reload_page bool

Whether to reload the page after updating the property.

 False
Source code in python_notion_api/async_api/notion_page.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@ensure_loaded
async def set(
    self, prop_key: str, value: Any, reload_page: bool = False
) -> None:
    """Sets a single page property.

    Args:
        prop_key: Name or id of the property to update.
        value: A new value of the property.
        reload_page: Whether to reload the page after updating the property.
    """

    prop_name = self._get_prop_name(prop_key=prop_key)

    if prop_name is None:
        raise ValueError(f"Unknown property '{prop_name}'")

    assert self._object is not None

    prop_type = self._object.properties[prop_name]["type"]

    value = generate_value(prop_type, value)
    request = NotionPage.PatchRequest(properties={prop_name: value})

    data = request.json(by_alias=True, exclude_unset=True)

    await self._api._patch(endpoint=f"pages/{self._page_id}", data=data)

    if reload_page:
        await self.reload()

to_dict(include_rels=True, rels_only=False, properties=None) async

"Returns all properties of the page as a dict of builtin type values.

Parameters:

Name Type Description Default
include_rels bool

Include relations.

 True
rels_only

Return relations only.

 False
properties Optional[dict]

List of properties to return. If None, will get values for all properties.

 None
Source code in python_notion_api/async_api/notion_page.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@ensure_loaded
async def to_dict(
    self,
    include_rels: bool = True,
    rels_only=False,
    properties: Optional[dict] = None,
) -> dict[str, Union[str, list]]:
    """ "Returns all properties of the page as a dict of builtin type values.

    Args:
        include_rels: Include relations.
        rels_only: Return relations only.
        properties: List of properties to return. If `None`, will
            get values for all properties.
    """
    if properties is None:
        assert self._object is not None
        properties = self._object.properties
    vals = {}

    for prop_name in properties:
        prop = await self.get(prop_name, raw=True)

        if isinstance(prop, AsyncPropertyItemIterator):
            value = await prop.get_value()
        else:
            value = prop.value

        if prop.property_type == "relation":
            if include_rels:
                vals[prop_name] = value
        else:
            if not rels_only:
                vals[prop_name] = value
    return vals

unarchive() async

Unarchives the page

Source code in python_notion_api/async_api/notion_page.py 
82
83
84
async def unarchive(self):
    """Unarchives the page"""
    await self._archive(False)

update(properties, reload_page=False) async

Updates the page with a dictionary of new values.

Parameters:

Name Type Description Default
properties dict[str, Any]

A dictionary mapping property keys to new values.

 required
reload_page bool

Whether to reload the page after updating the properties.

 False
Source code in python_notion_api/async_api/notion_page.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@ensure_loaded
async def update(
    self, properties: dict[str, Any], reload_page: bool = False
) -> None:
    """Updates the page with a dictionary of new values.

    Args:
        properties: A dictionary mapping property keys to new
            values.
        reload_page: Whether to reload the page after updating the properties.
    """
    values = {}
    for prop_key, value in properties.items():
        prop_name = self._get_prop_name(prop_key=prop_key)

        if prop_name is None:
            raise ValueError(f"Unknown property '{prop_name}'")

        assert self._object is not None

        prop_type = self._object.properties[prop_name]["type"]

        value = generate_value(prop_type, value)
        values[prop_name] = value

    request = NotionPage.PatchRequest(properties=values)

    data = request.json(by_alias=True, exclude_unset=True)

    await self._api._patch(endpoint=f"pages/{self._page_id}", data=data)

    if reload_page:
        await self.reload()