【Notion API】100個以上のブロックをページに追加するためのPythonスクリプト

1
import json
2
import requests
3
from typing import Union
4

5
class NotionAPI:
6
    def __init__(self, token):
7
        self.token = token
8
        self.headers = {
9
            "Authorization": f"Bearer {token}",
10
            "Content-Type": "application/json",
11
            "Notion-Version": "2022-06-28"
12
        }
13
        self.base_url = "<https://api.notion.com/v1>"
14

15
    def fetch_notion(self, endpoint: str, method: str, payload: dict) -> requests.Response:
16
        """Notionへのリクエストを実行する関数"""
17
        url = f"{self.base_url}{endpoint}"
18

19
        if method == "GET":
20
            response = requests.request(
21
                method=method,
22
                url=url,
23
                headers=self.headers,
24
                timeout=60
25
            )
26
        else:
27
            response = requests.request(
28
                method=method,
29
                url=url,
30
                headers=self.headers,
31
                data=json.dumps(payload),
32
                timeout=60
33
            )
34

35
        return response
36

37
    def get_endpoint_to_retrieve_block_children(self, block_id: str) -> str:
38
        """ブロックの子要素を取得するためのエンドポイントを生成する関数"""
39
        return f"/blocks/{block_id}/children"
40

41
def add_blocks_over_threshold(notion_api: NotionAPI, blocks: list, page_id: str = None,
42
                             database_id: str = None, properties: dict = None,
43
                             threshold: int = 100) -> str:
44
    """
45
    100個以上のブロックをNotionページに追加する関数
46

47
    Args:
48
        notion_api: NotionAPIのインスタンス
49
        blocks: 追加するブロックのリスト
50
        page_id: 既存のページIDがある場合に指定
51
        database_id: 新規ページを作成する場合のデータベースID
52
        properties: 新規ページを作成する場合のプロパティ
53
        threshold: 一度に追加するブロックの最大数（デフォルト: 100）
54

55
    Returns:
56
        str: 作成または更新されたページのID
57
    """
58
    if not blocks:
59
        raise ValueError("ブロックリストが空です")
60

61
    # ブロックを閾値（デフォルト100個）ごとに分割
62
    block_chunks = [blocks[i:i + threshold] for i in range(0, len(blocks), threshold)]
63
    print(f"ブロックを{len(block_chunks)}つのチャンクに分割しました")
64

65
    # 最初のチャンク処理
66
    if page_id is None:
67
        # 新規ページ作成
68
        if database_id is None or properties is None:
69
            raise ValueError("新規ページ作成には database_id と properties が必要です")
70

71
        payload = {
72
            "parent": {"database_id": database_id},
73
            "properties": properties,
74
            "children": block_chunks[0]
75
        }
76

77
        response = notion_api.fetch_notion("/pages", "POST", payload)
78

79
        if not response.ok:
80
            print(f"エラー: {response.status_code} - {response.text}")
81
            return None
82

83
        response_data = response.json()
84
        page_id = response_data["id"]
85
        print(f"新規ページを作成しました: {page_id}")
86
    else:
87
        # 既存ページにブロックを追加
88
        endpoint = notion_api.get_endpoint_to_retrieve_block_children(page_id)
89
        payload = {"children": block_chunks[0]}
90

91
        response = notion_api.fetch_notion(endpoint, "PATCH", payload)
92

93
        if not response.ok:
94
            print(f"エラー: {response.status_code} - {response.text}")
95
            return None
96

97
        print(f"最初のブロックチャンクを追加しました")
98

99
    # 残りのチャンクを処理
100
    if len(block_chunks) > 1:
101
        for i, chunk in enumerate(block_chunks[1:], 1):
102
            # 最後に追加されたブロックのIDを取得
103
            endpoint = notion_api.get_endpoint_to_retrieve_block_children(page_id)
104
            response = notion_api.fetch_notion(endpoint, "GET", {})
105

106
            if not response.ok:
107
                print(f"エラー: {response.status_code} - {response.text}")
108
                continue
109

110
            response_data = response.json()
111
            results = response_data.get("results", [])
112

113
            if not results:
114
                print("前回追加したブロックが見つかりません")
115
                continue
116

117
            # 最後のブロックIDを取得
118
            last_block_id = results[-1]["id"]
119

120
            # 次のチャンクを追加（前回追加したブロックの後に追加）
121
            endpoint = notion_api.get_endpoint_to_retrieve_block_children(page_id)
122
            payload = {
123
                "children": chunk,
124
                "after": last_block_id
125
            }
126

127
            response = notion_api.fetch_notion(endpoint, "PATCH", payload)
128

