0
# Helper Utilities
1

2
Utility functions for pagination, URL handling, response validation, and rich text processing. These helpers simplify common operations when working with the Notion API and provide convenient abstractions for complex tasks.
3

4
## Capabilities
5

6
### URL Utilities
7

8
Functions for converting between Notion object IDs and URLs.
9

10
```python { .api }
11
def get_url(object_id):
12
    """
13
    Return the URL for the object with the given id.
14
    
15
    Parameters:
16
    - object_id: str, Notion object ID (page, database, or block)
17
    
18
    Returns:
19
    str, Notion URL for the object
20
    
21
    Example:
22
    get_url("d40e767c-d7af-4b18-a86d-55c61f1e39a4")
23
    # Returns: "https://notion.so/d40e767cd7af4b18a86d55c61f1e39a4"
24
    """
25

26
def get_id(url):
27
    """
28
    Return the id of the object behind the given URL.
29
    
30
    Parameters:
31
    - url: str, Notion URL
32
    
33
    Returns:
34
    str, Notion object ID extracted from URL
35
    
36
    Raises:
37
    - ValueError: If URL is not a valid Notion URL or path is incorrect
38
    
39
    Example:
40
    get_id("https://notion.so/My-Page-d40e767cd7af4b18a86d55c61f1e39a4")
41
    # Returns: "d40e767c-d7af-4b18-a86d-55c61f1e39a4"
42
    """
43
```
44

45
### Data Utilities
46

47
Helper functions for data manipulation and filtering.
48

49
```python { .api }
50
def pick(base, *keys):
51
    """
52
    Return a dict composed of key value pairs for keys passed as args.
53
    
54
    Parameters:
55
    - base: dict, source dictionary
56
    - *keys: str, keys to extract from base dictionary
57
    
58
    Returns:
59
    dict, new dictionary with only the specified keys
60
    
61
    Note: 
62
    - Skips keys that don't exist in base
63
    - Skips start_cursor if value is None
64
    
65
    Example:
66
    pick({"a": 1, "b": 2, "c": 3}, "a", "c")
67
    # Returns: {"a": 1, "c": 3}
68
    """
69
```
70

71
### Pagination Utilities
72

73
Functions for handling paginated API responses.
74

75
```python { .api }
76
def iterate_paginated_api(function, **kwargs):
77
    """
78
    Return an iterator over the results of any paginated Notion API.
79
    
80
    Parameters:
81
    - function: callable, API function that returns paginated results
82
    - **kwargs: keyword arguments to pass to the function
83
    
84
    Yields:
85
    Individual result objects from paginated responses
86
    
87
    Example:
88
    for page in iterate_paginated_api(notion.databases.query, database_id="xxx"):
89
        print(page["id"])
90
    """
91

92
def collect_paginated_api(function, **kwargs):
93
    """
94
    Collect all the results of paginating an API into a list.
95
    
96
    Parameters:
97
    - function: callable, API function that returns paginated results
98
    - **kwargs: keyword arguments to pass to the function
99
    
100
    Returns:
101
    list, all results from all pages collected into a single list
102
    
103
    Warning: This loads all results into memory. Use iterate_paginated_api 
104
    for large datasets to avoid memory issues.
105
    
106
    Example:
107
    all_pages = collect_paginated_api(notion.databases.query, database_id="xxx")
108
    """
109

110
def async_iterate_paginated_api(function, **kwargs):
111
    """
112
    Return an async iterator over the results of any paginated Notion API.
113
    
114
    Parameters:
115
    - function: callable, async API function that returns paginated results
116
    - **kwargs: keyword arguments to pass to the function
117
    
118
    Yields:
119
    Individual result objects from paginated responses
120
    
121
    Example:
122
    async for page in async_iterate_paginated_api(notion.databases.query, database_id="xxx"):
123
        print(page["id"])
124
    """
125

126
def async_collect_paginated_api(function, **kwargs):
127
    """
128
    Collect asynchronously all the results of paginating an API into a list.
129
    
130
    Parameters:
131
    - function: callable, async API function that returns paginated results
132
    - **kwargs: keyword arguments to pass to the function
133
    
134
    Returns:
135
    list, all results from all pages collected into a single list
136
    
137
    Example:
138
    all_pages = await async_collect_paginated_api(notion.databases.query, database_id="xxx")
139
    """
140
```
141

