Notion API で​一括登録すると​重複行が​できる​理由｜​現物と​突き合わせる​冪等の​作り方

Notion API で、Excel などの行をデータベースへ一括で登録すると、同じ行が2つ以上できることがあります。

弊社が読んだ Notion の文書には、同じ値の行の作成を断る仕組みが書かれていません。弊社の実測でも、同じキーの行がそのまま作られました。

弊社が踏んだ重複は、どの行をもう作ったかを、データベースの現物でなく手元の記録で判断する作りで起きました。同時に動いた処理どうしで、互いの記録が見えていませんでした。

弊社が試した構成では、数千行の投入を同じ処理3つで同時に流し、データベースの行が約3倍になりました（2026年）。

直し方は、書く前にデータベースの現物からキーを全件読み、無い行だけを作ることです。あわせて、ロックで処理を1本ずつにし、最後に全行を読み戻して照合します。

最後の照合は省けません。弊社が試した構成では、数百行の一括の更新で、要求が200を返したのに書けていない行が1行ありました（2026年9月）。

この記事は、Notion のデータベースへの移行や一括登録を任されている業務の担当者・情報システムの担当者に向けて書きます。コードは、仕組みを作る人にそのまま渡せる形で載せます。

原因の切り分け、見分け方、直し方、落とし穴、確かめ方を、弊社の実測と Notion の公開の文書で整理します。

原因: Notion は​同じ行を​断らず、​作ったかの​判断は​登録する​側に​ある

Notion の文書「Create a page」は、データソース（データベースの表）の下に、新しいページ（行）を作る API です。読んだ範囲では、同じ値の行がすでにあるときに作成を断る仕組みは書かれていません。

冪等とは、同じ処理を何回流しても、結果が1回流したときと同じになることです。送る側が付ける冪等の鍵の説明も、読んだ頁（Create a page・Request limits・Status codes）にはありません。

Create a page の頁に付いた API の定義（OpenAPI）には、409 の誤りの名前として idempotency_key_reused が載っています。使い方の説明は無く、弊社は試していません。

文書「Request limits」は、500・502・503・504の送り直しを、GET や DELETE のような冪等な要求に限っています。アプリに独自の冪等の守りがある場合は除く、と書いています。

同じ文書は、503 を返した書き込みでも、変更が保存されていることがあると書いています。その書き込みを繰り返すと、変更が2回かかります。

ページを作る要求で保存されていたときは、503 の本文の additional_data.committed_resource_id に、新しいページの ID が入ります。ほかの503には、違う案内が入るか、案内が無いこともあります。

文書は、書き込みが保存されたかを Notion がいつも伝えられるわけではないので、送り直す前に対象の今の状態を確かめるよう書いています。文書の内容は、2026年11月1日に確かめたものです。

作ったかを手元の記録で判断すると、記録と現物がずれる場面があります。弊社が実際に踏んだのは、1つ目と3つ目です。

• 同じ処理が同時に動き、互いの記録が見えない（重複になる）
• 作成の応答が返らずに切れ、届いたか分からないまま送り直す（重複になりうる）
• 要求は200を返したのに、書かれていない（抜けになる）

弊社の​実測: 数千行の​投入が、​約3倍の​行に​なった

弊社は、お客様の Notion のデータベースへ、既存のデータを API で投入する仕組みを作っています。その作業の中で、行が重複してできました。

項目	内容
実測の日	2026年
投入した行	数千行
動いた処理	同じ投入の処理が3つ同時
作ったかの判断	作ったキーを記録のファイル（JSON）に残して飛ばす
API の版	記録していない
結果	投入した行の約3倍

3つの処理は、それぞれ起動したときに記録のファイルを読み、その中身を持ったまま走り、終わるときに自分の記録で上書きしていました。このため、ほかの処理が作った行が見えませんでした。

重複した行を1行ずつ消すと、重複した行の数だけ API を呼ぶことになります。弊社は、データベースごとゴミ箱に移しました。

弊社の別の同期は、データベースの現物を API で読み、違う値だけを書き、書いた後にもう一度比べて差が0かを機械で照合する作りです。

弊社は、新しく書く投入の処理を、この作りにそろえています。下の直し方は、その作りで書いた投入の道具から取ったものです。

弊社が試した別の構成では、数百行の一括の更新で、1行だけ書けていませんでした（2026年9月）。処理の記録には更新したと残り、要求は200を返していました。

その行は、ページの最終更新日時（last_edited_time）が動かず、2つの列が空のままでした。最後の読み戻しの照合で見つかり、同じ行の同じ列を送り直すと、1回で入りました。

