NotionのUIでデータベースのプロパティ名を「ステータス」から「進行状況」に変えた翌朝、380ページを更新する夜間バッチのログがHTTP 400で埋まっていました。Notion APIはプロパティを「名前」で指定するため、UIでの1回のリネームがAPI側の全書き込みを無効化します。さらに私はtry/exceptで400を握りつぶしていたので、週次レポートで「3日間、1ページも更新が進んでいない」ことに気づくまで事故に気づけませんでした。この記事では、同じ事故を「バッチの起動時に止める」3つの保険を、コピペで動く完全実装とともに示します。

何が起きたか: UIの1操作がAPIの全指定を無効化する

Notion APIでページを更新するリクエストは、プロパティを「名前」をキーにして組みます。

{
  "properties": {
    "進行状況": { "status": { "name": "処理済み" } },
    "最終更新日": { "date": { "start": "2026-10-22T06:00:00+09:00" } }
  }
}

このキーは「リクエスト時点でそのデータベースに実在する名前」でなければなりません。UIでプロパティ名を変更すると古い名前はスキーマから消えるため、古い名前を送り続けたバッチは全リクエストで400 validation_errorを受け取りました。怖いのは次の3点です。

特徴	何が起きるか
名前はUIで誰でも、いつでも変えられる	API利用者に通知は一切来ない（私の場合はチームメンバーの画面整理）
400はページ単位でなく全リクエストに出る	1回のリネームでバッチ380/380ページが全滅
レスポンスのJSONは初見では読みにくい	400がスキーマ起因だと気づくにはメッセージとのにらめっこが必要

私の事故の発端は、チームメンバーが一覧を見やすくするために「ステータス」を「進行状況」へ変えただけでした。悪意もミスもない、Notionの通常操作です。つまりこれは「誰かがミスをした事故」ではなく「UIとAPIで名前の寿命が違う構造事故」であり、書き込み側が自衛するしかありません。

失敗パターン: try/exceptで握りつぶした400が、事故を3日隠した

最初の失敗は、バッチの更新部にあったこの3行です。

try:
    update_page(page_id, values)
except Exception:
    pass  # 1件失敗してもバッチは続ける

連続処理を止めないための「やさしさ」が、全滅を無音に変えました。400は「今すぐ直せ」の意味なのにログには1行も残らず、翌朝の確認では「バッチは正常終了」とだけ出ていました。実際には3日分、全ページの更新が失敗し続けていました。復旧そのものは失敗分の再実行で済みましたが、原因の切り分けに数時間、週次レポートの訂正にさらに時間がかかりました。

教訓は2つです。**400はリトライや握りつぶしの対象ではなく「設計契約違反」**であり、原因はページの内容ではなくスキーマにあること。待機してよいのはレート制限の429だけです。

HTTP	意味	正しい扱い
400 validation_error	スキーマがリクエストと一致しない	即停止、診断表示（リトライしない）
429 rate_limited	共有上限（3リクエスト/秒）の制限	1秒待って1回だけ再試行

3つの保険の完全実装: notion_batch_guard.py

やることはシンプルです。(1) バッチが書き込む名前を起動時にスキーマへ照合し、(2) 書き込みのキーを名前ではなくスキーマから引いたプロパティIDに変換し、(3) 400が来たら「死亡した名前」と現行の一覧を並べて即停止します。

#!/usr/bin/env python3
# notion_batch_guard.py - Notionバッチがプロパティ名変更の400で全滅するのを止める3つの保険
#   保険1: 起動時スキーマ検証（バッチが書く名前がすべて実在するか）
#   保険2: プロパティIDベースの書き込み（UIでのリネームに強い）
#   保険3: 400を握りつぶさない（死亡した名前と現行一覧を出して即停止）
# 使い方:
#   export NOTION_TOKEN="secret_xxx"
#   python3 notion_batch_guard.py <database_id> --dry-run   # 検証だけ実施（書き込みなし）
#   python3 notion_batch_guard.py <database_id>             # 検証後にバッチ実行
import json
import os
import sys
import time
import urllib.error
import urllib.request

API = "https://api.notion.com/v1"
NOTION_VERSION = "2022-06-01"
TOKEN = os.environ.get("NOTION_TOKEN", "")

