airflow-provider-notion 2.11.0.post9

Apache Airflow provider for Notion API integration

Airflow Notion Provider

This provider enables Apache Airflow to integrate with Notion API, allowing you to automate workflows involving Notion databases, pages, and other content.

API Version: This provider uses Notion API version 2025-09-03, which supports multi-source databases. See Migration Guide for upgrading from older versions.

Installation

pip install airflow-provider-notion

Configuration

1. Get your Notion API token from Notion Integrations
   • New tokens use ntn_ prefix (after Sept 2024)
   • Legacy tokens with secret_ prefix still work
2. Set the connection in Airflow:
   • Connection ID: notion_default
   • Connection Type: notion (custom type registered by this provider)
   • Password: YOUR_NOTION_API_TOKEN (format: ntn_xxxxx... or secret_xxxxx...)
   • Extra (optional): {"headers": {"Notion-Version": "2025-09-03"}}

Configuration Methods

Method 1: Airflow UI

Admin → Connections → Add Connection
- Connection Id: notion_default
- Connection Type: notion
- Password: ntn_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

Method 2: Environment Variable

export AIRFLOW_CONN_NOTION_DEFAULT='{"conn_type": "notion", "password": "ntn_YOUR_TOKEN"}'

Method 3: Airflow CLI

airflow connections add notion_default \
    --conn-type notion \
    --conn-password ntn_YOUR_NOTION_TOKEN

Operators

NotionQueryDatabaseOperator

Query a Notion database or data source and return the results.

Note: For API 2025-09-03+, prefer using data_source_id. If only database_id is provided, the operator will automatically discover the first data source.

from airflow.providers.notion.operators import NotionQueryDatabaseOperator

# Recommended: Use data_source_id
query_database = NotionQueryDatabaseOperator(
    task_id='query_notion_datasource',
    data_source_id='your-data-source-id',  # Preferred
    filter_params={
        'property': 'Status',
        'select': {
            'equals': 'Done'
        }
    },
    sorts=[
        {
            'property': 'Created',
            'direction': 'descending'
        }
    ],
    page_size=50,
    dag=dag
)

# Legacy: Auto-discover from database_id (backward compatible)
query_database_legacy = NotionQueryDatabaseOperator(
    task_id='query_notion_database',
    database_id='your-database-id',  # Auto-discovers first data source
    filter_params={
        'property': 'Status',
        'select': {
            'equals': 'Done'
        }
    },
    dag=dag
)

NotionCreatePageOperator

Create a new page in a Notion data source.

Note: For API 2025-09-03+, prefer using data_source_id. If only database_id is provided, the operator will automatically discover the first data source.

from airflow.providers.notion.operators import NotionCreatePageOperator

# Recommended: Use data_source_id
create_page = NotionCreatePageOperator(
    task_id='create_notion_page',
    data_source_id='your-data-source-id',  # Preferred
    properties={
        'Title': {
            'title': [
                {
                    'text': {
                        'content': 'New Task'
                    }
                }
            ]
        },
        'Status': {
            'select': {
                'name': 'In Progress'
            }
        }
    },
    children=[  # Optional page content
        {
            'object': 'block',
            'type': 'paragraph',
            'paragraph': {
                'rich_text': [{'type': 'text', 'text': {'content': 'Page content here'}}]
            }
        }
    ],
    dag=dag
)

# Legacy: Auto-discover from database_id (backward compatible)
create_page_legacy = NotionCreatePageOperator(
    task_id='create_notion_page_legacy',
    database_id='your-database-id',  # Auto-discovers first data source
    properties={...},
    dag=dag
)

NotionUpdatePageOperator

Update an existing Notion page.

from airflow.providers.notion.operators import NotionUpdatePageOperator

update_page = NotionUpdatePageOperator(
    task_id='update_notion_page',
    page_id='your-page-id',
    properties={
        'Status': {
            'select': {
                'name': 'Completed'
            }
        }
    },
    dag=dag
)

NotionSearchOperator

Search pages and databases in your Notion workspace.

Note: The Search API only returns pages and databases that the integration has permission to access.

from airflow.providers.notion.operators import NotionSearchOperator

# Search all pages
search_pages = NotionSearchOperator(
    task_id='search_all_pages',
    filter_object_type='page',  # Only return pages
    sort_direction='descending',  # Most recently edited first
    page_size=100,
    dag=dag
)

# Search with keyword
search_projects = NotionSearchOperator(
    task_id='search_projects',
    query='project',  # Search for pages containing "project"
    filter_object_type='page',
    page_size=50,
    dag=dag
)

# Search all databases
search_databases = NotionSearchOperator(
    task_id='search_databases',
    filter_object_type='database',  # Only return databases
    page_size=100,
    dag=dag
)

# Search everything (pages + databases)
search_all = NotionSearchOperator(
    task_id='search_all',
    page_size=50,  # No filter = returns both
    dag=dag
)

NotionCreateCommentOperator

Create a comment on a Notion page or reply to an existing discussion thread.

Note: The integration must have "Insert comments" capability enabled in the Notion Integrations dashboard.

See: Working with Comments

from airflow.providers.notion.operators import NotionCreateCommentOperator

# Add a simple text comment to a page
add_comment = NotionCreateCommentOperator(
    task_id='add_comment',
    page_id='your-page-id',
    comment_text='Hello from Airflow workflow!',  # Simple text
    dag=dag
)