なぜ書かれなかったかは、切り分けていません。このときの API の版も、記録していません。

見分け方​: 行の​数と​キーの​重なりで、​重複か​抜けかを​分ける

データベースの全行を読み、元のデータとキーで突き合わせます。見え方で、疑う所が変わります。

見え方	疑う所
同じキーの行が2つ以上ある	同時に動いた処理か、作成の送り直し
行の数が元のデータより少ない	途中で止まったか、書けていない行
数は合うが、値が違う行がある	書けていない列（弊社の9月の事例）
元のデータに無いキーの行がある	人の手の追加か、別の処理の書き込み

弊社の9月の事例では、書けていない行は、最終更新日時が書いた時刻より前のままでした。

同時に動いている処理があるかは、プロセスの一覧で見ます。弊社の記録では、ps aux | grep が、コマンド行に日本語のファイル名を含む処理を拾わず、0件と返ったことがありました。

弊社は、pgrep -f を使うか、進捗の実数が増えているかで、処理が動いているかを判定する決まりにしています。

直し方​: 現物の​キーを​読み、​1本ず​つ流し、​最後に​全行を​照合する

手順は次のとおりです。1は重複した件の直しで、2〜6は弊社の投入の道具で、実際に通したものです。

1. 投入の前に、元のデータのキーの列に重複が無いかを確かめる
2. ロックをかけ、同じデータベースへ書く処理を1本ずつにする
3. データベースの現物から、全行のキーを読む（記録のファイルは、何をしたかの控えにする）
4. 現物に無いキーの行だけを作る。同じキーの行が2つ以上あれば、作らずに止める
5. 作成の応答が返らずに切れたら、送り直す前に、そのキーの行ができていないかを問い合わせる
6. 全部を作ったら全行を読み戻し、行の数・キー・値を、元のデータと照合する

弊社はこの直しで、キーにした列の値に重複が無いことを確かめてから、その列をキーに使いました。キーが重複していると、現物と突き合わせても、どの行か決まりません。

元のデータの中の重複そのものを数えて整える話は、Excel台帳のデータベース化にまとめています。そちらは、移す前の中身の層です。

弊社が試した構成では、行の作成（POST /v1/pages）を1本の処理で順に流すと、毎秒約1.1件でした（2026年10月の2つの投入で、毎秒1.08件と1.13件）。4,000行で約62分です。

1.08件の処理は、書き込みを毎秒1.2件以下に抑えていました。この速さは Notion の上限ではなく、弊社の構成での速さです。

弊社の道具は2時間で止まる環境で動くため、1回に作る行の数の上限を渡して、分けて流します。途中で止まっても、同じ指示で流し直せば、現物にある行を飛ばして続きから作ります。

次のコードは、弊社の投入の道具を縮めたものです。Python の標準のライブラリだけで動きます。キーは2列（タイトルの列と、テキストの列）の例です。

import fcntl
import json
import socket
import time
import urllib.error
import urllib.request
from collections import Counter, defaultdict

API = "https://api.notion.com/v1"
TOKEN = "<トークン>"
NV = "2025-09-03"          # 弊社の実装の値
DS = "<データソースの ID>"
KEY_TITLE = "管理番号"      # キーの1列目（タイトルの列）
KEY_TEXT = "区分コード"     # キーの2列目（テキストの列）
LOCK_PATH = ".notion_import.lock"


class Stop(Exception):
    """止めて、人が見る。"""


class Unsure(Exception):
    """書き込みが応答なしで切れた（相手に届いたか分からない）。"""


class Pace:
    """書き込みは毎秒1.2件以下、全体で毎秒1.6件以下にする（弊社の実装の値）。"""

    def __init__(self, all_per_s=1.6, write_per_s=1.2):
        self.gap_all, self.gap_w = 1.0 / all_per_s, 1.0 / write_per_s
        self.t_all = self.t_w = 0.0

    def wait(self, write):
        now = time.monotonic()
        at = self.t_all + self.gap_all
        if write:
            at = max(at, self.t_w + self.gap_w)
        if at > now:
            time.sleep(at - now)
        self.t_all = time.monotonic()
        if write:
            self.t_w = self.t_all


PACE = Pace()


