Notion API Integration

Build automations and integrations with Notion workspaces using the official REST API.

Authentication

Internal Integration (own workspace)

Create an integration at https://www.notion.so/my-integrations, copy the token, and share target pages/databases with the integration.

export NOTION_TOKEN="ntn_..."

Public Integration (OAuth)

For multi-user apps, implement the OAuth flow using the authorization URL

https://api.notion.com/v1/oauth/authorize

https://api.notion.com/v1/oauth/token

client_id:client_secret

Core API Patterns

All requests use

Notion-Version: 2022-06-28

Database Operations

"""notion_db.py — Query, create, and update Notion databases."""
import requests

API = "https://api.notion.com/v1"
HEADERS = {
    "Authorization": "Bearer ntn_...",
    "Notion-Version": "2022-06-28",
    "Content-Type": "application/json",
}

def query_database(database_id: str, filter_obj: dict = None, sorts: list = None) -> list:
    """Query a Notion database with optional filters and sorting.

    Args:
        database_id: UUID of the database (from URL or API).
        filter_obj: Notion filter object for narrowing results.
        sorts: List of sort objects (property + direction).

    Returns:
        List of page objects matching the query.
    """
    body = {}
    if filter_obj:
        body["filter"] = filter_obj
    if sorts:
        body["sorts"] = sorts

    pages = []
    has_more = True
    start_cursor = None

    while has_more:
        if start_cursor:
            body["start_cursor"] = start_cursor
        resp = requests.post(f"{API}/databases/{database_id}/query",
                             json=body, headers=HEADERS)
        data = resp.json()
        pages.extend(data["results"])
        has_more = data.get("has_more", False)
        start_cursor = data.get("next_cursor")

    return pages

def create_page(database_id: str, properties: dict, children: list = None) -> dict:
    """Create a new page (row) in a Notion database.

    Args:
        database_id: Target database UUID.
        properties: Property values matching database schema.
        children: Optional list of block objects for page content.
    """
    body = {
        "parent": {"database_id": database_id},
        "properties": properties,
    }
    if children:
        body["children"] = children
    resp = requests.post(f"{API}/pages", json=body, headers=HEADERS)
    return resp.json()

def update_page(page_id: str, properties: dict) -> dict:
    """Update properties of an existing page.

    Args:
        page_id: UUID of the page to update.
        properties: Property values to change.
    """
    resp = requests.patch(f"{API}/pages/{page_id}",
                          json={"properties": properties}, headers=HEADERS)
    return resp.json()

Property Types

Common property value formats for

create_page

update_page

# Property value examples for database pages
properties = {
    # Title (required — every database has one title property)
    "Name": {"title": [{"text": {"content": "New task"}}]},

    # Rich text
    "Description": {"rich_text": [{"text": {"content": "Details here"}}]},

    # Select (single choice)
    "Status": {"select": {"name": "In Progress"}},

    # Multi-select
    "Tags": {"multi_select": [{"name": "frontend"}, {"name": "urgent"}]},

    # Number
    "Story Points": {"number": 5},

    # Date (with optional end for ranges)
    "Due Date": {"date": {"start": "2025-03-15", "end": "2025-03-20"}},

    # Checkbox
    "Done": {"checkbox": True},

    # URL / Email / People / Relation
    "Link": {"url": "https://example.com"},
    "Assignee": {"people": [{"id": "user-uuid-here"}]},
    "Project": {"relation": [{"id": "related-page-uuid"}]},
}

Block Operations

Pages are made of blocks. Append, read, or delete blocks to build page content:

def append_blocks(page_id: str, blocks: list) -> dict:
    """Append content blocks to a page.

    Args:
        page_id: UUID of the page.
        blocks: List of block objects to append.
    """
    resp = requests.patch(f"{API}/blocks/{page_id}/children",
                          json={"children": blocks}, headers=HEADERS)
    return resp.json()