# Add a rich text comment with formatting
add_rich_comment = NotionCreateCommentOperator(
    task_id='add_rich_comment',
    page_id='your-page-id',
    rich_text=[
        {'type': 'text', 'text': {'content': 'Task completed by '}},
        {'type': 'text', 'text': {'content': 'Airflow'}, 'annotations': {'bold': True}},
        {'type': 'text', 'text': {'content': ' successfully!'}}
    ],
    dag=dag
)

# Reply to an existing discussion thread
reply_to_discussion = NotionCreateCommentOperator(
    task_id='reply_to_discussion',
    discussion_id='your-discussion-id',  # Get from list_comments or Notion UI
    comment_text='This is a reply to the discussion.',
    dag=dag
)

NotionListCommentsOperator

Retrieve all un-resolved (open) comments on a page or block.

Note: The integration must have "Read comments" capability enabled in the Notion Integrations dashboard.

from airflow.providers.notion.operators import NotionListCommentsOperator

# Get all comments on a page
list_comments = NotionListCommentsOperator(
    task_id='list_comments',
    block_id='your-page-id',  # Pages are blocks too
    page_size=100,
    dag=dag
)

# Get comments with pagination (using XCom)
list_more_comments = NotionListCommentsOperator(
    task_id='list_more_comments',
    block_id='your-page-id',
    start_cursor="{{ task_instance.xcom_pull(task_ids='list_comments', key='next_cursor') }}",
    dag=dag
)

Hooks

NotionHook

The base hook for interacting with Notion API (version 2025-09-03).

from airflow.providers.notion.hooks import NotionHook

hook = NotionHook(notion_conn_id='notion_default')

# Search pages and databases
results = hook.search(
    query='project',  # Optional: search keyword
    filter_params={'property': 'object', 'value': 'page'},  # Optional: filter by type
    sort={'direction': 'descending', 'timestamp': 'last_edited_time'},
    page_size=100
)

# Get data sources for a database
db_info = hook.get_data_sources('database-id')
data_sources = db_info.get('data_sources', [])
data_source_id = data_sources[0]['id']

# Query data source (recommended)
results = hook.query_data_source(
    data_source_id='data-source-id',
    filter_params={...},
    sorts=[...],
    page_size=50
)

# Query database (legacy, auto-discovers first data source)
results = hook.query_database(database_id='database-id', filter_params={...})

# Create page with data_source_id (recommended)
page = hook.create_page(
    data_source_id='data-source-id',
    properties={...}
)

# Create page with database_id (legacy, auto-discovers first data source)
page = hook.create_page(database_id='database-id', properties={...})

# Update page (unchanged)
page = hook.update_page(page_id='page-id', properties={...})

# Get page
page = hook.get_page(page_id='page-id')

# Block operations
children = hook.get_block_children(block_id='block-id')
hook.append_block_children(block_id='block-id', children=[...])

# Comment operations (new!)
# See: https://developers.notion.com/docs/working-with-comments

# Add a comment to a page
comment = hook.create_comment(
    page_id='page-id',
    rich_text=[{'type': 'text', 'text': {'content': 'Hello from Airflow!'}}]
)

# Reply to an existing discussion thread
reply = hook.create_comment(
    discussion_id='discussion-id',
    rich_text=[{'type': 'text', 'text': {'content': 'This is a reply.'}}]
)

# List all comments on a page/block
comments = hook.list_comments(block_id='page-id', page_size=100)
for comment in comments.get('results', []):
    print(f"Comment: {comment['rich_text'][0]['plain_text']}")
    print(f"Discussion ID: {comment['discussion_id']}")

Migration from Legacy API

This provider uses Notion API version 2025-09-03, which introduces multi-source databases. Key changes:

What Changed?

1. Database → Data Source paradigm:

   • Old: One database = one data table
   • New: One database can contain multiple data sources (tables)
   • Each data source has its own ID and schema

2. API Endpoints:

   • Old: POST /v1/databases/{database_id}/query
   • New: POST /v1/data_sources/{data_source_id}/query

3. Parent Type for Pages:

   • Old: {"parent": {"database_id": "..."}}
   • New: {"parent": {"data_source_id": "..."}}

How to Migrate?

Option 1: Automatic Migration (Recommended)

• Keep using database_id parameter
• The provider automatically discovers the first data source
• Works for single-source databases (most common case)

# No changes needed - backward compatible
query = NotionQueryDatabaseOperator(
    task_id='query',
    database_id='your-database-id',  # Auto-discovers data_source_id
    ...
)

Option 2: Explicit Data Source IDs

• Get data source ID from database
• Use data_source_id parameter explicitly
• Required for multi-source databases

# Step 1: Get data source ID (one-time setup)
hook = NotionHook()
db_info = hook.get_data_sources('your-database-id')
data_source_id = db_info['data_sources'][0]['id']  # First data source

# Step 2: Use data_source_id in operators
query = NotionQueryDatabaseOperator(
    task_id='query',
    data_source_id=data_source_id,  # Explicit data source
    ...
)

Finding Your Data Source ID:

1. In Notion app: Database Settings → Manage data sources → Copy data source ID
2. Via API: GET /v1/databases/{database_id} returns data_sources array
3. Via Hook: hook.get_data_sources(database_id)

Breaking Changes

If users add a second data source to a database in Notion, integrations using database_id will:

• Still work with automatic discovery (uses first data source)
• May not query the intended data source if multiple exist
• Should be updated to use explicit data_source_id

Development

Install development dependencies:

pip install -e ".[dev]"

Run tests:

pytest tests/

Format code:

black airflow/

Check types:

mypy airflow/

License

Apache License 2.0