def api(path, method="GET", body=None, tries=8, unsure_ok=True):
    """unsure_ok=False の書き込み（行の作成）は、応答なしの切断・409・5xx で送り直さず Unsure を上げる。"""
    write = method in ("POST", "PATCH", "DELETE") and not path.endswith("/query")
    last = None
    for attempt in range(tries):
        PACE.wait(write)
        data = json.dumps(body, ensure_ascii=False).encode() if body is not None else None
        req = urllib.request.Request(API + path, method=method, data=data,
                                     headers={"Authorization": "Bearer " + TOKEN, "Notion-Version": NV,
                                              "Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(req, timeout=90) as r:
                return json.load(r)
        except urllib.error.HTTPError as e:
            last = "HTTP %s: %s" % (e.code, e.read().decode("utf-8", "replace")[:300])
            if e.code == 429:
                time.sleep(float(e.headers.get("Retry-After", "5")) + 1)
                continue
            if e.code == 409 or e.code >= 500:
                if write and not unsure_ok:
                    raise Unsure(last)
                time.sleep(min(60, 2 ** attempt))
                continue
            raise RuntimeError("%s %s → %s" % (method, path, last))
        except (urllib.error.URLError, TimeoutError, ConnectionError, socket.timeout) as e:
            last = repr(e)
            if write and not unsure_ok:
                raise Unsure(last)
            time.sleep(min(60, 2 ** attempt))
    raise RuntimeError("%s %s → 送り直しの上限: %s" % (method, path, last))


def lock():
    fh = open(LOCK_PATH, "w")
    try:
        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        fh.close()
        raise Stop("別の処理がこのデータベースに書いている（ロック中）")
    return fh


def query_all(flt=None):
    out, cur = [], None
    while True:
        body = {"page_size": 100}
        if flt:
            body["filter"] = flt
        if cur:
            body["start_cursor"] = cur
        r = api("/data_sources/%s/query" % DS, "POST", body)
        if (r.get("request_status") or {}).get("type") == "incomplete":
            raise Stop("問い合わせが1万件で打ち切られた（作成日時などで分けて読む）")
        out += r["results"]
        if not r.get("has_more"):
            return out
        cur = r["next_cursor"]


def plain(rts):
    return "".join(x.get("plain_text", "") for x in rts or [])


def row_key(row):
    return (row[KEY_TEXT], row[KEY_TITLE])


def page_key(pg):
    g = pg["properties"]
    return (plain((g.get(KEY_TEXT) or {}).get("rich_text")), plain((g.get(KEY_TITLE) or {}).get("title")))


def rt(v):
    return [{"type": "text", "text": {"content": v}}] if v else []


def to_props(row):
    return {k: {"title": rt(v)} if k == KEY_TITLE else {"rich_text": rt(v)} for k, v in row.items()}


def got_values(pg):
    g = {}
    for name, p in pg["properties"].items():
        if p["type"] in ("title", "rich_text"):
            g[name] = plain(p[p["type"]])
    return g


def check_source_keys(rows):
    dup = [k for k, n in Counter(row_key(r) for r in rows).items() if n > 1]
    if dup:
        raise Stop("元のデータでキーが重複している: %s" % dup[:5])


def row_state(rows, bot):
    """(作る行, 値が違う行, 合っている行の数, 止める理由, この道具の行の数, ほかが作った行の数)"""
    pages = query_all()
    mine = [pg for pg in pages if (pg.get("created_by") or {}).get("id") == bot]
    by = defaultdict(list)
    for pg in mine:
        by[page_key(pg)].append(pg)
    stop = []
    dup = {k: len(v) for k, v in by.items() if len(v) > 1}
    if dup:
        stop.append("同じキーの行が2つ以上: %s" % list(dup.items())[:5])
    keys = {row_key(r) for r in rows}
    extra = [k for k in by if k not in keys]
    if extra:
        stop.append("元のデータに無いキーの行を、この道具が作っている: %s" % extra[:10])
    create, fix, same = [], [], 0
    for r in rows:
        have = by.get(row_key(r))
        if not have:
            create.append(r)
            continue
        got = got_values(have[0])
        d = [k for k in r if r[k] != got.get(k)]
        if d:
            fix.append((have[0], r, d))
        else:
            same += 1
    return create, fix, same, stop, len(mine), len(pages) - len(mine)


def create_row(row):
    body = {"parent": {"type": "data_source_id", "data_source_id": DS}, "properties": to_props(row)}
    text, title = row_key(row)
    for _attempt in range(4):
        try:
            return api("/pages", "POST", body, unsure_ok=False)["id"], "created"
        except Unsure:
            time.sleep(10)  # 応答なしで切れた: 送り直す前に、同じキーの行ができていないかを問い合わせる
            hits = query_all({"and": [{"property": KEY_TITLE, "title": {"equals": title}},
                                      {"property": KEY_TEXT, "rich_text": {"equals": text}}]})
            if len(hits) == 1:
                return hits[0]["id"], "found_after_unsure"
            if len(hits) > 1:
                raise Stop("応答なしの後に、同じキーの行が %d 件: %s" % (len(hits), row_key(row)))
    raise Stop("行の作成が4回とも応答なしで切れた: %s" % (row_key(row),))


def verify(rows, bot):
    """全行を読み戻して、元のデータと照合する。(合っているか, 内容)"""
    create, fix, same, stop, n_mine, n_others = row_state(rows, bot)
    ok = not create and not fix and not stop and n_mine == len(rows)
    return ok, {"この道具の行": n_mine, "ほかが作った行": n_others, "元のデータの行": len(rows),
                "合っている": same, "無い": len(create), "値が違う": len(fix), "止める理由": stop[:5]}


def run_import(rows, max_create=0):
    """rows は {列の名前: 文字} の並び。戻り値は、次の回に残した行の数（0なら全部入れて照合まで済んだ）。"""
    check_source_keys(rows)
    fh = lock()
    try:
        bot = api("/users/me")["id"]
        create, fix, same, stop, n_mine, n_others = row_state(rows, bot)
        if fix:
            stop.append("値が違う行がある（このコードは直さない）: %s" % [row_key(r) for _p, r, _d in fix][:5])
        if stop:
            raise Stop("現物が止める条件に当たった: %s" % stop)
        print("この道具の行 %d・ほかが作った行 %d・合っている %d・作る %d" % (n_mine, n_others, same, len(create)))
        rest = 0
        if max_create and len(create) > max_create:
            rest, create = len(create) - max_create, create[:max_create]
        for i, r in enumerate(create, 1):
            pid, how = create_row(r)
            print(i, row_key(r), pid, how)
        if rest:
            print("続きあり: 残り %d 行（同じ指示で続きから。全行の照合は全部入れた回にする）" % rest)
            return rest
        ok, info = verify(rows, bot)
        print("全行の読み戻し:", "OK" if ok else "NG", info)
        if not ok:
            raise Stop("全行の読み戻しが合わない")
        return 0
    finally:
        fh.close()

row_state は、データベースの現物を全件読み、この道具の接続が作った行だけをキーで束ねます。この接続が作った行以外は、キーで束ねず、触りません。これは元の道具の決めです。

このため、人が同じキーの行を手で作っていても、道具は別に1行を作ります。照合の結果には、ほかが作った行の数を出すので、その数が0でなければ中身を見ます。

create_row は、作成が応答なしで切れたときや、409・5xx のときに送り直しません。10秒待ってキーで問い合わせ、1行あれば作れていたと見ます。2行以上なら止め、無ければ作り直します（4回まで）。

api は、読み取りと作成以外の書き込みでは、409・5xx と通信の切れを、待ちを延ばしながら8回まで送り直します。429 は、Retry-After の秒数に1秒を足して待ちます。

ロックは、OS のファイルのロック（flock）です。同じパソコンの中の処理どうしでしか効きません。弊社は、別の機械からも動く同期では、共有のデータベースの鍵（PostgreSQL の advisory lock）で1本ずつにしています。

同じデータベースに書く道具が複数あるときは、弊社はロックを道具どうしで共有する決まりにしています。

元の道具から変えたものは、次のとおりです。

• 縮めたもの値は文字の列だけにしました（元は数・日付・複数選択も扱います）。列は、元の列の ID でなく名前で指しています。
• 省いたもの元の道具が残す、作った行と直した行の控えの記録と、データベース・列・タブの作成、先方が書く列の扱いです。
• 変えたもの元の道具は、値が違う行を、最後に編集したのが弊社のときだけ直し、ほかの人なら止めます。このコードは、値が違えば直さずに止めます。
• 足したもの元のデータのキーの重複の確かめ（重複した件の直しの記録から）と、問い合わせの1万件の打ち切りで止める条件（文書から）です。

文書「Query a data source」によると、1回の問い合わせでたどれる結果は1万件までです。超えると has_more が false になり、request_status に打ち切りの印が入ります。

上限は問い合わせ（絞り込みと並べ方）ごとです。1万行を超えるデータベースでは、文書は作成日時の区切りで問い合わせを分けるよう勧めています。このコードは、打ち切りを見つけたら止めるところまでです。

文書の503の案内（committed_resource_id）は、このコードでは使っていません。弊社の道具は、どの切れ方でも、キーでの問い合わせで作れていたかを確かめています。

Notion-Version の値は、弊社の実装のもの（2025-09-03）です。文書が示す最新の版は 2026-03-11 です。429 の待ち方は、下の関連記事に詳しく書いています。

落とし穴: 実際に​踏んだ​もの

• 記録のファイルは、同時に動く処理どうしで見えなかった起動したときに読み、終わるときに上書きする作りでは、ほかの処理が作った行が記録に残りませんでした。
• 重複した行は、消すほうが重い重複した行を1行ずつ消すと、その数だけ呼び出しになります。弊社は、データベースごとゴミ箱に移しました。
• 200 が返っても、書けていない行があった数百行の更新で1行、2つの列が空のままでした。最後の読み戻しの照合で見つかりました。
• ps aux | grep が、動いている処理を0件と返したコマンド行に日本語のファイル名を含む処理でした。止まったと見て、次の処理を起動したことがあります。

確かめ方​: 行の​数・​キーの​重なり・​値を、​全行で​見る

直ったかは、全行を読み戻して、元のデータと照合して判定します。上のコードの verify は、次の5つがそろったときだけ合格にします。

1. 元のデータの行が、全部ある
2. 同じキーの行が、2つ以上ない
3. 元のデータに無いキーの行を、この道具が作っていない
4. 値が、元のデータと一致する
5. この道具が作った行の数が、元のデータの行の数と同じ

弊社の9月の事例の書けていない1行は、読み戻しの照合で見つかりました。分けて流すときは、全部を入れた回に全行を照合します（元の道具の決め）。

同じ指示を2回目に流して、作る行が0になることも見てください。0でなければ、現物に無いキーの行が残っています。

上のコードは、2026年11月1日に Python 3.14.6 で、Notion の応答を作り物の部品に差し替えて確かめました。Notion の本物の API には送っていません。

確かめたのは、空のデータベースに全行を作って照合が通ること、2回目に作る行が0になること、分けて流すと続きから作ることです。

応答なしで切れた作成が届いていたときは作り直さず、届いていなかったときは1回だけ作り直すことも見ました。ロック中の2本目、キーの重複、200 で書けていない行、1万件の打ち切りでは止まることも確かめました。

自力で​どこまで​やれるか

1回だけ、1本の処理で入れ、入れた後に行の数とキーの重なりを数えられるなら、自力で足ります。この記事のコードのように、現物のキーを読んでから作る形にします。

自力の壁は次の3つです。

• 定期の同期や、複数の人・処理が同じデータベースに書く運用で、機械をまたいで処理を1本ずつにする仕組みが要る
• 1万行を超え、問い合わせを分けて読む作りが要る
• 数時間かかる投入を、時間の上限のある環境で分けて流し、途中で止まっても続きから入れる作りが要る

外注する​場合の​判断基準

外注の分かれ目は、移行の後も、同じデータベースへ書き続ける仕組みが要るかです。外注するなら、見積もりを比べる前に次の3点を確かめてください。

• すでにある行の判断を、手元の記録でなく、データベースの現物で行うか
• 同じデータベースへ書く処理を、1本ずつにする仕組みがあるか
• 受け入れを、行の数・キーの重なり・値の全件の照合で行うか

ゴムマリでは、Notion のデータベースの設計と構築、既存のデータの移行、更新の決まりまでを、業務データベースの開発・構築（Notion 対応）として承ります。

既存のシステムとの連携が要る場合は、Notion API や Notion Workers での定期の同期も組み込みます。料金は初期30万円から、保守は月3万円から、期間は2週間からです。

進め方は相談からご連絡ください。

Notion で頼めることの全体は、Notion の構築と開発にまとめています。

次の​一歩

一括登録したデータベースがあれば、全行を読み、キーごとの行の数を数えてください。2行以上のキーがあれば重複で、行の数が元のデータより少なければ抜けです。

そのうえで、投入の処理を、現物のキーを読んで無い行だけを作る形に変え、最後に全行を照合します。

要求が 429 で止まる件は、Notion API が 429 で止まる理由にまとめています。そちらは、要求が通るまでの待ち方の層です。

同期で消した値が残る件は、Salesforce から Notion への同期で消した値が残る理由にまとめています。そちらは、すでにある行の値の正しさの層です。

関連記事: Notion のカスタムエージェントで同じ行が2つできる理由／Notion API が 429 で止まる理由／Salesforce から Notion への同期で消した値が残る理由