129
            if not response.ok:
130
                print(f"エラー: {response.status_code} - {response.text}")
131
                continue
132

133
            print(f"ブロックチャンク {i+1}/{len(block_chunks)} を追加しました")
134

135
    return page_id
136

137
# 使用例
138
if __name__ == "__main__":
139
    # FIXME: Notion APIトークン
140
    token = "your_notion_api_token"
141
    notion_api = NotionAPI(token)
142

143
    # FIXME: データベースID
144
    database_id = "your_database_id"
145

146
    # FIXME: ページプロパティ: プロパティ名は各自編集してください。
147
    properties = {
148
        "Name": {
149
            "title": [
150
                {
151
                    "text": {
152
                        "content": "100個以上のブロックを持つページ"
153
                    }
154
                }
155
            ]
156
        },
157
        "タグ": {
158
            "select": {
159
                "name": "テスト"
160
            }
161
        }
162
    }
163

164
    # 追加するブロック（例として200個のパラグラフブロック）
165
    blocks = []
166
    for i in range(200):
167
        blocks.append({
168
            "object": "block",
169
            "type": "paragraph",
170
            "paragraph": {
171
                "rich_text": [
172
                    {
173
                        "type": "text",
174
                        "text": {
175
                            "content": f"これはテスト用のパラグラフ {i+1} です。"
176
                        }
177
                    }
178
                ]
179
            }
180
        })
181

182
    # 実行
183
    page_id = add_blocks_over_threshold(
184
        notion_api=notion_api,
185
        blocks=blocks,
186
        database_id=database_id,
187
        properties=properties
188
    )
189

190
    if page_id:
191
        print(f"すべてのブロックが正常に追加されました。ページID: {page_id}")
192
    else:
193
        print("ブロックの追加に失敗しました。")

1
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]:
2
    """複数のペイロードを使用してNotionページを作成・更新する関数"""
3
    # 入力検証（省略）
4

5
    is_duplicated = False
6
    is_updated = False
7
    inserted_page_url = None
8

9
    npg = NotionDailyPropertyGenerator()
10
    notion_product_url = notion_cfg.NOTION_PRODUCT_URL
11
    notion_db_id = notion_cfg.DB_ID_OF_TASKS
12
    notion_view_id = notion_cfg.VIEW_ID_MY_TASKS_POWERING_BOARD
13

14
    nc = NotifyingCategory()
15
    na = NotionAPI()
16
    category_for_content_modified = nc.CONTENT_FOR_NOTHING
17
    page_id = ""
18
    block_id = ""
19
    response = ""
20
    method = "POST"
21

22
    for i, payload in enumerate(payloads):
23
        print(f"Processing payload {i + 1}/{len(payloads)}")
24
        if i == 0:
25
            # 最初のペイロードは新規ページ作成
26
            endpoint = notion_cfg.ENDPOINT_OF_NOTION_API_PAGES
27
        else:
28
            # 2回目以降は既存ページにブロックを追加
29
            method = "PATCH"
30
            endpoint = notion_cfg.ENDPOINT_OF_NOTION_API_BLOCKS
31
            endpoint = na.get_endpoint_to_retrieve_block_children(page_id, endpoint)
32
            payload = {
33
                "children": payload["children"],
34
                "after": block_id
35
            }
36

37
        response = na.fetch_notion(endpoint, method, payload)
38

39
        if response.ok:
40
            # 成功時の処理（省略）
41
            if i == 0:
42
                # 最初のペイロード成功時、ページIDを取得
43
                page_id = npg.get_inserted_notion_page_id_from_response(response)
44
                category_for_content_modified = category_for_content
45
                is_updated = True
46
            else:
47
                # 2回目以降のペイロード成功時、最後のブロックIDを取得
48
                block_id = npg.get_inserted_notion_last_block_id_from_response(response)
49
                block_id = npg.reform_block_id_for_requesting(block_id)
50
                continue
51

52
            # 最後に追加されたブロックのIDを取得
53
            endpoint = notion_cfg.ENDPOINT_OF_NOTION_API_BLOCKS
54
            endpoint = na.get_endpoint_to_retrieve_block_children(page_id, endpoint)
55
            response = na.fetch_notion(endpoint, "GET", payload)
56

57
            if response.ok:
58
                block_id = npg.get_inserted_notion_last_block_id_from_response(response)
59
            else:
60
                print(f"Error: with HTTP status code: {response.status_code} - {response.text}")
61
        else:
62
            print(f"Error: with HTTP status code: {response.status_code} - {response.text}")
63
            inserted_page_url = None
64

65
    return inserted_page_url, is_duplicated, is_updated, category_for_content_modified