Skip to content

Session

Provides direct access to the Notion API.

BlocksEndpoint

Bases: Endpoint

Notional interface to the API 'blocks' endpoint.

Source code in src/notional/session.py 
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
class BlocksEndpoint(Endpoint):
    """Notional interface to the API 'blocks' endpoint."""

    class ChildrenEndpoint(Endpoint):
        """Notional interface to the API 'blocks/children' endpoint."""

        def __call__(self):
            """Return the underlying endpoint in the Notion SDK."""
            return self.session.client.blocks.children

        # https://developers.notion.com/reference/patch-block-children
        def append(self, parent, *blocks: Block, after: Block = None):
            """Add the given blocks as children of the specified parent.

            If `after` keyword is specified, they are appended after given Block.

            The blocks info will be refreshed based on returned data.

            `parent` may be any suitable `ObjectReference` type.
            """

            parent_id = ObjectReference[parent].id

            children = [block.dict() for block in blocks if block is not None]

            logger.info("Appending %d blocks to %s ...", len(children), parent_id)

            if after is None:
                data = self().append(block_id=parent_id, children=children)
            else:
                after_id = str(after.id) if isinstance(after, Block) else after
                data = self().append(
                    block_id=parent_id, children=children, after=after_id
                )

            if "results" in data:
                # in case of `after`, there is second result
                if len(blocks) == len(data["results"]) or after is not None:
                    for idx in range(len(blocks)):
                        block = blocks[idx]
                        result = data["results"][idx]
                        block.refresh(**result)

                else:
                    logger.warning("Unable to refresh results; size mismatch")

            else:
                logger.warning("Unable to refresh results; not provided")

            return parent

        # https://developers.notion.com/reference/get-block-children
        def list(self, parent):
            """Return all Blocks contained by the specified parent.

            `parent` may be any suitable `ObjectReference` type.
            """

            parent_id = ObjectReference[parent].id

            logger.info("Listing blocks for %s...", parent_id)

            blocks = EndpointIterator(endpoint=self().list)

            return blocks(block_id=parent_id)

    def __init__(self, *args, **kwargs):
        """Initialize the `blocks` endpoint for the Notion API."""
        super().__init__(*args, **kwargs)

        self.children = BlocksEndpoint.ChildrenEndpoint(*args, **kwargs)

    def __call__(self):
        """Return the underlying endpoint in the Notion SDK."""
        return self.session.client.blocks

    # https://developers.notion.com/reference/delete-a-block
    def delete(self, block):
        """Delete (archive) the specified Block.

        `block` may be any suitable `ObjectReference` type.
        """

        block_id = ObjectReference[block].id
        logger.info("Deleting block :: %s", block_id)

        data = self().delete(block_id)

        return Block.parse_obj(data)

    def restore(self, block):
        """Restore (unarchive) the specified Block.

        `block` may be any suitable `ObjectReference` type.
        """

        block_id = ObjectReference[block].id
        logger.info("Restoring block :: %s", block_id)

        data = self().update(block_id, archived=False)

        return Block.parse_obj(data)

    # https://developers.notion.com/reference/retrieve-a-block
    def retrieve(self, block):
        """Return the requested Block.

        `block` may be any suitable `ObjectReference` type.
        """

        block_id = ObjectReference[block].id
        logger.info("Retrieving block :: %s", block_id)

        data = self().retrieve(block_id)

        return Block.parse_obj(data)

    # https://developers.notion.com/reference/update-a-block
    def update(self, block: Block):
        """Update the block content on the server.

        The block info will be refreshed to the latest version from the server.
        """

        logger.info("Updating block :: %s", block.id)

        data = self().update(block.id.hex, **block.dict())

        return block.refresh(**data)

ChildrenEndpoint

Bases: Endpoint

Notional interface to the API 'blocks/children' endpoint.