# このバッチが書き込むプロパティ名の「契約」。ここにない名前は書かない。
REQUIRED = ["進行状況", "最終更新日"]


def call(method, path, payload=None):
    req = urllib.request.Request(
        API + path,
        data=json.dumps(payload).encode("utf-8") if payload else None,
        method=method,
        headers={
            "Authorization": "Bearer " + TOKEN,
            "Notion-Version": NOTION_VERSION,
            "Content-Type": "application/json",
        },
    )
    try:
        r = urllib.request.urlopen(req, timeout=30)
        raw = r.read().decode("utf-8")
        return r.status, json.loads(raw) if raw.strip() else {}
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "ignore") or "{}"
        try:
            parsed = json.loads(body)
        except json.JSONDecodeError:
            parsed = {"message": body[:400]}
        return e.code, parsed


class Schema:
    """起動時に1回だけスキーマを取得し、名前とプロパティIDのマップを持つ。"""

    def __init__(self, database_id):
        status, db = call("GET", "/databases/" + database_id)
        if status != 200:
            raise SystemExit("STOP: スキーマ取得失敗 HTTP %s: %s"
                             % (status, str(db.get("message", ""))[:300]))
        self.by_name = {name: meta["id"] for name, meta in db["properties"].items()}

    def verify(self, required):
        missing = [n for n in required if n not in self.by_name]
        if missing:
            # 保険3: 「死亡した名前」と「現行の一覧」を並べて即停止
            raise SystemExit(
                "STOP: 実在しないプロパティに書き込もうとしています: %s\n"
                "  現在このデータベースに存在するプロパティ: %s\n"
                "  UIで名前が変わっていないか確認してください"
                % (missing, ", ".join(self.by_name))
            )

    def payload(self, values_by_name):
        """名前ベースの値を、IDをキーにしたpayloadへ変換する（保険2）。"""
        props = {}
        for name, value in values_by_name.items():
            if name not in self.by_name:
                raise SystemExit("STOP: プロパティ「%s」は存在しません。現行: %s"
                                 % (name, ", ".join(self.by_name)))
            props[self.by_name[name]] = value
        return {"properties": props}


def update_page(page_id, payload, retries=1):
    status, res = call("PATCH", "/pages/" + page_id, payload)
    if status == 429 and retries > 0:
        time.sleep(1.0)  # 429はレート制限。1秒待って1回だけ再試行
        return update_page(page_id, payload, retries - 1)
    if status != 200:
        # 保険3: 400はリトライしない。診断を出して止める
        raise SystemExit("STOP: HTTP %s page=%s\n  %s"
                         % (status, page_id, json.dumps(res, ensure_ascii=False)[:400]))
    return res


def main():
    if len(sys.argv) < 2 or not TOKEN:
        print("usage: NOTION_TOKEN=... python3 notion_batch_guard.py <database_id> [--dry-run]")
        return 1
    database_id = sys.argv[1]
    dry_run = "--dry-run" in sys.argv

    schema = Schema(database_id)
    print("スキーマ取得OK: %d プロパティ" % len(schema.by_name))

    # 保険1: 1ページも書く前に「契約」を検証
    schema.verify(REQUIRED)
    for name in REQUIRED:
        print("  保険2: %s -> id=%s" % (name, schema.by_name[name]))
    if dry_run:
        print("検証PASS（書き込みは行っていません）")
        return 0

    status, pages = call("POST", "/databases/%s/query" % database_id,
                         {"page_size": 20})
    if status != 200:
        raise SystemExit("STOP: query失敗 HTTP %s: %s" % (status, str(pages)[:300]))
    for page in pages["results"]:
        values = {
            "進行状況": {"status": {"name": "処理済み"}},
            "最終更新日": {"date": {"start": time.strftime("%Y-%m-%dT%H:%M:%S+09:00")}},
        }
        update_page(page["id"], schema.payload(values))
        print("updated:", page["id"])
        time.sleep(0.35)  # Notionの共有上限3リクエスト/秒を守る間隔
    print("done: %d 件" % len(pages["results"]))
    return 0


if __name__ == "__main__":
    sys.exit(main())