142
### Response Validation Utilities
143

144
Functions for determining the type and completeness of API response objects.
145

146
```python { .api }
147
def is_full_block(response):
148
    """
149
    Return True if response is a full block.
150
    
151
    Parameters:
152
    - response: dict, API response object
153
    
154
    Returns:
155
    bool, True if response is a complete block object
156
    """
157

158
def is_full_page(response):
159
    """
160
    Return True if response is a full page.
161
    
162
    Parameters:
163
    - response: dict, API response object
164
    
165
    Returns:
166
    bool, True if response is a complete page object
167
    """
168

169
def is_full_database(response):
170
    """
171
    Return True if response is a full database.
172
    
173
    Parameters:
174
    - response: dict, API response object
175
    
176
    Returns:
177
    bool, True if response is a complete database object
178
    """
179

180
def is_full_page_or_database(response):
181
    """
182
    Return True if response is a full database or a full page.
183
    
184
    Parameters:
185
    - response: dict, API response object
186
    
187
    Returns:
188
    bool, True if response is a complete page or database object
189
    """
190

191
def is_full_user(response):
192
    """
193
    Return True if response is a full user.
194
    
195
    Parameters:
196
    - response: dict, API response object
197
    
198
    Returns:
199
    bool, True if response is a complete user object
200
    """
201

202
def is_full_comment(response):
203
    """
204
    Return True if response is a full comment.
205
    
206
    Parameters:
207
    - response: dict, API response object
208
    
209
    Returns:
210
    bool, True if response is a complete comment object
211
    """
212
```
213

214
### Rich Text Utilities
215

216
Functions for working with rich text objects in Notion content.
217

218
```python { .api }
219
def is_text_rich_text_item_response(rich_text):
220
    """
221
    Return True if rich_text is a text.
222
    
223
    Parameters:
224
    - rich_text: dict, rich text item object
225
    
226
    Returns:
227
    bool, True if rich text item is of type "text"
228
    """
229

230
def is_equation_rich_text_item_response(rich_text):
231
    """
232
    Return True if rich_text is an equation.
233
    
234
    Parameters:
235
    - rich_text: dict, rich text item object
236
    
237
    Returns:
238
    bool, True if rich text item is of type "equation"
239
    """
240

241
def is_mention_rich_text_item_response(rich_text):
242
    """
243
    Return True if rich_text is a mention.
244
    
245
    Parameters:
246
    - rich_text: dict, rich text item object
247
    
248
    Returns:
249
    bool, True if rich text item is of type "mention"
250
    """
251
```
252

253
## Usage Examples
254

255
### Working with URLs
256

257
```python
258
from notion_client.helpers import get_url, get_id
259

260
# Convert object ID to URL
261
page_id = "d40e767c-d7af-4b18-a86d-55c61f1e39a4"
262
page_url = get_url(page_id)
263
print(page_url)  # https://notion.so/d40e767cd7af4b18a86d55c61f1e39a4
264

265
# Extract ID from URL
266
notion_url = "https://notion.so/My-Page-d40e767cd7af4b18a86d55c61f1e39a4"
267
extracted_id = get_id(notion_url)
268
print(extracted_id)  # d40e767c-d7af-4b18-a86d-55c61f1e39a4
269

270
# Works with different URL formats
271
url_with_params = "https://www.notion.so/My-Database-abc123def456789012345678901234567890?v=xyz"
272
db_id = get_id(url_with_params)
273
print(db_id)  # abc123de-f456-7890-1234-567890123456
274
```
275

276
### Handling Pagination
277