Source code in src/notional/session.py 
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
class ChildrenEndpoint(Endpoint):
    """Notional interface to the API 'blocks/children' endpoint."""

    def __call__(self):
        """Return the underlying endpoint in the Notion SDK."""
        return self.session.client.blocks.children

    # https://developers.notion.com/reference/patch-block-children
    def append(self, parent, *blocks: Block, after: Block = None):
        """Add the given blocks as children of the specified parent.

        If `after` keyword is specified, they are appended after given Block.

        The blocks info will be refreshed based on returned data.

        `parent` may be any suitable `ObjectReference` type.
        """

        parent_id = ObjectReference[parent].id

        children = [block.dict() for block in blocks if block is not None]

        logger.info("Appending %d blocks to %s ...", len(children), parent_id)

        if after is None:
            data = self().append(block_id=parent_id, children=children)
        else:
            after_id = str(after.id) if isinstance(after, Block) else after
            data = self().append(
                block_id=parent_id, children=children, after=after_id
            )

        if "results" in data:
            # in case of `after`, there is second result
            if len(blocks) == len(data["results"]) or after is not None:
                for idx in range(len(blocks)):
                    block = blocks[idx]
                    result = data["results"][idx]
                    block.refresh(**result)

            else:
                logger.warning("Unable to refresh results; size mismatch")

        else:
            logger.warning("Unable to refresh results; not provided")

        return parent

    # https://developers.notion.com/reference/get-block-children
    def list(self, parent):
        """Return all Blocks contained by the specified parent.

        `parent` may be any suitable `ObjectReference` type.
        """

        parent_id = ObjectReference[parent].id

        logger.info("Listing blocks for %s...", parent_id)

        blocks = EndpointIterator(endpoint=self().list)

        return blocks(block_id=parent_id)

__call__()

Return the underlying endpoint in the Notion SDK.

Source code in src/notional/session.py 
114
115
116
def __call__(self):
    """Return the underlying endpoint in the Notion SDK."""
    return self.session.client.blocks.children

append(parent, *blocks, after=None)

Add the given blocks as children of the specified parent.

If after keyword is specified, they are appended after given Block.

The blocks info will be refreshed based on returned data.

parent may be any suitable ObjectReference type.

Source code in src/notional/session.py 
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
def append(self, parent, *blocks: Block, after: Block = None):
    """Add the given blocks as children of the specified parent.

    If `after` keyword is specified, they are appended after given Block.

    The blocks info will be refreshed based on returned data.

    `parent` may be any suitable `ObjectReference` type.
    """

    parent_id = ObjectReference[parent].id

    children = [block.dict() for block in blocks if block is not None]

    logger.info("Appending %d blocks to %s ...", len(children), parent_id)

    if after is None:
        data = self().append(block_id=parent_id, children=children)
    else:
        after_id = str(after.id) if isinstance(after, Block) else after
        data = self().append(
            block_id=parent_id, children=children, after=after_id
        )

    if "results" in data:
        # in case of `after`, there is second result
        if len(blocks) == len(data["results"]) or after is not None:
            for idx in range(len(blocks)):
                block = blocks[idx]
                result = data["results"][idx]
                block.refresh(**result)

        else:
            logger.warning("Unable to refresh results; size mismatch")

    else:
        logger.warning("Unable to refresh results; not provided")

    return parent

list(parent)

Return all Blocks contained by the specified parent.

parent may be any suitable ObjectReference type.

Source code in src/notional/session.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
def list(self, parent):
    """Return all Blocks contained by the specified parent.

    `parent` may be any suitable `ObjectReference` type.
    """

    parent_id = ObjectReference[parent].id

    logger.info("Listing blocks for %s...", parent_id)

    blocks = EndpointIterator(endpoint=self().list)

    return blocks(block_id=parent_id)

__call__()

Return the underlying endpoint in the Notion SDK.

Source code in src/notional/session.py 
180
181
182
def __call__(self):
    """Return the underlying endpoint in the Notion SDK."""
    return self.session.client.blocks

__init__(*args, **kwargs)

Initialize the blocks endpoint for the Notion API.

Source code in src/notional/session.py 
174
175
176
177
178
def __init__(self, *args, **kwargs):
    """Initialize the `blocks` endpoint for the Notion API."""
    super().__init__(*args, **kwargs)

    self.children = BlocksEndpoint.ChildrenEndpoint(*args, **kwargs)

delete(block)

Delete (archive) the specified Block.

block may be any suitable ObjectReference type.

Source code in src/notional/session.py 
185
186
187
188
189
190
191
192
193
194
195
196
def delete(self, block):
    """Delete (archive) the specified Block.

    `block` may be any suitable `ObjectReference` type.
    """

    block_id = ObjectReference[block].id
    logger.info("Deleting block :: %s", block_id)

    data = self().delete(block_id)

    return Block.parse_obj(data)

restore(block)

Restore (unarchive) the specified Block.

block may be any suitable ObjectReference type.

Source code in src/notional/session.py 
198
199
200
201
202
203
204
205
206
207
208
209
def restore(self, block):
    """Restore (unarchive) the specified Block.

    `block` may be any suitable `ObjectReference` type.
    """

    block_id = ObjectReference[block].id
    logger.info("Restoring block :: %s", block_id)

    data = self().update(block_id, archived=False)

    return Block.parse_obj(data)