```## 保険の効き方: どこで止まり、何が出るか

**保険1（起動時検証）**: `REQUIRED` は「このバッチが書くプロパティ名の契約」です。`schema.verify(REQUIRED)` が起動時に照合するため、名前が消えていれば1ページ目を書く前に止まります。`--dry-run` は書き込みを伴わないので、UI変更が入った晩に念押しで回せます。

$ python3 notion_batch_guard.py --dry-run
スキーマ取得OK: 9 プロパティ
保険2: 進行状況 -> id=JfEw
保険2: 最終更新日 -> id=HxKd
検証PASS（書き込みは行っていません）


名前が死んでいた晩には、こう止まります。メンバーの変更作業をAPI側で初検知する形になり、「誰がいつ変えたか」の問い合わせが不要になります。

$ python3 notion_batch_guard.py --dry-run
スキーマ取得OK: 9 プロパティ
STOP: 実在しないプロパティに書き込もうとしています: ['ステータス']
現在このデータベースに存在するプロパティ: 進行状況, 担当者, タイトル, ...
UIで名前が変わっていないか確認してください


**保険2（IDベース書き込み）**: `Schema.payload()` は、名前をスキーマから引いたプロパティIDに変換してからpayloadを組みます。Notion APIはpropertiesのキーにIDを指定でき、IDはリネームしても不変です。つまり検証を通過した後に誰かが名前を変えても、書き込み自体は成功し続けます。クエリのfilterやsortでプロパティを指定する箇所も同名で壊れるので、フィルタを使うバッチでは同じマップからIDを渡してください。

**保険3（400の即時診断）**: `update_page` は400をリトライしません。`raise SystemExit` でバッチ全体を止め、レスポンスJSON（消えた名前を含むメッセージ）とpage_idを出力します。「1件だけ失敗して続く」のではなく「0件で即停止」が、夜間バッチの正しい失敗の形です。止まった夜に出力を読めば、対応はREQUIREDの2ワード変更で済みます。

## 比較: 3つの書き方で事故コストがどう変わるか

| 書き方 | 気づく時期 | 復旧コスト | リネーム耐性 |
|---|---|---|---|
| 名前直書き + except: pass（事故時の私） | 週次レポート（3日後） | 3日分の再実行と信頼回復 | ゼロ（全滅を無音化） |
| 名前直書き + 例外を投げる | 最初の400でバッチ停止 | メッセージとのにらめっこに数時間 | ゼロ（止まるが壊れる） |
| 保険1-3（本記事の実装） | 起動時（0ページを書く前にSTOP） | REQUIREDの2ワード更新のみ | あり（IDは不変） |

## まとめ

- NotionのUIでプロパティ名を変える操作は、APIの全書き込みを400で止める。通知は来ないので、書き込み側が自衛するしかない
- 400は握りつぶさず即停止。待機してよいのは429だけ
- **次のアクション**: Notion APIに書き込むコードがあるなら、今晩 `REQUIRED`（書き込む名前の契約）と `--dry-run` 検証を入れる。スキーマ取得は1回で済み、既存バッチへの追加は数行。チームで名前を変える運用があるなら、変更が入った晩にdry-runを回す1行を運用手順に足すこと

> 🎁 [読者特典を受け取る（無料・メール登録）](https://hayama.yuenreach.com/?utm_source=qiita&utm_medium=article&utm_campaign=reader_bonus&utm_content=notion_schema_drift_400)

### 関連記事（同テーマ）

- [Notion APIのレート制限で詰まった話 — 3リクエスト/秒を守る実装と、失敗した設計](https://qiita.com/yukihayama/items/b6fc82730f80bfdcac18)
- [Notionデータベースを極める — タスク管理からリレーションまで](https://qiita.com/yukihayama/items/7179ab121820e9bee709)

**参考書籍:** [Notion完全活用ガイド — すべてを一つにまとめる知的生産術](https://www.amazon.co.jp/dp/B0HDRYPGCM)（Kindle・Kindle Unlimited読み放題対象）（葉山悠希 著）

著者: 葉山悠希 — 書籍シリーズは [Zenn](https://zenn.dev/yukihayama) / [Amazon](https://www.amazon.co.jp/dp/B0HDGKSNY1) で公開中