278
```python
279
from notion_client import Client
280
from notion_client.helpers import (
281
    iterate_paginated_api, 
282
    collect_paginated_api,
283
    async_iterate_paginated_api,
284
    async_collect_paginated_api
285
)
286

287
notion = Client(auth="your_token")
288

289
# Memory-efficient iteration over all pages
290
print("Processing pages one by one:")
291
for page in iterate_paginated_api(notion.databases.query, database_id="db_id"):
292
    print(f"Page: {page['id']}")
293
    # Process each page individually
294

295
# Collect all pages into memory at once
296
print("Loading all pages into memory:")
297
all_pages = collect_paginated_api(notion.databases.query, database_id="db_id")
298
print(f"Total pages: {len(all_pages)}")
299

300
# Async pagination
301
async def process_async_pages():
302
    async_notion = AsyncClient(auth="your_token")
303
    
304
    # Async iteration
305
    async for page in async_iterate_paginated_api(async_notion.databases.query, database_id="db_id"):
306
        print(f"Async page: {page['id']}")
307
    
308
    # Async collection
309
    all_async_pages = await async_collect_paginated_api(async_notion.databases.query, database_id="db_id")
310
    print(f"Total async pages: {len(all_async_pages)}")
311
    
312
    await async_notion.aclose()
313

314
# Use with other endpoints
315
all_users = collect_paginated_api(notion.users.list)
316
all_comments = collect_paginated_api(notion.comments.list, block_id="block_id")
317

318
# Custom pagination parameters
319
recent_pages = collect_paginated_api(
320
    notion.databases.query,
321
    database_id="db_id",
322
    sorts=[{"property": "Created", "direction": "descending"}],
323
    page_size=50  # Custom page size
324
)
325
```
326

327
### Response Validation
328

329
```python
330
from notion_client.helpers import (
331
    is_full_page, is_full_database, is_full_block,
332
    is_full_page_or_database, is_full_user, is_full_comment
333
)
334

335
# Get mixed results from search
336
search_results = notion.search(query="project")
337

338
for result in search_results["results"]:
339
    if is_full_page(result):
340
        print(f"Found page: {result['properties']['title']['title'][0]['text']['content']}")
341
    elif is_full_database(result):
342
        print(f"Found database: {result['title'][0]['text']['content']}")
343
    
344
    # Alternative: check for either
345
    if is_full_page_or_database(result):
346
        print(f"Found page or database: {result['id']}")
347

348
# Validate specific object types
349
users = notion.users.list()
350
for user in users["results"]:
351
    if is_full_user(user):
352
        print(f"Valid user object: {user['name']}")
353

354
# Check blocks
355
blocks = notion.blocks.children.list(block_id="page_id")
356
for block in blocks["results"]:
357
    if is_full_block(block):
358
        print(f"Valid block: {block['type']}")
359
```
360

361
### Rich Text Processing
362

363
```python
364
from notion_client.helpers import (
365
    is_text_rich_text_item_response,
366
    is_equation_rich_text_item_response, 
367
    is_mention_rich_text_item_response
368
)
369

370
def extract_text_content(rich_text_array):
371
    """Extract plain text from rich text array."""
372
    text_content = []
373
    
374
    for item in rich_text_array:
375
        if is_text_rich_text_item_response(item):
376
            text_content.append(item["text"]["content"])
377
        elif is_equation_rich_text_item_response(item):
378
            text_content.append(f"[Equation: {item['equation']['expression']}]")
379
        elif is_mention_rich_text_item_response(item):
380
            mention_type = item["mention"]["type"]
381
            if mention_type == "page":
382
                text_content.append(f"[Page: {item['mention']['page']['id']}]")
383
            elif mention_type == "user":
384
                text_content.append(f"[User: {item['mention']['user']['id']}]")
385
            else:
386
                text_content.append(f"[{mention_type.title()} mention]")
387
        else:
388
            # Handle other rich text types
389
            text_content.append(f"[{item.get('type', 'unknown')} content]")
390
    
391
    return "".join(text_content)
392

393
# Usage with page title
394
page = notion.pages.retrieve(page_id="page_id")
395
title_rich_text = page["properties"]["title"]["title"]
396
title = extract_text_content(title_rich_text)
397
print(f"Page title: {title}")
398

399
# Usage with block content
400
blocks = notion.blocks.children.list(block_id="page_id")
401
for block in blocks["results"]:
402
    if block["type"] == "paragraph":
403
        content = extract_text_content(block["paragraph"]["rich_text"])
404
        print(f"Paragraph: {content}")
405
```
406