retrieve(block)

Return the requested Block.

block may be any suitable ObjectReference type.

Source code in src/notional/session.py 
212
213
214
215
216
217
218
219
220
221
222
223
def retrieve(self, block):
    """Return the requested Block.

    `block` may be any suitable `ObjectReference` type.
    """

    block_id = ObjectReference[block].id
    logger.info("Retrieving block :: %s", block_id)

    data = self().retrieve(block_id)

    return Block.parse_obj(data)

update(block)

Update the block content on the server.

The block info will be refreshed to the latest version from the server.

Source code in src/notional/session.py 
226
227
228
229
230
231
232
233
234
235
236
def update(self, block: Block):
    """Update the block content on the server.

    The block info will be refreshed to the latest version from the server.
    """

    logger.info("Updating block :: %s", block.id)

    data = self().update(block.id.hex, **block.dict())

    return block.refresh(**data)

DatabasesEndpoint

Bases: Endpoint

Notional interface to the API 'databases' endpoint.

Source code in src/notional/session.py 
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
class DatabasesEndpoint(Endpoint):
    """Notional interface to the API 'databases' endpoint."""

    def __call__(self):
        """Return the underlying endpoint in the Notion SDK."""
        return self.session.client.databases

    def _build_request(
        self,
        parent: ParentRef = None,
        schema: Dict[str, PropertyObject] = None,
        title=None,
    ):
        """Build a request payload from the given items.

        *NOTE* this method does not anticipate what the request will be used for and as
        such does not validate the inputs for any particular requests.
        """
        request = {}

        if parent is not None:
            request["parent"] = parent.dict()

        if title is not None:
            prop = TextObject[title]
            request["title"] = [prop.dict()]

        if schema is not None:
            request["properties"] = {
                name: value.dict() if value is not None else None
                for name, value in schema.items()
            }

        return request

    # https://developers.notion.com/reference/create-a-database
    def create(self, parent, schema: Dict[str, PropertyObject], title=None):
        """Add a database to the given Page parent.

        `parent` may be any suitable `PageRef` type.
        """

        parent_ref = PageRef[parent]

        logger.info("Creating database @ %s - %s", parent_ref.page_id, title)

        request = self._build_request(parent_ref, schema, title)

        data = self().create(**request)

        return Database.parse_obj(data)

    # https://developers.notion.com/reference/retrieve-a-database
    def retrieve(self, dbref):
        """Return the Database with the given ID.

        `dbref` may be any suitable `DatabaseRef` type.
        """

        dbid = DatabaseRef[dbref].database_id

        logger.info("Retrieving database :: %s", dbid)

        data = self().retrieve(dbid)

        return Database.parse_obj(data)

    # https://developers.notion.com/reference/update-a-database
    def update(self, dbref, title=None, schema: Dict[str, PropertyObject] = None):
        """Update the Database object on the server.

        The database info will be refreshed to the latest version from the server.

        `dbref` may be any suitable `DatabaseRef` type.
        """

        dbid = DatabaseRef[dbref].database_id

        logger.info("Updating database info :: %s", dbid)

        request = self._build_request(schema=schema, title=title)

        if request:
            data = self().update(dbid, **request)
            dbref = dbref.refresh(**data)

        return dbref

    def delete(self, dbref):
        """Delete (archive) the specified Database.

        `dbref` may be any suitable `DatabaseRef` type.
        """

        dbid = DatabaseRef[dbref].database_id

        logger.info("Deleting database :: %s", dbid)

        return self.session.blocks.delete(dbid)

    def restore(self, dbref):
        """Restore (unarchive) the specified Database.

        `dbref` may be any suitable `DatabaseRef` type.
        """

        dbid = DatabaseRef[dbref].database_id

        logger.info("Restoring database :: %s", dbid)

        return self.session.blocks.restore(dbid)

    # https://developers.notion.com/reference/post-database-query
    def query(self, target):
        """Initialize a new Query object with the target data class.

        :param target: either a `DatabaseRef` type or an ORM class
        """

        if isclass(target) and issubclass(target, ConnectedPage):
            cls = target
            dbid = target._notional__database

            if cls._notional__session != self.session:
                raise ValueError("ConnectedPage belongs to a different session")

        else:
            cls = None
            dbid = DatabaseRef[target].database_id

        logger.info("Initializing database query :: {%s} [%s]", dbid, cls)

        return QueryBuilder(endpoint=self().query, datatype=cls, database_id=dbid)

