Notion APIとPythonで売上とタスクの自動ダッシュボードを作る手順

毎朝の売上・タスク・稼働ログの手入力は、月換算で2〜5時間を消費し、転記ミスがダッシュボードの数字そのものを疑わせる原因になります。これは放置してよい小さな手間ではなく、仕組みで解消すべき作業です。

手動更新に消える時間を数字で把握する

結論から言うと、Notionでタスク管理と売上管理を両方している人ほど、毎朝の更新作業に時間を奪われやすくなります。昨日完了したタスクにチェックを入れ、請求書の金額を売上DBに転記し、稼働時間を集計する。1回あたり5分から15分の作業でも、平日だけで20日続けば月に2〜5時間に積み上がります。

理由はシンプルで、この作業は「見るためのデータ」を作るための手作業であり、本来の仕事とは別のコストだからです。さらに厄介なのは、転記ミスが起きても気づきにくいことです。数字が1件ずれていても、ダッシュボードの見た目は普通に表示されます。ミスが積み重なると「このダッシュボードの数字は本当に合っているのか」という疑いが生まれ、結局Excelで二重管理するようになり、Notion側が形骸化します。

この手入力をなくすために自動化効果が大きいのは、次の3つのユースケースです。

• 売上集計:案件DBから当月の売上合計・成約件数・平均単価を自動で出す
• タスク完了率:タスクDBのステータスを数えて、完了率をリアルタイムで表示する
• 稼働ログ:作業時間やAPI利用量などの記録を、手で書かずにDBへ自動で書き込む

この3つは、どれも「既存DBを読み取り、集計し、別のページに書き込む」という同じ構造を持っています。つまり1本のスクリプトの骨格を作れば、対象のDBとプロパティ名を差し替えるだけで複数の用途に転用できます。次の章では、このスクリプトを動かすためにNotion側で先にやっておくべき準備を説明します。

Notion API連携の下準備:インテグレーションとDB設計

結論として、スクリプトを書く前にインテグレーションの作成とDBへの接続許可を済ませておかないと、コードが正しくても必ずエラーで止まります。この章では、つまずきやすい2点を先に潰しておきます。

1. ブラウザで Notion の my-integrations ページを開き、「新しいインテグレーション」から名前(例:Dashboard Updater)とワークスペースを指定して作成します。
2. 作成直後に表示される Internal Integration Token をコピーしておきます。このトークンは環境変数として扱い、コードに直接書き込みません。
3. 連携したいデータベースをNotion上で開き、右上の「…」メニューから「接続先」を選び、先ほど作ったインテグレーションを追加します。
4. 書き込み先(ダッシュボードページ)にも同じ手順で接続を許可します。読み取り元のDBだけ許可して、書き込み先を許可し忘れるケースが多いので、ここは意識して確認します。

この接続許可を1つでも飛ばすと、スクリプト実行時に「Object not found」というエラーが返ります。これはAPIキーが間違っているのではなく、「トークンは有効だが、そのDBへのアクセス権がない」状態で起きるエラーです。対処の手順は、まずエラーメッセージに含まれるdatabase_idをNotionのURLと照合し、そのIDのDBを開いて接続先の一覧に自分のインテグレーションが入っているかを確認します。入っていなければ、先の手順3・4をやり直します。

もう1つ詰まりやすいのが、プロパティ名の不一致です。Notion APIはプロパティ名を文字列として参照するため、「タスク完了率(%)」のように全角括弧やパーセント記号が含まれる名前は、半角との違いやスペースの有無だけでエラーになります。確実な対策は、Notionの管理画面でプロパティ名を選択してコピーし、そのままコードに貼り付けることです。手で入力し直すと、全角スペースが紛れ込んで気づかないまま動かなくなることがあります。

PythonからNotion APIを呼ぶ方法自体については、PythonでNotion APIを実行する方法 | TEMPブログ や NotionのデータをPythonで操る技術:API連携で実現する情報管理の自動化へ - uepon日々の備忘録 が参考になります。

取得・集計・書き込みの全コード

ここでは、実際に動く状態までコードを完成させます。途中で切れた疑似コードではなく、レート制限対策まで含めた形で示します。

まず必要なライブラリを入れます。

pip install notion-client python-dotenv

環境変数は.envファイルにまとめ、Gitリポジトリには含めません。

# .env
NOTION_TOKEN=secret_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TASK_DB_ID=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
SALES_DB_ID=yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy
DASHBOARD_PAGE_ID=zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz

各IDはNotionでDBを開いたときのURLから取得できます。?v=より前の32桁がデータベースIDです。

次に、ページネーション対応の取得処理と、レート制限(1秒あたり3リクエストまで)を守るためのスリープ処理を含めたコードです。

import os
import time
import logging
from datetime import datetime, timezone
from dotenv import load_dotenv
from notion_client import Client
from notion_client.errors import APIResponseError

load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

notion = Client(auth=os.environ["NOTION_TOKEN"])

REQUEST_INTERVAL_SEC = 0.35  # 1秒3リクエストの上限を守るための最小間隔


def get_current_month_range():
    now = datetime.now(timezone.utc)
    start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    if now.month == 12:
        end = now.replace(year=now.year + 1, month=1, day=1, hour=0, minute=0, second=0, microsecond=0)
    else:
        end = now.replace(month=now.month + 1, day=1, hour=0, minute=0, second=0, microsecond=0)
    return start.isoformat(), end.isoformat()


def query_all_pages(database_id, filter_obj):
    results = []
    cursor = None
    while True:
        params = {"database_id": database_id, "filter": filter_obj, "page_size": 100}
        if cursor:
            params["start_cursor"] = cursor
        try:
            response = notion.databases.query(**params)
        except APIResponseError as e:
            if e.code == "rate_limited":
                logger.warning("rate limited, sleeping 1 second")
                time.sleep(1.0)
                continue
            logger.error(f"query error: {e}")
            break
        results.extend(response["results"])
        time.sleep(REQUEST_INTERVAL_SEC)
        if not response.get("has_more"):
            break
        cursor = response.get("next_cursor")
    logger.info(f"fetched {len(results)} pages from {database_id}")
    return results


def aggregate_monthly_data():
    start_date, end_date = get_current_month_range()

    task_filter = {
        "and": [
            {"property": "作成日時", "date": {"on_or_after": start_date}},
            {"property": "作成日時", "date": {"before": end_date}},
        ]
    }
    task_pages = query_all_pages(os.environ["TASK_DB_ID"], task_filter)
    total_tasks = len(task_pages)
    completed_tasks = sum(
        1 for p in task_pages
        if p["properties"].get("ステータス", {}).get("status", {}).get("name") == "完了"
    )

    sales_filter = {
        "and": [
            {"property": "請求日", "date": {"on_or_after": start_date}},
            {"property": "請求日", "date": {"before": end_date}},
        ]
    }
    sales_pages = query_all_pages(os.environ["SALES_DB_ID"], sales_filter)
    total_revenue = sum(
        p["properties"].get("金額", {}).get("number") or 0 for p in sales_pages
    )
    deal_count = len(sales_pages)
    avg_deal = round(total_revenue / deal_count, 2) if deal_count > 0 else 0.0

    return {
        "total_tasks": total_tasks,
        "completed_tasks": completed_tasks,
        "completion_rate": round(completed_tasks / total_tasks * 100, 1) if total_tasks > 0 else 0.0,
        "total_revenue": total_revenue,
        "deal_count": deal_count,
        "avg_deal_value": avg_deal,
    }


def update_dashboard(summary):
    page_id = os.environ["DASHBOARD_PAGE_ID"]
    properties = {
        "月間タスク数": {"number": summary["total_tasks"]},
        "完了タスク数": {"number": summary["completed_tasks"]},
        "タスク完了率(%)": {"number": summary["completion_rate"]},
        "月間売上合計": {"number": summary["total_revenue"]},
        "案件数": {"number": summary["deal_count"]},
        "平均案件単価": {"number": summary["avg_deal_value"]},
        "最終更新日時": {"date": {"start": datetime.now(timezone.utc).isoformat()}},
    }
    try:
        notion.pages.update(page_id=page_id, properties=properties)
        logger.info("dashboard updated")
    except APIResponseError as e:
        logger.error(f"update failed: {e}")
        raise


def main():
    summary = aggregate_monthly_data()
    update_dashboard(summary)
    logger.info(f"done: {summary}")


if __name__ == "__main__":
    main()

処理の流れは、当月範囲を計算し、タスクDBと売上DBをページネーションつきで取得し、集計し、ダッシュボードページへpatchで書き込む、という順番です。query_all_pagesの中にtime.sleep(0.35)を入れているのが今回のレート制限対策で、1回のリクエストごとに0.35秒待つことで1秒3リクエストの上限内に収めています。rate_limitedエラーが返ってきた場合は1秒待って同じリクエストをやり直す処理も入れてあります。件数が多いDBほどこのスリープの合計時間が効いてくるため、取得に時間がかかっても異常ではありません。