407
### Data Filtering and Manipulation
408

409
```python
410
from notion_client.helpers import pick
411

412
# Filter request parameters
413
user_input = {
414
    "database_id": "abc123",
415
    "filter": {"property": "Status", "select": {"equals": "Active"}},
416
    "sorts": [{"property": "Created", "direction": "desc"}],
417
    "extra_param": "should_be_ignored",
418
    "start_cursor": None,  # Will be filtered out
419
    "page_size": 25
420
}
421

422
# Extract only valid query parameters
423
query_params = pick(
424
    user_input, 
425
    "filter", "sorts", "start_cursor", "page_size"
426
)
427
print(query_params)  # {"filter": {...}, "sorts": [...], "page_size": 25}
428

429
# Use with API call
430
response = notion.databases.query(
431
    database_id=user_input["database_id"],
432
    **query_params
433
)
434

435
# Helpful for building reusable functions
436
def safe_database_query(notion, database_id, **kwargs):
437
    """Query database with only valid parameters."""
438
    valid_params = pick(
439
        kwargs,
440
        "filter", "sorts", "start_cursor", "page_size", 
441
        "archived", "in_trash", "filter_properties"
442
    )
443
    return notion.databases.query(database_id=database_id, **valid_params)
444

445
# Usage
446
results = safe_database_query(
447
    notion,
448
    "db_id",
449
    filter={"property": "Status", "select": {"equals": "Done"}},
450
    page_size=50,
451
    invalid_param="ignored"  # This will be filtered out
452
)
453
```
454

455
### Advanced Pagination Patterns
456

457
```python
458
from notion_client.helpers import iterate_paginated_api
459
import time
460

461
def process_with_progress(notion, database_id):
462
    """Process paginated results with progress tracking."""
463
    total_processed = 0
464
    start_time = time.time()
465
    
466
    for page in iterate_paginated_api(notion.databases.query, database_id=database_id):
467
        # Process each page
468
        process_page(page)
469
        total_processed += 1
470
        
471
        # Progress update every 100 pages
472
        if total_processed % 100 == 0:
473
            elapsed = time.time() - start_time
474
            rate = total_processed / elapsed
475
            print(f"Processed {total_processed} pages ({rate:.1f} pages/sec)")
476
    
477
    total_time = time.time() - start_time
478
    print(f"Finished processing {total_processed} pages in {total_time:.1f} seconds")
479

480
def process_page(page):
481
    """Process individual page."""
482
    # Your processing logic here
483
    pass
484

485
# Batch processing with pagination
486
def batch_process_pages(notion, database_id, batch_size=50):
487
    """Process pages in batches."""
488
    batch = []
489
    
490
    for page in iterate_paginated_api(notion.databases.query, database_id=database_id):
491
        batch.append(page)
492
        
493
        if len(batch) >= batch_size:
494
            process_batch(batch)
495
            batch = []
496
    
497
    # Process remaining pages
498
    if batch:
499
        process_batch(batch)
500

501
def process_batch(pages):
502
    """Process a batch of pages."""
503
    print(f"Processing batch of {len(pages)} pages")
504
    # Your batch processing logic here
505

506
# Filtered pagination
507
def get_pages_by_status(notion, database_id, status):
508
    """Get all pages with specific status using pagination."""
509
    pages_with_status = []
510
    
511
    for page in iterate_paginated_api(
512
        notion.databases.query,
513
        database_id=database_id,
514
        filter={
515
            "property": "Status",
516
            "select": {"equals": status}
517
        }
518
    ):
519
        pages_with_status.append(page)
520
    
521
    return pages_with_status
522

523
# Usage
524
active_pages = get_pages_by_status(notion, "db_id", "Active")
525
print(f"Found {len(active_pages)} active pages")
526
```