tessl/pypi-notion-client

Python client for the official Notion API

—

Pending

Overview

Eval results

Files

Helper Utilities

Name: tessl/pypi-notion-client
Author: tessl

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.

Capabilities

URL Utilities

Functions for converting between Notion object IDs and URLs.

def get_url(object_id):
    """
    Return the URL for the object with the given id.
    
    Parameters:
    - object_id: str, Notion object ID (page, database, or block)
    
    Returns:
    str, Notion URL for the object
    
    Example:
    get_url("d40e767c-d7af-4b18-a86d-55c61f1e39a4")
    # Returns: "https://notion.so/d40e767cd7af4b18a86d55c61f1e39a4"
    """

def get_id(url):
    """
    Return the id of the object behind the given URL.
    
    Parameters:
    - url: str, Notion URL
    
    Returns:
    str, Notion object ID extracted from URL
    
    Raises:
    - ValueError: If URL is not a valid Notion URL or path is incorrect
    
    Example:
    get_id("https://notion.so/My-Page-d40e767cd7af4b18a86d55c61f1e39a4")
    # Returns: "d40e767c-d7af-4b18-a86d-55c61f1e39a4"
    """

Data Utilities

Helper functions for data manipulation and filtering.

def pick(base, *keys):
    """
    Return a dict composed of key value pairs for keys passed as args.
    
    Parameters:
    - base: dict, source dictionary
    - *keys: str, keys to extract from base dictionary
    
    Returns:
    dict, new dictionary with only the specified keys
    
    Note: 
    - Skips keys that don't exist in base
    - Skips start_cursor if value is None
    
    Example:
    pick({"a": 1, "b": 2, "c": 3}, "a", "c")
    # Returns: {"a": 1, "c": 3}
    """

Pagination Utilities

Functions for handling paginated API responses.

def iterate_paginated_api(function, **kwargs):
    """
    Return an iterator over the results of any paginated Notion API.
    
    Parameters:
    - function: callable, API function that returns paginated results
    - **kwargs: keyword arguments to pass to the function
    
    Yields:
    Individual result objects from paginated responses
    
    Example:
    for page in iterate_paginated_api(notion.databases.query, database_id="xxx"):
        print(page["id"])
    """

def collect_paginated_api(function, **kwargs):
    """
    Collect all the results of paginating an API into a list.
    
    Parameters:
    - function: callable, API function that returns paginated results
    - **kwargs: keyword arguments to pass to the function
    
    Returns:
    list, all results from all pages collected into a single list
    
    Warning: This loads all results into memory. Use iterate_paginated_api 
    for large datasets to avoid memory issues.
    
    Example:
    all_pages = collect_paginated_api(notion.databases.query, database_id="xxx")
    """

def async_iterate_paginated_api(function, **kwargs):
    """
    Return an async iterator over the results of any paginated Notion API.
    
    Parameters:
    - function: callable, async API function that returns paginated results
    - **kwargs: keyword arguments to pass to the function
    
    Yields:
    Individual result objects from paginated responses
    
    Example:
    async for page in async_iterate_paginated_api(notion.databases.query, database_id="xxx"):
        print(page["id"])
    """

def async_collect_paginated_api(function, **kwargs):
    """
    Collect asynchronously all the results of paginating an API into a list.
    
    Parameters:
    - function: callable, async API function that returns paginated results
    - **kwargs: keyword arguments to pass to the function
    
    Returns:
    list, all results from all pages collected into a single list
    
    Example:
    all_pages = await async_collect_paginated_api(notion.databases.query, database_id="xxx")
    """

Response Validation Utilities

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

def is_full_block(response):
    """
    Return True if response is a full block.
    
    Parameters:
    - response: dict, API response object
    
    Returns:
    bool, True if response is a complete block object
    """

def is_full_page(response):
    """
    Return True if response is a full page.
    
    Parameters:
    - response: dict, API response object
    
    Returns:
    bool, True if response is a complete page object
    """

def is_full_database(response):
    """
    Return True if response is a full database.
    
    Parameters:
    - response: dict, API response object
    
    Returns:
    bool, True if response is a complete database object
    """

def is_full_page_or_database(response):
    """
    Return True if response is a full database or a full page.
    
    Parameters:
    - response: dict, API response object
    
    Returns:
    bool, True if response is a complete page or database object
    """

def is_full_user(response):
    """
    Return True if response is a full user.
    
    Parameters:
    - response: dict, API response object
    
    Returns:
    bool, True if response is a complete user object
    """

def is_full_comment(response):
    """
    Return True if response is a full comment.
    
    Parameters:
    - response: dict, API response object
    
    Returns:
    bool, True if response is a complete comment object
    """

Rich Text Utilities

Functions for working with rich text objects in Notion content.

def is_text_rich_text_item_response(rich_text):
    """
    Return True if rich_text is a text.
    
    Parameters:
    - rich_text: dict, rich text item object
    
    Returns:
    bool, True if rich text item is of type "text"
    """

def is_equation_rich_text_item_response(rich_text):
    """
    Return True if rich_text is an equation.
    
    Parameters:
    - rich_text: dict, rich text item object
    
    Returns:
    bool, True if rich text item is of type "equation"
    """

def is_mention_rich_text_item_response(rich_text):
    """
    Return True if rich_text is a mention.
    
    Parameters:
    - rich_text: dict, rich text item object
    
    Returns:
    bool, True if rich text item is of type "mention"
    """

Usage Examples

Working with URLs

from notion_client.helpers import get_url, get_id

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

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

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

Handling Pagination

from notion_client import Client
from notion_client.helpers import (
    iterate_paginated_api, 
    collect_paginated_api,
    async_iterate_paginated_api,
    async_collect_paginated_api
)