__call__()

Return the underlying endpoint in the Notion SDK.

Source code in src/notional/session.py 
242
243
244
def __call__(self):
    """Return the underlying endpoint in the Notion SDK."""
    return self.session.client.databases

create(parent, schema, title=None)

Add a database to the given Page parent.

parent may be any suitable PageRef type.

Source code in src/notional/session.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def create(self, parent, schema: Dict[str, PropertyObject], title=None):
    """Add a database to the given Page parent.

    `parent` may be any suitable `PageRef` type.
    """

    parent_ref = PageRef[parent]

    logger.info("Creating database @ %s - %s", parent_ref.page_id, title)

    request = self._build_request(parent_ref, schema, title)

    data = self().create(**request)

    return Database.parse_obj(data)

delete(dbref)

Delete (archive) the specified Database.

dbref may be any suitable DatabaseRef type.

Source code in src/notional/session.py 
327
328
329
330
331
332
333
334
335
336
337
def delete(self, dbref):
    """Delete (archive) the specified Database.

    `dbref` may be any suitable `DatabaseRef` type.
    """

    dbid = DatabaseRef[dbref].database_id

    logger.info("Deleting database :: %s", dbid)

    return self.session.blocks.delete(dbid)

query(target)

Initialize a new Query object with the target data class.

:param target: either a DatabaseRef type or an ORM class

Source code in src/notional/session.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def query(self, target):
    """Initialize a new Query object with the target data class.

    :param target: either a `DatabaseRef` type or an ORM class
    """

    if isclass(target) and issubclass(target, ConnectedPage):
        cls = target
        dbid = target._notional__database

        if cls._notional__session != self.session:
            raise ValueError("ConnectedPage belongs to a different session")

    else:
        cls = None
        dbid = DatabaseRef[target].database_id

    logger.info("Initializing database query :: {%s} [%s]", dbid, cls)

    return QueryBuilder(endpoint=self().query, datatype=cls, database_id=dbid)

restore(dbref)

Restore (unarchive) the specified Database.

dbref may be any suitable DatabaseRef type.

Source code in src/notional/session.py 
339
340
341
342
343
344
345
346
347
348
349
def restore(self, dbref):
    """Restore (unarchive) the specified Database.

    `dbref` may be any suitable `DatabaseRef` type.
    """

    dbid = DatabaseRef[dbref].database_id

    logger.info("Restoring database :: %s", dbid)

    return self.session.blocks.restore(dbid)

retrieve(dbref)

Return the Database with the given ID.

dbref may be any suitable DatabaseRef type.

Source code in src/notional/session.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
def retrieve(self, dbref):
    """Return the Database with the given ID.

    `dbref` may be any suitable `DatabaseRef` type.
    """

    dbid = DatabaseRef[dbref].database_id

    logger.info("Retrieving database :: %s", dbid)

    data = self().retrieve(dbid)

    return Database.parse_obj(data)

update(dbref, title=None, schema=None)

Update the Database object on the server.

The database info will be refreshed to the latest version from the server.

dbref may be any suitable DatabaseRef type.

Source code in src/notional/session.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def update(self, dbref, title=None, schema: Dict[str, PropertyObject] = None):
    """Update the Database object on the server.

    The database info will be refreshed to the latest version from the server.

    `dbref` may be any suitable `DatabaseRef` type.
    """

    dbid = DatabaseRef[dbref].database_id

    logger.info("Updating database info :: %s", dbid)

    request = self._build_request(schema=schema, title=title)

    if request:
        data = self().update(dbid, **request)
        dbref = dbref.refresh(**data)

    return dbref

Endpoint

Notional wrapper for the API endpoints.

Source code in src/notional/session.py 
100
101
102
103
104
105
class Endpoint:
    """Notional wrapper for the API endpoints."""

    def __init__(self, session: Session):
        """Initialize the `Endpoint` for the supplied session."""
        self.session = session

__init__(session)

Initialize the Endpoint for the supplied session.

Source code in src/notional/session.py 
103
104
105
def __init__(self, session: Session):
    """Initialize the `Endpoint` for the supplied session."""
    self.session = session

PagesEndpoint

Bases: Endpoint

Notional interface to the API 'pages' endpoint.

