Load Notion pages into LlamaIndex with NotionPageReader

This llamaindex notionpagereader tutorial walks through the complete ingestion pipeline: authenticating with Notion, configuring the reader, handling pagination, and building an incremental sync that avoids re-indexing unchanged pages. By the end you’ll have a production-ready pattern you can drop into any RAG system.

Step 1: Set up Notion integration and credentials

Notion requires an internal integration token with read access to the pages and databases you want to ingest. Create one at https://www.notion.so/my-integrations, then share each target page or database with that integration (use the “Connect to” menu in the page’s three-dot menu).

Store the token in your environment — never hardcode it.

export NOTION_TOKEN="secret_xxxxxxxxxxxxxxxxxxxxxxxx"

If you’re pulling from a database rather than individual pages, note the database ID from the URL: https://www.notion.so/workspace/Database-Title-<DATABASE_ID>?v=...

Step 2: Install dependencies

LlamaIndex packages the Notion reader in a separate integration package. Install it alongside the core library:

pip install llama-index llama-index-readers-notion

Verify the import works:

from llama_index.readers.notion import NotionPageReader
print(NotionPageReader.__module__)  # llama_index.readers.notion.base

Step 3: Basic page ingestion

The simplest usage passes a list of page IDs. You can get these from the Notion URL (https://www.notion.so/Page-Title-<PAGE_ID>) or via the API.

import os
from llama_index.readers.notion import NotionPageReader

reader = NotionPageReader(integration_token=os.getenv("NOTION_TOKEN"))

page_ids = [
    "1a2b3c4d5e6f7g8h9i0j1k2l3m4n5o6p",  # replace with real IDs
    "abcdef1234567890abcdef1234567890",
]

documents = reader.load_data(page_ids=page_ids)

print(f"Loaded {len(documents)} documents")
for doc in documents[:2]:
    print(f"  - {doc.metadata.get('title', 'Untitled')}: {len(doc.text)} chars")

Each Document includes metadata: title, page_id, created_time, last_edited_time, url, and parent_id. The text field contains the rendered page content in markdown-ish format (headings, lists, code blocks preserved).

Verify success: You should see non-zero character counts and recognizable titles. If documents is empty, confirm the integration has access to those pages.

Step 4: Load from a database (with filtering)

Most real workloads pull from a Notion database. The reader accepts a database_id and optional filter dict that maps directly to Notion’s filter API.

from llama_index.readers.notion import NotionPageReader

reader = NotionPageReader(integration_token=os.getenv("NOTION_TOKEN"))

# Example: only pages where Status == "Published" and Tag contains "engineering"
filter_params = {
    "and": [
        {"property": "Status", "select": {"equals": "Published"}},
        {"property": "Tags", "multi_select": {"contains": "engineering"}},
    ]
}

documents = reader.load_data(database_id="YOUR_DATABASE_ID", filter=filter_params)
print(f"Loaded {len(documents)} documents from database")

Verify success: Cross-check the count against a manual Notion filter view. The reader handles pagination automatically (100 pages per request).

Step 5: Handle large pages and rate limits

Notion’s API returns page content as a block tree. Very large pages (thousands of blocks) can exceed the default request timeout or hit rate limits (3 requests/second per integration). The reader exposes two knobs:

reader = NotionPageReader(
    integration_token=os.getenv("NOTION_TOKEN"),
    request_timeout=60.0,        # seconds for block-tree fetch
    max_retries=3,               # retry on 429/5xx
)

For bulk ingestion across hundreds of pages, add a small delay between batches:

import time

BATCH_SIZE = 50
all_docs = []

for i in range(0, len(page_ids), BATCH_SIZE):
    batch = page_ids[i:i + BATCH_SIZE]
    docs = reader.load_data(page_ids=batch)
    all_docs.extend(docs)
    time.sleep(0.4)  # stay under 3 req/s

Step 6: Build an incremental sync using last_edited_time

Re-indexing everything on every run wastes tokens and time. Notion exposes last_edited_time on each page; store the high-water mark and filter on the next run.

import json
import os
from pathlib import Path
from llama_index.readers.notion import NotionPageReader

STATE_FILE = Path(".notion_sync_state.json")

def load_state() -> dict:
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"last_sync": None, "page_timestamps": {}}

def save_state(state: dict):
    STATE_FILE.write_text(json.dumps(state, indent=2))

state = load_state()
reader = NotionPageReader(integration_token=os.getenv("NOTION_TOKEN"))

# Fetch all pages from database (no filter = full scan for timestamp check)
all_docs = reader.load_data(database_id="YOUR_DATABASE_ID")

new_docs = []
updated_pages = {}

for doc in all_docs:
    page_id = doc.metadata["page_id"]
    last_edited = doc.metadata["last_edited_time"]  # ISO 8601 string
    
    prev = state["page_timestamps"].get(page_id)
    if prev is None or last_edited > prev:
        new_docs.append(doc)
        updated_pages[page_id] = last_edited

print(f"Total pages: {len(all_docs)}, New/updated: {len(new_docs)}")

if new_docs:
    # --- your indexing logic here ---
    # index.insert_documents(new_docs)
    pass

# Update state
state["page_timestamps"].update(updated_pages)
state["last_sync"] = max(updated_pages.values()) if updated_pages else state["last_sync"]
save_state(state)

Verify success: Run twice with no Notion changes — second run should report “New/updated: 0”. Edit one page in Notion, re-run — it should pick up only that page.

Step 7: Transform documents for vector indexing