notion = Client(auth="your_token")

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

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

# Async pagination
async def process_async_pages():
    async_notion = AsyncClient(auth="your_token")
    
    # Async iteration
    async for page in async_iterate_paginated_api(async_notion.databases.query, database_id="db_id"):
        print(f"Async page: {page['id']}")
    
    # Async collection
    all_async_pages = await async_collect_paginated_api(async_notion.databases.query, database_id="db_id")
    print(f"Total async pages: {len(all_async_pages)}")
    
    await async_notion.aclose()

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

# Custom pagination parameters
recent_pages = collect_paginated_api(
    notion.databases.query,
    database_id="db_id",
    sorts=[{"property": "Created", "direction": "descending"}],
    page_size=50  # Custom page size
)

Response Validation

from notion_client.helpers import (
    is_full_page, is_full_database, is_full_block,
    is_full_page_or_database, is_full_user, is_full_comment
)

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

for result in search_results["results"]:
    if is_full_page(result):
        print(f"Found page: {result['properties']['title']['title'][0]['text']['content']}")
    elif is_full_database(result):
        print(f"Found database: {result['title'][0]['text']['content']}")
    
    # Alternative: check for either
    if is_full_page_or_database(result):
        print(f"Found page or database: {result['id']}")

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

# Check blocks
blocks = notion.blocks.children.list(block_id="page_id")
for block in blocks["results"]:
    if is_full_block(block):
        print(f"Valid block: {block['type']}")

Rich Text Processing

from notion_client.helpers import (
    is_text_rich_text_item_response,
    is_equation_rich_text_item_response, 
    is_mention_rich_text_item_response
)

def extract_text_content(rich_text_array):
    """Extract plain text from rich text array."""
    text_content = []
    
    for item in rich_text_array:
        if is_text_rich_text_item_response(item):
            text_content.append(item["text"]["content"])
        elif is_equation_rich_text_item_response(item):
            text_content.append(f"[Equation: {item['equation']['expression']}]")
        elif is_mention_rich_text_item_response(item):
            mention_type = item["mention"]["type"]
            if mention_type == "page":
                text_content.append(f"[Page: {item['mention']['page']['id']}]")
            elif mention_type == "user":
                text_content.append(f"[User: {item['mention']['user']['id']}]")
            else:
                text_content.append(f"[{mention_type.title()} mention]")
        else:
            # Handle other rich text types
            text_content.append(f"[{item.get('type', 'unknown')} content]")
    
    return "".join(text_content)

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

# Usage with block content
blocks = notion.blocks.children.list(block_id="page_id")
for block in blocks["results"]:
    if block["type"] == "paragraph":
        content = extract_text_content(block["paragraph"]["rich_text"])
        print(f"Paragraph: {content}")

Data Filtering and Manipulation

from notion_client.helpers import pick

# Filter request parameters
user_input = {
    "database_id": "abc123",
    "filter": {"property": "Status", "select": {"equals": "Active"}},
    "sorts": [{"property": "Created", "direction": "desc"}],
    "extra_param": "should_be_ignored",
    "start_cursor": None,  # Will be filtered out
    "page_size": 25
}

# Extract only valid query parameters
query_params = pick(
    user_input, 
    "filter", "sorts", "start_cursor", "page_size"
)
print(query_params)  # {"filter": {...}, "sorts": [...], "page_size": 25}

# Use with API call
response = notion.databases.query(
    database_id=user_input["database_id"],
    **query_params
)

# Helpful for building reusable functions
def safe_database_query(notion, database_id, **kwargs):
    """Query database with only valid parameters."""
    valid_params = pick(
        kwargs,
        "filter", "sorts", "start_cursor", "page_size", 
        "archived", "in_trash", "filter_properties"
    )
    return notion.databases.query(database_id=database_id, **valid_params)

# Usage
results = safe_database_query(
    notion,
    "db_id",
    filter={"property": "Status", "select": {"equals": "Done"}},
    page_size=50,
    invalid_param="ignored"  # This will be filtered out
)

Advanced Pagination Patterns

from notion_client.helpers import iterate_paginated_api
import time

def process_with_progress(notion, database_id):
    """Process paginated results with progress tracking."""
    total_processed = 0
    start_time = time.time()
    
    for page in iterate_paginated_api(notion.databases.query, database_id=database_id):
        # Process each page
        process_page(page)
        total_processed += 1
        
        # Progress update every 100 pages
        if total_processed % 100 == 0:
            elapsed = time.time() - start_time
            rate = total_processed / elapsed
            print(f"Processed {total_processed} pages ({rate:.1f} pages/sec)")
    
    total_time = time.time() - start_time
    print(f"Finished processing {total_processed} pages in {total_time:.1f} seconds")

def process_page(page):
    """Process individual page."""
    # Your processing logic here
    pass

# Batch processing with pagination
def batch_process_pages(notion, database_id, batch_size=50):
    """Process pages in batches."""
    batch = []
    
    for page in iterate_paginated_api(notion.databases.query, database_id=database_id):
        batch.append(page)
        
        if len(batch) >= batch_size:
            process_batch(batch)
            batch = []
    
    # Process remaining pages
    if batch:
        process_batch(batch)

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

# Filtered pagination
def get_pages_by_status(notion, database_id, status):
    """Get all pages with specific status using pagination."""
    pages_with_status = []
    
    for page in iterate_paginated_api(
        notion.databases.query,
        database_id=database_id,
        filter={
            "property": "Status",
            "select": {"equals": status}
        }
    ):
        pages_with_status.append(page)
    
    return pages_with_status

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

Install with Tessl CLI