API is very slow for returning full page content

[nick-youngblut]

My code:

import os
import asyncio
from notion_client import AsyncClient
from notion_client.errors import APIResponseError
from datetime import datetime

def get_plain_text(rich_text):
    if not rich_text:
        return ""
    return "".join(text.get("plain_text", "") for text in rich_text)

async def get_all_blocks(block_id, notion):
    """Get all blocks including nested ones concurrently"""
    blocks = []
    
    async def fetch_blocks_recursive(current_block_id):
        """Recursively fetch all blocks and their children"""
        cursor = None
        
        # If this is the initial block_id and not a page, get that block first
        if len(blocks) == 0 and not current_block_id.endswith('-'):
            try:
                initial_block = await notion.blocks.retrieve(block_id=current_block_id)
                blocks.append(initial_block)
            except Exception as e:
                print(f"Error fetching initial block: {e}")
                return
        
        # Get all children
        while True:
            try:
                response = await notion.blocks.children.list(
                    block_id=current_block_id,
                    start_cursor=cursor,
                    page_size=100
                )
                current_blocks = response["results"]
                blocks.extend(current_blocks)
                
                # Recursively fetch children
                tasks = []
                for block in current_blocks:
                    if block.get("has_children"):
                        tasks.append(fetch_blocks_recursive(block["id"]))
                if tasks:
                    await asyncio.gather(*tasks)
                
                if not response.get("has_more"):
                    break
                cursor = response["next_cursor"]
                
            except Exception as e:
                print(f"Error fetching blocks: {e}")
                break
    
    # Start the recursive fetch
    await fetch_blocks_recursive(block_id)
    return blocks

def process_block(block, indent_level=0):
    """Process a single block without async operations"""
    block_type = block["type"]
    text = ""
    indent = "    " * indent_level
    
    if block_type == "paragraph":
        content = get_plain_text(block["paragraph"]["rich_text"])
        text = f"{indent}{content}\n\n" if content else "\n"
        
    elif block_type == "heading_1":
        text = f"{indent}# {get_plain_text(block['heading_1']['rich_text'])}\n\n"
        
    elif block_type == "heading_2":
        text = f"{indent}## {get_plain_text(block['heading_2']['rich_text'])}\n\n"
        
    elif block_type == "heading_3":
        text = f"{indent}### {get_plain_text(block['heading_3']['rich_text'])}\n\n"
        
    elif block_type == "bulleted_list_item":
        text = f"{indent}* {get_plain_text(block['bulleted_list_item']['rich_text'])}\n"
        
    elif block_type == "numbered_list_item":
        text = f"{indent}1. {get_plain_text(block['numbered_list_item']['rich_text'])}\n"
        
    elif block_type == "to_do":
        checked = "x" if block["to_do"]["checked"] else " "
        text = f"{indent}- [{checked}] {get_plain_text(block['to_do']['rich_text'])}\n"
        
    elif block_type == "code":
        code = get_plain_text(block["code"]["rich_text"])
        language = block["code"]["language"]
        text = f"{indent}```{language}\n{indent}{code}\n{indent}```\n\n"
        
    elif block_type == "quote":
        text = f"{indent}> {get_plain_text(block['quote']['rich_text'])}\n\n"
        
    elif block_type == "divider":
        text = f"{indent}---\n\n"
        
    elif block_type == "callout":
        emoji = block["callout"].get("icon", {}).get("emoji", "")
        text = f"{indent}> {emoji} {get_plain_text(block['callout']['rich_text'])}\n\n"
        
    elif block_type == "toggle":
        # Handle toggle blocks with a ▶ symbol
        content = get_plain_text(block["toggle"]["rich_text"])
        text = f"{indent}▶ {content}\n"
    
    return text, block.get("has_children", False)


def build_markdown(blocks, all_blocks, level=0):
    result = ""
    for block in blocks:
        text, has_children = process_block(block, level)
        result += text
        if has_children:
            child_blocks = [b for b in all_blocks if b.get("parent", {}).get("block_id") == block["id"]]
            result += build_markdown(child_blocks, all_blocks, level + 1)
    return result

async def convert_notion_to_markdown(notion_id, input_type="page"):
    """
    Convert Notion content to markdown
    
    Args:
        notion_id (str): ID of the Notion page or block
        input_type (str): Either "page" or "block"
    """
    notion = AsyncClient(auth=os.getenv("NOTION_API_KEY"))
    markdown_content = ""
    
    # Get all blocks concurrently
    all_blocks = await get_all_blocks(notion_id, notion)
    
    # For pages, add the title
    try:
        page = await notion.pages.retrieve(page_id=notion_id)
        title = page.get("properties", {}).get("title", {}).get("title", [{}])[0].get("plain_text", "Untitled")
        markdown_content = f"# {title}\n\n"
        
        # Build page hierarchy
        root_blocks = []
        for block in all_blocks:
            parent_id = block.get("parent", {}).get("block_id")
            if parent_id == notion_id or not parent_id:
                root_blocks.append(block)
    except APIResponseError:
        # For blocks, start with the first block and process its children
        root_blocks = [all_blocks[0]]  # The first block will be our target block
        all_blocks = all_blocks[1:]    # The rest are its children
    
    # Process blocks maintaining hierarchy
    markdown_content += build_markdown(root_blocks, all_blocks)
    
    await notion.aclose()
    return markdown_content

# main
if __name__ == '__main__': 
    from dotenv import load_dotenv
    load_dotenv()

    id = "######"    
    page_content = asyncio.run(convert_notion_to_markdown(id))
    print(page_content)

For a moderately sized page, but with many nested blocks (nested toggle lists), fetching the entire page contents takes >10 seconds!

Is there a faster method of obtaining the entire contents of a page? >10 seconds per page is way too slow for many applications.

[ramnes]

Yes, from my experience as well, Notion API is quite slow, especially for reading large pages. From their docs:

Reading large pages may take some time. We recommend using asynchronous operations in your architecture, such as a job queue. You will also need to be mindful of rate limits to appropriately slow down making new requests after the limit is met.

On a side note, consider using async_iterate_paginated_api(function, **kwargs) or async_collect_paginated_api(function, **kwargs) to make your code simpler, as seen in the README. :)

[nick-youngblut]

Thanks @ramnes !

I was hoping to get around creating another database on top of Notion for fast queries of our docs in Notion.

I'm sure that Notion uses a vector database for their AI tooling. Why can't they expose it via the API, and instead, each developer must make their own? Better yet, why can't they exposure their AI tooling via their API? I'd be happy to pay per API call of their AI tools.

On a side note, consider using async_iterate_paginated_api

That's always good to point out. I wrote my own code since I was going to customize the functionality.

[ramnes]

notion-py showed that it's possible to use their non-official public APIs. If you have enough time and resources, that could be an option to explore here.

[nick-youngblut]

On a side note, consider using async_iterate_paginated_api(function, **kwargs) or async_collect_paginated_api(function, **kwargs) to make your code simpler, as seen in the README. :)

It would be great to have examples of using these helpful functions for iterating through pages and blocks in pages. I've tried using the helpful functions but just ran into TypeError: 'NoneType' object is not iterable errors.

The implementation should be collect_paginated_api(notion.pages.retrieve, page_id=page_id), correct? How does the helper then iterate through all blocks on the page?