Source code in src/notional/session.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
class PagesEndpoint(Endpoint):
    """Notional interface to the API 'pages' endpoint."""

    class PropertiesEndpoint(Endpoint):
        """Notional interface to the API 'pages/properties' endpoint."""

        def __call__(self):
            """Return the underlying endpoint in the Notion SDK."""
            return self.session.client.pages.properties

        # https://developers.notion.com/reference/retrieve-a-page-property
        def retrieve(self, page_id, property_id):
            """Return the Property on a specific Page with the given ID."""

            logger.info("Retrieving property :: %s [%s]", property_id, page_id)

            data = self().retrieve(page_id, property_id)

            # TODO should PropertyListItem return an iterator instead?
            return parse_obj_as(Union[PropertyItem, PropertyItemList], obj=data)

    def __init__(self, *args, **kwargs):
        """Initialize the `pages` endpoint for the Notion API."""
        super().__init__(*args, **kwargs)

        self.properties = PagesEndpoint.PropertiesEndpoint(*args, **kwargs)

    def __call__(self):
        """Return the underlying endpoint in the Notion SDK."""
        return self.session.client.pages

    # https://developers.notion.com/reference/post-page
    def create(self, parent, title=None, properties=None, children=None):
        """Add a page to the given parent (Page or Database).

        `parent` may be a `ParentRef`, `Page`, or `Database` object.
        """

        if parent is None:
            raise ValueError("'parent' must be provided")

        if isinstance(parent, Page):
            parent = PageRef[parent]
        elif isinstance(parent, Database):
            parent = DatabaseRef[parent]
        elif not isinstance(parent, ParentRef):
            raise ValueError("Unsupported 'parent'")

        request = {"parent": parent.dict()}

        # the API requires a properties object, even if empty
        if properties is None:
            properties = {}

        if title is not None:
            properties["title"] = Title[title]

        request["properties"] = {
            name: prop.dict() if prop is not None else None
            for name, prop in properties.items()
        }

        if children is not None:
            request["children"] = [
                child.dict() for child in children if child is not None
            ]

        logger.info("Creating page :: %s => %s", parent, title)

        data = self().create(**request)

        return Page.parse_obj(data)

    def delete(self, page):
        """Delete (archive) the specified Page.

        `page` may be any suitable `PageRef` type.
        """

        return self.set(page, archived=True)

    def restore(self, page):
        """Restore (unarchive) the specified Page.

        `page` may be any suitable `PageRef` type.
        """

        return self.set(page, archived=False)

    # https://developers.notion.com/reference/retrieve-a-page
    def retrieve(self, page):
        """Return the requested Page.

        `page` may be any suitable `PageRef` type.
        """

        page_id = PageRef[page].page_id

        logger.info("Retrieving page :: %s", page_id)

        data = self().retrieve(page_id)

        # XXX would it make sense to (optionally) expand the full properties here?
        # e.g. call the PropertiesEndpoint to make sure all data is retrieved

        return Page.parse_obj(data)

    # https://developers.notion.com/reference/patch-page
    def update(self, page: Page, **properties):
        """Update the Page object properties on the server.

        An optional `properties` may be specified as `"name"`: `PropertyValue` pairs.

        If `properties` are provided, only those values will be updated.
        If `properties` is empty, all page properties will be updated.

        The page info will be refreshed to the latest version from the server.
        """

        logger.info("Updating page info :: %s", page.id)

        if not properties:
            properties = page.properties

        props = {
            name: value.dict() if value is not None else None
            for name, value in properties.items()
        }

        data = self().update(page.id.hex, properties=props)

        return page.refresh(**data)

    def set(self, page, cover=False, icon=False, archived=None):
        """Set specific page attributes (such as cover, icon, etc.) on the server.

        `page` may be any suitable `PageRef` type.

        To remove an attribute, set its value to None.
        """

        page_id = PageRef[page].page_id

        props = {}

        if cover is None:
            logger.info("Removing page cover :: %s", page_id)
            props["cover"] = {}
        elif cover is not False:
            logger.info("Setting page cover :: %s => %s", page_id, cover)
            props["cover"] = cover.dict()

        if icon is None:
            logger.info("Removing page icon :: %s", page_id)
            props["icon"] = {}
        elif icon is not False:
            logger.info("Setting page icon :: %s => %s", page_id, icon)
            props["icon"] = icon.dict()

        if archived is False:
            logger.info("Restoring page :: %s", page_id)
            props["archived"] = False
        elif archived is True:
            logger.info("Archiving page :: %s", page_id)
            props["archived"] = True

        data = self().update(page_id.hex, **props)

        return page.refresh(**data)

PropertiesEndpoint

Bases: Endpoint

Notional interface to the API 'pages/properties' endpoint.

