はじまり
リサちゃん
あー、入れたいブロックが多過ぎる・・・
135ml
ちゃんとチャンクを分けよう。
Notion APIのブロック数制限とは？
Notion APIを使ってページにブロックを追加する際、一度のリクエストで追加できるブロック数は最大100個に制限されています。この制限を超えるとエラーが発生してしまいます。
例えば、マークダウンから変換した200個のブロックを一度に追加しようとすると・・・、
# これはエラーになる例
response = notion.blocks.children.append(
block_id=page_id,
children=blocks_list # 200個のブロックが含まれている
)
このコードを実行すると、以下のようなエラーが返されます。
Error: Request failed with status code 400
{"object":"error","status":400,"code":"validation_error","message":"children exceeds maximum size of 100 items"}
解決策：複数のリクエストに分割する
この制限を突破するには、ブロックを100個以下のグループに分割して、複数のリクエストに分けて送信する必要があります。
ポイントは以下の2つです。
- 最初のリクエストでページを作成し、その後のリクエストでブロックを追加していく。
- 2回目以降のリクエストでは、前回追加したブロックの後に新しいブロックを追加する。
実装例：100個以上のブロックを追加するPythonスクリプト
それでは、具体的な実装例を見ていきましょう。以下のコードは、Notion APIを使って100個以上のブロックをページに追加するためのPythonスクリプトです。
ちなみにこのコードは、実際に僕が書いた100個以上のブロックを追加する処理をAIに読ませて、AIに作ってもらったものです。（AIモデルはClaud 3.7 Sonnetです。3.5でも試したんですけど、そっちはCline上で僕のスクリプトを読んでくれず、さらにちゃんと実装できていませんでした。リクエストの回数が足りていなかったりしました。）
import json
import requests
from typing import Union
class NotionAPI:
def __init__(self, token):
self.token = token
self.headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Notion-Version": "2022-06-28"
}
self.base_url = "<https://api.notion.com/v1>"
def fetch_notion(self, endpoint: str, method: str, payload: dict) -> requests.Response:
"""Notionへのリクエストを実行する関数"""
url = f"{self.base_url}{endpoint}"
if method == "GET":
response = requests.request(
method=method,
url=url,
headers=self.headers,
timeout=60
)
else:
response = requests.request(
method=method,
url=url,
headers=self.headers,
data=json.dumps(payload),
timeout=60
)
return response
def get_endpoint_to_retrieve_block_children(self, block_id: str) -> str:
"""ブロックの子要素を取得するためのエンドポイントを生成する関数"""
return f"/blocks/{block_id}/children"
def add_blocks_over_threshold(notion_api: NotionAPI, blocks: list, page_id: str = None,
database_id: str = None, properties: dict = None,
threshold: int = 100) -> str:
"""
100個以上のブロックをNotionページに追加する関数
Args:
notion_api: NotionAPIのインスタンス
blocks: 追加するブロックのリスト
page_id: 既存のページIDがある場合に指定
database_id: 新規ページを作成する場合のデータベースID
properties: 新規ページを作成する場合のプロパティ
threshold: 一度に追加するブロックの最大数（デフォルト: 100）
Returns:
str: 作成または更新されたページのID
"""
if not blocks:
raise ValueError("ブロックリストが空です")
# ブロックを閾値（デフォルト100個）ごとに分割
block_chunks = [blocks[i:i + threshold] for i in range(0, len(blocks), threshold)]
print(f"ブロックを{len(block_chunks)}つのチャンクに分割しました")
# 最初のチャンク処理
if page_id is None:
# 新規ページ作成
if database_id is None or properties is None:
raise ValueError("新規ページ作成には database_id と properties が必要です")
payload = {
"parent": {"database_id": database_id},
"properties": properties,
"children": block_chunks[0]
}
response = notion_api.fetch_notion("/pages", "POST", payload)
if not response.ok:
print(f"エラー: {response.status_code} - {response.text}")
return None
response_data = response.json()
page_id = response_data["id"]
print(f"新規ページを作成しました: {page_id}")
else:
# 既存ページにブロックを追加
endpoint = notion_api.get_endpoint_to_retrieve_block_children(page_id)
payload = {"children": block_chunks[0]}
response = notion_api.fetch_notion(endpoint, "PATCH", payload)
if not response.ok:
print(f"エラー: {response.status_code} - {response.text}")
return None
print(f"最初のブロックチャンクを追加しました")
# 残りのチャンクを処理
if len(block_chunks) > 1:
for i, chunk in enumerate(block_chunks[1:], 1):
# 最後に追加されたブロックのIDを取得
endpoint = notion_api.get_endpoint_to_retrieve_block_children(page_id)
response = notion_api.fetch_notion(endpoint, "GET", {})
if not response.ok:
print(f"エラー: {response.status_code} - {response.text}")
continue
response_data = response.json()
results = response_data.get("results", [])
if not results:
print("前回追加したブロックが見つかりません")
continue
# 最後のブロックIDを取得
last_block_id = results[-1]["id"]
# 次のチャンクを追加（前回追加したブロックの後に追加）
endpoint = notion_api.get_endpoint_to_retrieve_block_children(page_id)
payload = {
"children": chunk,
"after": last_block_id
}
response = notion_api.fetch_notion(endpoint, "PATCH", payload)
if not response.ok:
print(f"エラー: {response.status_code} - {response.text}")
continue
print(f"ブロックチャンク {i+1}/{len(block_chunks)} を追加しました")
return page_id
# 使用例
if __name__ == "__main__":
# FIXME: Notion APIトークン
token = "your_notion_api_token"
notion_api = NotionAPI(token)
# FIXME: データベースID
database_id = "your_database_id"
# FIXME: ページプロパティ: プロパティ名は各自編集してください。
properties = {
"Name": {
"title": [
{
"text": {
"content": "100個以上のブロックを持つページ"
}
}
]
},
"タグ": {
"select": {
"name": "テスト"
}
}
}
# 追加するブロック（例として200個のパラグラフブロック）
blocks = []
for i in range(200):
blocks.append({
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [
{
"type": "text",
"text": {
"content": f"これはテスト用のパラグラフ {i+1} です。"
}
}
]
}
})
# 実行
page_id = add_blocks_over_threshold(
notion_api=notion_api,
blocks=blocks,
database_id=database_id,
properties=properties
)
if page_id:
print(f"すべてのブロックが正常に追加されました。ページID: {page_id}")
else:
print("ブロックの追加に失敗しました。")
こんなの書けちゃうんだなあ・・・。APIとのやり取りは苦手だと思ってたんですけど。
上記のスクリプトを動かすと、ターミナル上で以下のように出力されます。
ブロックを2つのチャンクに分割しました
新規ページを作成しました: xxxxxxxx-yyyy-81fb-zzzz-123456a87890
ブロックチャンク 2/2 を追加しました
すべてのブロックが正常に追加されました。ページID: xxxxxxxx-yyyy-81fb-zzzz-123456a87890
追加されたページはこんな感じでした。ちゃんと100個以上の新規ブロックがNotionのページに反映されています。（まあ、パラグラフブロックしか追加していませんが。）
実装のポイント解説
このスクリプトの重要なポイントを詳しく解説します。
1. ブロックの分割
まず、追加するブロックのリストを100個（または指定した閾値）ごとに分割します。これにより、Notion APIの制限を回避できます。
# ブロックを閾値（デフォルト100個）ごとに分割
block_chunks = [blocks[i:i + threshold] for i in range(0, len(blocks), threshold)]
2. 最初のチャンクの処理
最初のチャンクは、新規ページを作成するか、既存のページにブロックを追加するかで処理が異なります。
# 新規ページ作成の場合
payload = {
"parent": {"database_id": database_id},
"properties": properties,
"children": block_chunks[0]
}
response = notion_api.fetch_notion("/pages", "POST", payload)
# 既存ページにブロックを追加する場合
endpoint = notion_api.get_endpoint_to_retrieve_block_children(page_id)
payload = {"children": block_chunks[0]}
response = notion_api.fetch_notion(endpoint, "PATCH", payload)
3. 残りのチャンクの処理
2回目以降のチャンクを追加する際は、前回追加したブロックの後に追加するために、afterパラメータを使用します。
# 最後に追加されたブロックのIDを取得
endpoint = notion_api.get_endpoint_to_retrieve_block_children(page_id)
response = notion_api.fetch_notion(endpoint, "GET", {})
response_data = response.json()
last_block_id = response_data.get("results", [])[-1]["id"]
# 次のチャンクを追加（前回追加したブロックの後に追加）
payload = {
"children": chunk,
"after": last_block_id
}
response = notion_api.fetch_notion(endpoint, "PATCH", payload)
このafterパラメータが重要で、これにより前回追加したブロックの直後に新しいブロックを追加できます。
実際のプロジェクトでの応用例
実際のプロジェクトでは、より複雑な処理が必要になることがあります。例えば、別のプロジェクトでは以下のような実装が行われています。
ちなみに以下のコードは僕が書いたもので、繰り返しているのでコードの長さは短くはなっていますが、AIが実装したコードは繰り返していないので長めになっていました。しかし、AIが作ったコードの方が見やすかったりするかもしれない。
def fetch_notion_to_insert_task_page_over_threshold_of_block(self, payloads: list[dict], category_for_content: str) -> tuple[Union[str, None], bool, bool, str]:
"""複数のペイロードを使用してNotionページを作成・更新する関数"""
# 入力検証（省略）
is_duplicated = False
is_updated = False
inserted_page_url = None
npg = NotionDailyPropertyGenerator()
notion_product_url = notion_cfg.NOTION_PRODUCT_URL
notion_db_id = notion_cfg.DB_ID_OF_TASKS
notion_view_id = notion_cfg.VIEW_ID_MY_TASKS_POWERING_BOARD
nc = NotifyingCategory()
na = NotionAPI()
category_for_content_modified = nc.CONTENT_FOR_NOTHING
page_id = ""
block_id = ""
response = ""
method = "POST"
for i, payload in enumerate(payloads):
print(f"Processing payload {i + 1}/{len(payloads)}")
if i == 0:
# 最初のペイロードは新規ページ作成
endpoint = notion_cfg.ENDPOINT_OF_NOTION_API_PAGES
else:
# 2回目以降は既存ページにブロックを追加
method = "PATCH"
endpoint = notion_cfg.ENDPOINT_OF_NOTION_API_BLOCKS
endpoint = na.get_endpoint_to_retrieve_block_children(page_id, endpoint)
payload = {
"children": payload["children"],
"after": block_id
}
response = na.fetch_notion(endpoint, method, payload)
if response.ok:
# 成功時の処理（省略）
if i == 0:
# 最初のペイロード成功時、ページIDを取得
page_id = npg.get_inserted_notion_page_id_from_response(response)
category_for_content_modified = category_for_content
is_updated = True
else:
# 2回目以降のペイロード成功時、最後のブロックIDを取得
block_id = npg.get_inserted_notion_last_block_id_from_response(response)
block_id = npg.reform_block_id_for_requesting(block_id)
continue
# 最後に追加されたブロックのIDを取得
endpoint = notion_cfg.ENDPOINT_OF_NOTION_API_BLOCKS
endpoint = na.get_endpoint_to_retrieve_block_children(page_id, endpoint)
response = na.fetch_notion(endpoint, "GET", payload)
if response.ok:
block_id = npg.get_inserted_notion_last_block_id_from_response(response)
else:
print(f"Error: with HTTP status code: {response.status_code} - {response.text}")
else:
print(f"Error: with HTTP status code: {response.status_code} - {response.text}")
inserted_page_url = None
return inserted_page_url, is_duplicated, is_updated, category_for_content_modified
このコードは、複数のペイロードを順番に処理し、100個以上のブロックを持つページを作成します。
注意点とベストプラクティス
Notion APIで大量のブロックを追加する際の注意点とノウハウ的なものをいくつか紹介します。
- エラーハンドリングを適切に行う APIリクエストが失敗した場合に適切に対応できるよう、エラーハンドリングを実装しましょう。
- レート制限に注意する Notion APIにはレート制限があります。短時間に大量のリクエストを送ると制限にかかる可能性があるため、必要に応じて遅延を入れましょう。
- ブロック数を事前に確認する 追加するブロック数が閾値を超える場合のみ分割処理を行うようにすると、効率的です。
- ネストされたブロックに注意する 見出しの中に子ブロックがネストされている場合、それらも合計ブロック数にカウントされます。
- PATCHでリクエストする時のボディに注意する PATCHメソッドでリクエストする際に、リクエストボディにプロパティが不要な場合もあります。
まとめ
Notion APIには一度に追加できるブロック数に100個という制限がありますが、適切な実装によってこの制限を回避し、100個以上のブロックを持つページを作成することができます。
ポイントは以下の通りです。
- ブロックを100個以下のチャンクに分割する。
- 最初のチャンクでページを作成し、残りのチャンクは既存ページに追加する。
- 2回目以降のチャンクを追加する際は、
afterパラメータを使用して前回追加したブロックの後に追加する。
この方法を使えば、Markdownから変換した大量のブロックや、プログラムで生成した複雑なページ構造も、Notion APIを使って簡単に作成できます。
おしまい
リサちゃん
そうだよ、これが欲しかったんだよ
135ml
ちゃんとチャンクを分けよう。
以上になります！
コメント