# Block examples
blocks = [
    # Heading
    {"type": "heading_2", "heading_2": {
        "rich_text": [{"text": {"content": "Sprint Summary"}}]
    }},
    # Paragraph
    {"type": "paragraph", "paragraph": {
        "rich_text": [{"text": {"content": "This sprint focused on..."}}]
    }},
    # Bulleted list
    {"type": "bulleted_list_item", "bulleted_list_item": {
        "rich_text": [{"text": {"content": "Shipped auth module"}}]
    }},
    # To-do
    {"type": "to_do", "to_do": {
        "rich_text": [{"text": {"content": "Write tests"}}],
        "checked": False,
    }},
    # Code block
    {"type": "code", "code": {
        "rich_text": [{"text": {"content": "console.log('hello')"}}],
        "language": "javascript",
    }},
]

Search

def search_workspace(query: str, object_type: str = None) -> list:
    """Search across all pages and databases in the workspace.

    Args:
        query: Search text.
        object_type: Optional filter — "page" or "database".
    """
    body = {"query": query}
    if object_type:
        body["filter"] = {"value": object_type, "property": "object"}
    resp = requests.post(f"{API}/search", json=body, headers=HEADERS)
    return resp.json()["results"]

Pagination

All list endpoints return max 100 items. Always paginate:

def get_all_blocks(block_id: str) -> list:
    """Retrieve all child blocks, handling pagination.

    Args:
        block_id: UUID of the parent block or page.
    """
    blocks, cursor = [], None
    while True:
        params = {"page_size": 100}
        if cursor:
            params["start_cursor"] = cursor
        resp = requests.get(f"{API}/blocks/{block_id}/children",
                            params=params, headers=HEADERS)
        data = resp.json()
        blocks.extend(data["results"])
        if not data.get("has_more"):
            break
        cursor = data["next_cursor"]
    return blocks

Rate Limits

• 3 requests per second per integration
• Implement exponential backoff on 429 responses
• Batch operations where possible (append multiple blocks in one call)

import time

def safe_request(method, url, **kwargs):
    """Make a rate-limit-aware request with retry."""
    for attempt in range(5):
        resp = requests.request(method, url, headers=HEADERS, **kwargs)
        if resp.status_code == 429:
            wait = int(resp.headers.get("Retry-After", 2 ** attempt))
            time.sleep(wait)
            continue
        resp.raise_for_status()
        return resp.json()
    raise Exception("Rate limit exceeded after 5 retries")

Common Patterns

Sync External Data into Notion

def sync_github_issues(database_id: str, issues: list[dict]):
    """Sync GitHub issues into a Notion database, updating existing or creating new.

    Args:
        database_id: Target Notion database.
        issues: List of dicts with keys: number, title, state, labels, url.
    """
    # Get existing pages to avoid duplicates
    existing = query_database(database_id)
    existing_numbers = {}
    for page in existing:
        num_prop = page["properties"].get("Issue #", {})
        if num_prop.get("number"):
            existing_numbers[int(num_prop["number"])] = page["id"]

    for issue in issues:
        props = {
            "Name": {"title": [{"text": {"content": issue["title"]}}]},
            "Issue #": {"number": issue["number"]},
            "Status": {"select": {"name": "Open" if issue["state"] == "open" else "Closed"}},
            "Labels": {"multi_select": [{"name": l} for l in issue["labels"]]},
            "URL": {"url": issue["url"]},
        }
        if issue["number"] in existing_numbers:
            update_page(existing_numbers[issue["number"]], props)
        else:
            create_page(database_id, props)

Guidelines

• Always share pages/databases with your integration before accessing them
• Use database queries with filters instead of fetching all pages and filtering client-side
• Notion API returns rich text as arrays of text objects -- always join them for plain text
• Block children can be nested (toggle lists, columns) -- recurse when reading full pages
• The API does not support creating databases with all property types -- some (like rollup, formula) must be configured in the Notion UI
• Archive pages instead of deleting them (

  update_page(id, {"archived": True})

  )