Source code in src/notional/session.py 
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
class PropertiesEndpoint(Endpoint):
    """Notional interface to the API 'pages/properties' endpoint."""

    def __call__(self):
        """Return the underlying endpoint in the Notion SDK."""
        return self.session.client.pages.properties

    # https://developers.notion.com/reference/retrieve-a-page-property
    def retrieve(self, page_id, property_id):
        """Return the Property on a specific Page with the given ID."""

        logger.info("Retrieving property :: %s [%s]", property_id, page_id)

        data = self().retrieve(page_id, property_id)

        # TODO should PropertyListItem return an iterator instead?
        return parse_obj_as(Union[PropertyItem, PropertyItemList], obj=data)

__call__()

Return the underlying endpoint in the Notion SDK.

Source code in src/notional/session.py 
380
381
382
def __call__(self):
    """Return the underlying endpoint in the Notion SDK."""
    return self.session.client.pages.properties

retrieve(page_id, property_id)

Return the Property on a specific Page with the given ID.

Source code in src/notional/session.py 
385
386
387
388
389
390
391
392
393
def retrieve(self, page_id, property_id):
    """Return the Property on a specific Page with the given ID."""

    logger.info("Retrieving property :: %s [%s]", property_id, page_id)

    data = self().retrieve(page_id, property_id)

    # TODO should PropertyListItem return an iterator instead?
    return parse_obj_as(Union[PropertyItem, PropertyItemList], obj=data)

__call__()

Return the underlying endpoint in the Notion SDK.

Source code in src/notional/session.py 
401
402
403
def __call__(self):
    """Return the underlying endpoint in the Notion SDK."""
    return self.session.client.pages

__init__(*args, **kwargs)

Initialize the pages endpoint for the Notion API.

Source code in src/notional/session.py 
395
396
397
398
399
def __init__(self, *args, **kwargs):
    """Initialize the `pages` endpoint for the Notion API."""
    super().__init__(*args, **kwargs)

    self.properties = PagesEndpoint.PropertiesEndpoint(*args, **kwargs)

create(parent, title=None, properties=None, children=None)

Add a page to the given parent (Page or Database).

parent may be a ParentRef, Page, or Database object.

Source code in src/notional/session.py 
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
def create(self, parent, title=None, properties=None, children=None):
    """Add a page to the given parent (Page or Database).

    `parent` may be a `ParentRef`, `Page`, or `Database` object.
    """

    if parent is None:
        raise ValueError("'parent' must be provided")

    if isinstance(parent, Page):
        parent = PageRef[parent]
    elif isinstance(parent, Database):
        parent = DatabaseRef[parent]
    elif not isinstance(parent, ParentRef):
        raise ValueError("Unsupported 'parent'")

    request = {"parent": parent.dict()}

    # the API requires a properties object, even if empty
    if properties is None:
        properties = {}

    if title is not None:
        properties["title"] = Title[title]

    request["properties"] = {
        name: prop.dict() if prop is not None else None
        for name, prop in properties.items()
    }

    if children is not None:
        request["children"] = [
            child.dict() for child in children if child is not None
        ]

    logger.info("Creating page :: %s => %s", parent, title)

    data = self().create(**request)

    return Page.parse_obj(data)

delete(page)

Delete (archive) the specified Page.

page may be any suitable PageRef type.

Source code in src/notional/session.py 
447
448
449
450
451
452
453
def delete(self, page):
    """Delete (archive) the specified Page.

    `page` may be any suitable `PageRef` type.
    """

    return self.set(page, archived=True)

restore(page)

Restore (unarchive) the specified Page.

page may be any suitable PageRef type.

Source code in src/notional/session.py 
455
456
457
458
459
460
461
def restore(self, page):
    """Restore (unarchive) the specified Page.

    `page` may be any suitable `PageRef` type.
    """

    return self.set(page, archived=False)

retrieve(page)

Return the requested Page.

page may be any suitable PageRef type.

Source code in src/notional/session.py 
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
def retrieve(self, page):
    """Return the requested Page.

    `page` may be any suitable `PageRef` type.
    """

    page_id = PageRef[page].page_id

    logger.info("Retrieving page :: %s", page_id)

    data = self().retrieve(page_id)

    # XXX would it make sense to (optionally) expand the full properties here?
    # e.g. call the PropertiesEndpoint to make sure all data is retrieved

    return Page.parse_obj(data)

set(page, cover=False, icon=False, archived=None)

Set specific page attributes (such as cover, icon, etc.) on the server.

page may be any suitable PageRef type.

To remove an attribute, set its value to None.