詰まりやすい場所と自社での運用実例

結論として、Notion APIの定期実行だけでは「人の確認が要る工程」までは自動化できません。どこまでをスクリプトに任せ、どこから先を人の目に残すかを最初に切り分けておくことが、運用を止めないコツです。

私のサイトはCloudflare Pagesで公開しており、定時処理はWindowsのタスクスケジューラとPythonで動かし、データはSupabaseに置いています。この構成で、案件収集・応募文の下書き・記事生成・計測まわりを含めて定時タスクを複数本運用しています。内容としては、Webから案件情報を収集し、AIで分類して下書きを作り、LINEのwebhookとCloudflare Workerでの処理を経て、Supabaseにデータをためる、という流れです。ここで重要なのは、公開や送信といった「外に出す」操作の手前で人が確認する工程を残していることです。

この切り分け方は、Notionのダッシュボード自動化にもそのまま使えます。売上集計やタスク完了率の計算は、数字を出すだけなので機械的に任せて問題ありません。一方で、例えば「この案件を成約として計上してよいか」「このタスクは完了扱いにしてよいか」といった判断が絡む工程は、自動化すると誤った数字が静かに積み上がるリスクがあります。判断が要る工程は、スクリプトの出力をいったん人が見てから確定させる形にしておくと安全です。

もう1つの実例が、点検記録のデータ化の仕事です。手書きの点検記録を読み取って業務ツールに登録する作業で、読み取り自体はAIに任せていますが、登録する前の確認は人が行う形にしています。読み取り精度がどれだけ高くても、現場の記録という性質上、誤登録の影響が大きいため、最後の確認だけは残しています。Notionのダッシュボード自動化でも、金額や完了率のように影響が大きい数字については、最初の数週間は自動更新結果を目視で照合し、ずれがないことを確認してから完全に任せる運用が現実的です。

cronでの定期実行と、この先の拡張先

結論として、スクリプトが動くことを確認したら、定期実行の設定と、失敗に気づける最低限の仕組みをセットで用意します。動くかどうかわからない状態で放置すると、気づかないうちに数週間数字が止まっていた、ということが起こります。

Mac/Linuxであればcrontabに登録します。

# 毎朝8時にダッシュボードを更新
0 8 * * * /usr/bin/python3 /home/yourname/notion-updater/dashboard_updater.py >> /home/yourname/notion-updater/cron.log 2>&1

設定時に詰まりやすいのは、cronの実行環境がログインシェルと異なるためpython3のパスが通っていないケースです。which python3で確認したフルパスを指定し、スクリプト側のパスも相対パスではなくフルパスで書きます。ログは標準出力・標準エラーとも1つのファイルにまとめてリダイレクトしておくと、失敗時にその場で原因を追えます。

失敗に気づくための最低限の仕組みとしては、スクリプトの中でtry/except全体を囲み、例外が出たらログに記録したうえで、簡単な通知(メール送信やチャットツールへのWebhook送信)を1行追加しておく形で十分です。通知の仕組み自体を凝らす必要はなく、「動かなかったことが当日中にわかる」状態を作ることが目的です。

この手の定時処理は、動かし続けていると実行回数と費用が積み上がっていきます。1件あたりのコストは小さくても、本数が増えるとログの件数で「止まっていないか」を確認する習慣が必要になります。

Notionのダッシュボード更新が安定して回るようになったら、次に広がるのは「人が手で記録しているものをデータ化する」方向です。例えば点検記録のデータ化のように、紙やメモで残っている記録をAIで読み取り、業務ツールに登録する作業も、構造としては今回のNotion自動化と同じで、読み取り・整形・登録という3段階に分けられます。登録前に人が確認する工程を残す点も含めて、考え方をそのまま応用できます。

補足

私がこの構成を見て最初に手をつけるべきだと思うのは、売上DBかタスクDBのどちらか1つだけを対象に、通知なしの最小構成で1週間動かしてみることです。複数のDBを同時に自動化しようとすると、プロパティ名の不一致やレート制限の挙動が混ざって、どこで止まったのかの切り分けに時間がかかります。

だからこそ最初の1本は、動作確認の手順がすべて目視でたどれる単純な構成にしておくことをすすめます。動かしてみて、プロパティ名の不一致やObject not foundのエラーを一度自分の手で踏んでおくと、2本目以降の追加がずっと速くなります。