Raw Notion documents often need cleaning before embedding: strip navigation chrome, split by heading, or extract code blocks separately. LlamaIndex’s SimpleNodeParser family handles this.

from llama_index.core.node_parser import MarkdownNodeParser
from llama_index.core.schema import Document

parser = MarkdownNodeParser()

nodes = []
for doc in new_docs:
    # Optionally pre-clean: remove Notion-specific artifacts
    text = doc.text.replace("{{TOC}}", "").strip()
    cleaned = Document(text=text, metadata=doc.metadata)
    nodes.extend(parser.get_nodes_from_documents([cleaned]))

print(f"Produced {len(nodes)} nodes from {len(new_docs)} pages")

Each node inherits the page metadata plus section (heading path) and chunk_index. This enables citation-aware retrieval later.

Step 8: Persist to a vector store (example with Chroma)

import chromadb
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore

chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("notion_pages")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)

storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex(nodes, storage_context=storage_context, show_progress=True)

print("Index built and persisted")

Verify success: Query the index directly:

query_engine = index.as_query_engine(similarity_top_k=3)
response = query_engine.query("How do we handle rate limits in the Notion sync?")
print(response)
print("\nSources:")
for src in response.source_nodes:
    print(f"  - {src.metadata.get('title')} (page_id={src.metadata.get('page_id')})")

You should see a coherent answer with source citations pointing back to specific Notion pages.

Step 9: Schedule and monitor

Wrap the incremental sync (Steps 6–8) in a script, containerize it, and run on a schedule (cron, GitHub Actions, Airflow, Prefect). Key observability points:

Latency: Track load_data duration per batch
Throughput: Pages indexed per run
Drift: Alert if last_sync stalls > 24h
Errors: Capture Notion API 429/5xx rates

# Minimal structured log example
import logging
import time

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("notion-sync")

start = time.time()
# ... sync logic ...
duration = time.time() - start
log.info("Sync complete", extra={
    "pages_scanned": len(all_docs),
    "pages_indexed": len(new_docs),
    "duration_sec": round(duration, 1),
})

Step 10: Handle Notion-specific gotchas

Issue	Mitigation
Callout/code block rendering	The reader renders callouts as `> Callout\n> content` and code blocks with language tags. Post-process if your embedder dislikes markdown fences.
Page hierarchy	`parent_id` and `parent_type` (page/database/workspace) let you reconstruct tree structure for breadcrumb metadata.
Deleted pages	Notion’s API doesn’t return deleted pages. Periodically diff `page_timestamps` keys against a fresh `load_data(database_id=...)` call to detect removals.
Property extraction	Database properties (Select, Multi-select, Date, Relation) are not included in `Document.metadata` by default. Extend the reader or query the database API separately if you need them for filtering.
Large workspaces	For 10k+ pages, consider the Notion `search` endpoint with a date filter instead of full database scans.

Putting it together: minimal runnable script

#!/usr/bin/env python3
"""
Incremental Notion → LlamaIndex sync.
Run daily via cron: 0 3 * * * /path/to/sync_notion.py
"""
import os
import json
import time
import logging
from pathlib import Path

from llama_index.readers.notion import NotionPageReader
from llama_index.core.node_parser import MarkdownNodeParser
from llama_index.core import VectorStoreIndex, StorageContext, Document
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("notion-sync")

NOTION_TOKEN = os.getenv("NOTION_TOKEN")
DATABASE_ID = os.getenv("NOTION_DATABASE_ID")
STATE_FILE = Path(".notion_sync_state.json")
CHROMA_PATH = "./chroma_db"

assert NOTION_TOKEN and DATABASE_ID, "Set NOTION_TOKEN and NOTION_DATABASE_ID"

def load_state():
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"page_timestamps": {}}

def save_state(state):
    STATE_FILE.write_text(json.dumps(state, indent=2))

def main():
    state = load_state()
    reader = NotionPageReader(integration_token=NOTION_TOKEN, request_timeout=60.0)
    parser = MarkdownNodeParser()

    log.info("Fetching pages from Notion database")
    all_docs = reader.load_data(database_id=DATABASE_ID)
    log.info(f"Fetched {len(all_docs)} pages")

    new_docs = []
    updated = {}

    for doc in all_docs:
        pid = doc.metadata["page_id"]
        let = doc.metadata["last_edited_time"]
        if state["page_timestamps"].get(pid) != let:
            new_docs.append(doc)
            updated[pid] = let

    if not new_docs:
        log.info("No changes since last sync")
        return

    log.info(f"Processing {len(new_docs)} new/updated pages")
    nodes = []
    for doc in new_docs:
        text = doc.text.replace("{{TOC}}", "").strip()
        cleaned = Document(text=text, metadata=doc.metadata)
        nodes.extend(parser.get_nodes_from_documents([cleaned]))

    log.info(f"Created {len(nodes)} nodes, indexing...")
    chroma = chromadb.PersistentClient(path=CHROMA_PATH)
    collection = chroma.get_or_create_collection("notion_pages")
    vector_store = ChromaVectorStore(chroma_collection=collection)
    storage_context = StorageContext.from_defaults(vector_store=vector_store)

    VectorStoreIndex(nodes, storage_context=storage_context, show_progress=True)

    state["page_timestamps"].update(updated)
    save_state(state)
    log.info("Sync complete")

if __name__ == "__main__":
    main()

Run it once to bootstrap, then schedule. The state file survives container restarts.

This pattern scales to tens of thousands of pages with minimal modification. The key decisions — incremental sync via last_edited_time, markdown-aware chunking, and persistent state — are the same whether you’re indexing a team wiki or a customer-facing knowledge base.