Source code in src/notional/session.py 
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
def set(self, page, cover=False, icon=False, archived=None):
    """Set specific page attributes (such as cover, icon, etc.) on the server.

    `page` may be any suitable `PageRef` type.

    To remove an attribute, set its value to None.
    """

    page_id = PageRef[page].page_id

    props = {}

    if cover is None:
        logger.info("Removing page cover :: %s", page_id)
        props["cover"] = {}
    elif cover is not False:
        logger.info("Setting page cover :: %s => %s", page_id, cover)
        props["cover"] = cover.dict()

    if icon is None:
        logger.info("Removing page icon :: %s", page_id)
        props["icon"] = {}
    elif icon is not False:
        logger.info("Setting page icon :: %s => %s", page_id, icon)
        props["icon"] = icon.dict()

    if archived is False:
        logger.info("Restoring page :: %s", page_id)
        props["archived"] = False
    elif archived is True:
        logger.info("Archiving page :: %s", page_id)
        props["archived"] = True

    data = self().update(page_id.hex, **props)

    return page.refresh(**data)

update(page, **properties)

Update the Page object properties on the server.

An optional properties may be specified as "name": PropertyValue pairs.

If properties are provided, only those values will be updated. If properties is empty, all page properties will be updated.

The page info will be refreshed to the latest version from the server.

Source code in src/notional/session.py 
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
def update(self, page: Page, **properties):
    """Update the Page object properties on the server.

    An optional `properties` may be specified as `"name"`: `PropertyValue` pairs.

    If `properties` are provided, only those values will be updated.
    If `properties` is empty, all page properties will be updated.

    The page info will be refreshed to the latest version from the server.
    """

    logger.info("Updating page info :: %s", page.id)

    if not properties:
        properties = page.properties

    props = {
        name: value.dict() if value is not None else None
        for name, value in properties.items()
    }

    data = self().update(page.id.hex, properties=props)

    return page.refresh(**data)

SearchEndpoint

Bases: Endpoint

Notional interface to the API 'search' endpoint.

Source code in src/notional/session.py 
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
class SearchEndpoint(Endpoint):
    """Notional interface to the API 'search' endpoint."""

    # https://developers.notion.com/reference/post-search
    def __call__(self, text=None):
        """Perform a search with the optional text.

        If specified, the call will perform a search with the given text.

        :return: a `QueryBuilder` with the requested search
        :rtype: query.QueryBuilder
        """

        params = {}

        if text is not None:
            params["query"] = text

        return QueryBuilder(endpoint=self.session.client.search, **params)

__call__(text=None)

Perform a search with the optional text.

If specified, the call will perform a search with the given text.

:return: a QueryBuilder with the requested search :rtype: query.QueryBuilder

Source code in src/notional/session.py 
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
def __call__(self, text=None):
    """Perform a search with the optional text.

    If specified, the call will perform a search with the given text.

    :return: a `QueryBuilder` with the requested search
    :rtype: query.QueryBuilder
    """

    params = {}

    if text is not None:
        params["query"] = text

    return QueryBuilder(endpoint=self.session.client.search, **params)

Session

An active session with the Notion SDK.

Source code in src/notional/session.py 
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
class Session:
    """An active session with the Notion SDK."""

    def __init__(self, **kwargs):
        """Initialize the `Session` object and the endpoints.

        `kwargs` will be passed direction to the Notion SDK Client.  For more details,
        see the (full docs)[https://ramnes.github.io/notion-sdk-py/reference/client/].

        :param auth: bearer token for authentication
        """
        self.client = notion_client.Client(**kwargs)

        self.blocks = BlocksEndpoint(self)
        self.databases = DatabasesEndpoint(self)
        self.pages = PagesEndpoint(self)
        self.search = SearchEndpoint(self)
        self.users = UsersEndpoint(self)

        logger.info("Initialized Notion SDK client")

    @property
    def IsActive(self):
        """Determine if the current session is active.

        The session is considered "active" if it has not been closed.  This does not
        determine if the session can connect to the Notion API.
        """
        return self.client is not None

    def close(self):
        """Close the session and release resources."""

        if self.client is None:
            raise SessionError("Session is not active.")

        self.client.close()
        self.client = None

    def ping(self):
        """Confirm that the session is active and able to connect to Notion.

        Raises SessionError if there is a problem, otherwise returns True.
        """

        if self.IsActive is False:
            return False

        error = None

        try:
            me = self.users.me()

            if me is None:
                raise SessionError("Unable to get current user")

        except ConnectError:
            error = "Unable to connect to Notion"

        except APIResponseError as err:
            error = str(err)

        if error is not None:
            raise SessionError(error)

        return True

