Python client for the official Notion API
npx @tessl/cli install tessl/pypi-notion-client@2.5.0A comprehensive Python client library for the official Notion API, providing both synchronous and asynchronous interfaces. This package enables seamless integration with Notion's database and page management capabilities, offering full coverage of Notion's API endpoints including databases, pages, blocks, users, comments, and file uploads with built-in support for structured logging, error handling, and proper authentication mechanisms.
pip install notion-clientfrom notion_client import Client, AsyncClientImport error handling classes:
from notion_client import APIErrorCode, APIResponseErrorimport os
from notion_client import Client
# Initialize client with authentication token
notion = Client(auth=os.environ["NOTION_TOKEN"])
# Query a database
response = notion.databases.query(
database_id="897e5a76-ae52-4b48-9fdf-e71f5945d1af",
filter={
"property": "Status",
"select": {
"equals": "Active"
}
}
)
# List users
users = notion.users.list()
# Create a new page
page = notion.pages.create(
parent={"database_id": "897e5a76-ae52-4b48-9fdf-e71f5945d1af"},
properties={
"Title": {
"title": [{"text": {"content": "New Page"}}]
}
}
)import asyncio
from notion_client import AsyncClient
async def main():
notion = AsyncClient(auth=os.environ["NOTION_TOKEN"])
# All the same methods available asynchronously
response = await notion.databases.query(
database_id="897e5a76-ae52-4b48-9fdf-e71f5945d1af"
)
users = await notion.users.list()
await notion.aclose() # Close connection when done
asyncio.run(main())# Synchronous context manager
with Client(auth=os.environ["NOTION_TOKEN"]) as notion:
response = notion.users.list()
# Asynchronous context manager
async with AsyncClient(auth=os.environ["NOTION_TOKEN"]) as notion:
response = await notion.users.list()The notion-client follows a modular endpoint-based architecture:
Client (sync) and AsyncClient (async) provide the main interfaceClient initialization and configuration options including authentication, timeouts, logging, and custom base URLs.
class Client:
def __init__(self, options=None, client=None, **kwargs): ...
def close(self): ...
def request(self, path, method, query=None, body=None, form_data=None, auth=None): ...
class AsyncClient:
def __init__(self, options=None, client=None, **kwargs): ...
async def aclose(self): ...
async def request(self, path, method, query=None, body=None, form_data=None, auth=None): ...
class ClientOptions:
auth: Optional[str] = None
timeout_ms: int = 60_000
base_url: str = "https://api.notion.com"
log_level: int = logging.WARNING
logger: Optional[logging.Logger] = None
notion_version: str = "2022-06-28"Complete interface to all Notion API endpoints including databases, pages, blocks, users, search, comments, and file uploads. Each endpoint provides CRUD operations with proper parameter handling.
# Database operations
def databases.query(database_id, **kwargs): ...
def databases.retrieve(database_id, **kwargs): ...
def databases.create(**kwargs): ...
def databases.update(database_id, **kwargs): ...
# Page operations
def pages.create(**kwargs): ...
def pages.retrieve(page_id, **kwargs): ...
def pages.update(page_id, **kwargs): ...
# Block operations
def blocks.retrieve(block_id, **kwargs): ...
def blocks.update(block_id, **kwargs): ...
def blocks.delete(block_id, **kwargs): ...
def blocks.children.list(block_id, **kwargs): ...
def blocks.children.append(block_id, **kwargs): ...Comprehensive error handling with specific exception types for different API error conditions and HTTP errors.
class APIResponseError(Exception):
code: APIErrorCode
status: int
headers: httpx.Headers
body: str
class APIErrorCode(Enum):
Unauthorized = "unauthorized"
ObjectNotFound = "object_not_found"
RateLimited = "rate_limited"
ValidationError = "validation_error"
# ... and othersUtility functions for pagination, URL handling, response validation, and rich text processing.
def get_url(object_id: str) -> str: ...
def get_id(url: str) -> str: ...
def iterate_paginated_api(function, **kwargs): ...
def collect_paginated_api(function, **kwargs): ...
def is_full_page(response): ...
def is_full_database(response): ...import logging
from typing import TypeVar, Union, Awaitable, Optional
T = TypeVar("T")
SyncAsync = Union[T, Awaitable[T]]The client supports two authentication methods:
# Using integration token
notion = Client(auth="secret_xxxxxxxxxxxxx")
# Using OAuth access token
notion = Client(auth="oauth_token_xxxxxxxxxxxxx")from notion_client.helpers import iterate_paginated_api, collect_paginated_api
# Iterator approach (memory efficient)
for page in iterate_paginated_api(notion.databases.query, database_id="xxx"):
print(page)
# Collect all results (loads everything into memory)
all_pages = collect_paginated_api(notion.databases.query, database_id="xxx")from notion_client import APIErrorCode, APIResponseError
try:
page = notion.pages.retrieve(page_id="invalid-id")
except APIResponseError as error:
if error.code == APIErrorCode.ObjectNotFound:
print("Page not found")
elif error.code == APIErrorCode.Unauthorized:
print("Authentication failed")
else:
print(f"API error: {error}")