IsActive property

Determine if the current session is active.

The session is considered "active" if it has not been closed. This does not determine if the session can connect to the Notion API.

__init__(**kwargs)

Initialize the Session object and the endpoints.

kwargs will be passed direction to the Notion SDK Client. For more details, see the (full docs)[https://ramnes.github.io/notion-sdk-py/reference/client/].

:param auth: bearer token for authentication

Source code in src/notional/session.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(self, **kwargs):
    """Initialize the `Session` object and the endpoints.

    `kwargs` will be passed direction to the Notion SDK Client.  For more details,
    see the (full docs)[https://ramnes.github.io/notion-sdk-py/reference/client/].

    :param auth: bearer token for authentication
    """
    self.client = notion_client.Client(**kwargs)

    self.blocks = BlocksEndpoint(self)
    self.databases = DatabasesEndpoint(self)
    self.pages = PagesEndpoint(self)
    self.search = SearchEndpoint(self)
    self.users = UsersEndpoint(self)

    logger.info("Initialized Notion SDK client")

close()

Close the session and release resources.

Source code in src/notional/session.py 
62
63
64
65
66
67
68
69
def close(self):
    """Close the session and release resources."""

    if self.client is None:
        raise SessionError("Session is not active.")

    self.client.close()
    self.client = None

ping()

Confirm that the session is active and able to connect to Notion.

Raises SessionError if there is a problem, otherwise returns True.

Source code in src/notional/session.py 
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
def ping(self):
    """Confirm that the session is active and able to connect to Notion.

    Raises SessionError if there is a problem, otherwise returns True.
    """

    if self.IsActive is False:
        return False

    error = None

    try:
        me = self.users.me()

        if me is None:
            raise SessionError("Unable to get current user")

    except ConnectError:
        error = "Unable to connect to Notion"

    except APIResponseError as err:
        error = str(err)

    if error is not None:
        raise SessionError(error)

    return True

SessionError

Bases: Exception

Raised when there are issues with the Notion session.

Source code in src/notional/session.py 
24
25
26
27
28
29
class SessionError(Exception):
    """Raised when there are issues with the Notion session."""

    def __init__(self, message):
        """Initialize the `SessionError` with a supplied message.."""
        super().__init__(message)

__init__(message)

Initialize the SessionError with a supplied message..

Source code in src/notional/session.py 
27
28
29
def __init__(self, message):
    """Initialize the `SessionError` with a supplied message.."""
    super().__init__(message)

UsersEndpoint

Bases: Endpoint

Notional interface to the API 'users' endpoint.

Source code in src/notional/session.py 
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
class UsersEndpoint(Endpoint):
    """Notional interface to the API 'users' endpoint."""

    def __call__(self):
        """Return the underlying endpoint in the Notion SDK."""
        return self.session.client.users

    # https://developers.notion.com/reference/get-users
    def list(self):
        """Return an iterator for all users in the workspace."""

        logger.info("Listing known users...")

        users = EndpointIterator(endpoint=self().list)

        return users()

    # https://developers.notion.com/reference/get-user
    def retrieve(self, user_id):
        """Return the User with the given ID."""

        logger.info("Retrieving user :: %s", user_id)

        data = self().retrieve(user_id)

        return User.parse_obj(data)

    # https://developers.notion.com/reference/get-self
    def me(self):
        """Return the current bot User."""

        logger.info("Retrieving current integration bot")

        data = self().me()

        return User.parse_obj(data)

__call__()

Return the underlying endpoint in the Notion SDK.

Source code in src/notional/session.py 
569
570
571
def __call__(self):
    """Return the underlying endpoint in the Notion SDK."""
    return self.session.client.users

list()

Return an iterator for all users in the workspace.

Source code in src/notional/session.py 
574
575
576
577
578
579
580
581
def list(self):
    """Return an iterator for all users in the workspace."""

    logger.info("Listing known users...")

    users = EndpointIterator(endpoint=self().list)

    return users()

me()

Return the current bot User.

Source code in src/notional/session.py 
594
595
596
597
598
599
600
601
def me(self):
    """Return the current bot User."""

    logger.info("Retrieving current integration bot")

    data = self().me()

    return User.parse_obj(data)

retrieve(user_id)

Return the User with the given ID.

Source code in src/notional/session.py 
584
585
586
587
588
589
590
591
def retrieve(self, user_id):
    """Return the User with the given ID."""

    logger.info("Retrieving user :: %s", user_id)

    data = self().retrieve(user_id)

    return User.parse_obj(data)