定期実行ジョブ（Scheduled Jobs）

TODO（内部向け・リリース前に対応して削除）

リリースまでに対応必須

• このページ全体を見直す（TODOの書き方は基本コンセプトを参照）

定期実行ジョブでできること

Keelson では、アプリの公開だけでなく、コードを定期的に自動実行できます。Web アプリと同じコードベース・同じ keelson.yaml でジョブを管理でき、社内業務の自動化に使えます。

代表的なユースケース:

• 毎朝のレポート生成 — 売上や KPI を集計して /data に保存
• 毎時間の API 同期 — 外部サービスからデータを取得して DB を更新
• 毎日の Slack 通知 — 期限切れタスクや承認待ちのリマインド
• 夜間の CSV 取込 — アップロードされた CSV を一括処理
• 定期的なデータクリーンアップ — 古いログや一時ファイルの削除

Scheduled Jobs は Active Apps の枠を消費しません。Web アプリとは独立してカウントされます。

---

どういう仕組みか

1. keelson.yaml の crons にジョブを定義する
2. デプロイすると、Keelson が定期実行ジョブとして登録する
3. 指定したスケジュールに従って、コマンドが自動で実行される
4. 実行ごとにログが記録される

Web アプリとの関係

ジョブの動作は、Web アプリの有無によって異なります。

構成	動作
cron のみ（command なし）	ジョブごとに独立して実行される
Web アプリ + cron（command あり）	Web アプリと同じ実行環境内でジョブが動く

どちらの構成でも、ジョブの定義方法は同じです。

成功・失敗の追跡

• 実行結果（成功・失敗）はダッシュボードで確認できます
• 標準出力・標準エラーがログとして記録されます
• 実行回数は成功・失敗・リトライ・手動実行をすべて含みます

---

基本設定

keelson.yaml の crons セクションでジョブを定義します。

slug: my-app
runtime: python-slim

crons:
  - name: daily-report
    schedule: "0 9 * * *"
    command: "python report.py"
    timeout: 120

フィールド

フィールド	必須	デフォルト	説明
name	はい	—	ジョブ名。小文字英数字とハイフン、1〜63 文字
schedule	はい	—	cron 式（5フィールド）
command	はい	—	実行するコマンド
timeout	いいえ	300 秒	タイムアウト。1〜3,600 秒

複数ジョブの定義

1つのアプリに最大10個のジョブを定義できます。

crons:
  - name: hourly-sync
    schedule: "0 * * * *"
    command: "python sync.py"

  - name: daily-cleanup
    schedule: "0 3 * * *"
    command: "python cleanup.py"
    timeout: 60

  - name: weekly-report
    schedule: "0 9 * * 1"
    command: "python weekly_report.py"
    timeout: 600

環境変数

トップレベルの env で定義した環境変数は、ジョブの実行時にも利用できます。

env:
  DB_PATH: "/data/main.db"
  SLACK_WEBHOOK_URL: "https://hooks.slack.com/..."

crons:
  - name: notify
    schedule: "0 9 * * *"
    command: "python notify.py"

---

cron 式の書き方

スケジュールは5フィールドの cron 式で指定します。

┌───────────── 分（0-59）
│ ┌─────────── 時（0-23）
│ │ ┌───────── 日（1-31）
│ │ │ ┌─────── 月（1-12）
│ │ │ │ ┌───── 曜日（0-6、0=日曜）
│ │ │ │ │
* * * * *

よく使うスケジュール

やりたいこと	cron 式	説明
毎日 9:00	0 9 * * *	毎日午前9時に実行
30分ごと	*/30 * * * *	毎時0分と30分に実行
毎時間	0 * * * *	毎時0分に実行
10分ごと	*/10 * * * *	10分間隔で実行
平日のみ 9:00	0 9 * * 1-5	月〜金の午前9時に実行
毎月1日 0:00	0 0 1 * *	月初に実行
毎日深夜 3:00	0 3 * * *	夜間バッチ向き
毎分	* * * * *	テスト・デバッグ用

ヒント

cron 式に不慣れな場合は、AI に「毎朝9時に実行する cron 式を教えて」と聞くのが早いです。

---

実用例

毎朝、売上 CSV を集計して保存

• 入力: /data/sales/ 配下の CSV ファイル
• 処理: 当日分を集計し、合計・件数を計算
• 出力: /data/reports/daily-sales-YYYY-MM-DD.json に保存

crons:
  - name: daily-sales
    schedule: "0 8 * * *"
    command: "python aggregate_sales.py"
    timeout: 120

毎時間、外部 API からデータ同期

• 入力: 外部サービスの REST API
• 処理: 最新データを取得し、差分を SQLite に反映
• 出力: /data/main.db の該当テーブルが更新される

crons:
  - name: hourly-sync
    schedule: "0 * * * *"
    command: "python sync_from_api.py"
    timeout: 180

毎日、期限切れデータを削除

• 入力: SQLite のレコード
• 処理: expired_at が過去のレコードを削除
• 出力: 削除件数をログに出力

crons:
  - name: cleanup-expired
    schedule: "0 2 * * *"
    command: "python cleanup_expired.py"
    timeout: 60

毎週、レポート PDF を生成

• 入力: SQLite の週次データ
• 処理: テンプレートからレポートを生成
• 出力: /data/reports/weekly-YYYY-WXX.pdf に保存

crons:
  - name: weekly-report
    schedule: "0 9 * * 1"
    command: "python generate_weekly_report.py"
    timeout: 300

---

安全に運用するための注意点

冪等性を意識する

ジョブは同じスケジュールで繰り返し実行されます。途中で失敗して再実行されても、データが二重に作成されないように設計してください。

# NG — 毎回 INSERT すると重複する
db.execute("INSERT INTO reports (date, total) VALUES (?, ?)", (today, total))

# OK — UPSERT で既存データを上書き
db.execute("""
    INSERT INTO reports (date, total) VALUES (?, ?)
    ON CONFLICT(date) DO UPDATE SET total = excluded.total
""", (today, total))

タイムアウトに注意する

デフォルトのタイムアウトは 300 秒（5分）です。処理に時間がかかる場合は timeout を明示的に設定してください。最大 3,600 秒（1時間）まで指定できます。

失敗時の再実行

ジョブが失敗しても自動リトライは行われません。冪等に設計しておけば、次のスケジュールで自然にリカバリされます。重要なジョブでは、失敗時に Slack 通知を送るなどの仕組みを入れておくと安心です。

外部 API のレート制限

外部 API を呼び出すジョブでは、レート制限に注意してください。短い間隔（毎分など）で外部 API を叩くと、制限に引っかかる場合があります。

ファイルロック

SQLite への書き込みを伴うジョブが、Web アプリと同じデータベースを使う場合、書き込みのタイミングが重なる可能性があります。SQLite は単一ライターの制約があるため、短時間の待ちが発生することがあります。通常の社内ツール規模であれば問題になりませんが、書き込み頻度が高い場合は注意してください。

---

ログとデバッグ

実行ログの確認

ジョブの標準出力（print や console.log）と標準エラーはすべてログとして記録されます。ダッシュボードからジョブ単位で確認できます。

デバッグ時は、処理の進捗やデータの状態を print で出力しておくと原因特定がしやすくなります。

import datetime

print(f"[{datetime.datetime.now()}] ジョブ開始")
# ... 処理 ...
print(f"処理件数: {count}")
print(f"[{datetime.datetime.now()}] ジョブ完了")

失敗時に確認すること

1. 実行ログを見る — エラーメッセージやスタックトレースを確認
2. 環境変数を確認する — API キーやデータベースパスが正しいか
3. タイムアウトを確認する — 処理が timeout の秒数内に終わっているか
4. 手元で再現する — 同じコマンドをローカルで実行してみる

AI にログを渡して修正する

Keelson のジョブログをそのまま AI エージェントに渡せば、エラーの原因特定と修正を依頼できます。

「このジョブのエラーログを確認して、原因を特定して修正してください」

AI エージェントがログを読み、コードの修正と再デプロイまで一連で対応できます。

---

Web アプリとジョブを組み合わせる

Keelson の強みは、Web アプリ・定期ジョブ・永続データを1つのアプリ内で組み合わせられることです。

パターン: 管理画面 + 夜間バッチ

Web アプリで設定やデータを入力し、重い処理は夜間ジョブに任せる構成です。

slug: sales-tool
runtime: python-slim
command: "pip install --user -r requirements.txt && python app.py"
env:
  PORT: "8080"
  DB_PATH: "/data/sales.db"

databases:
  - name: sales
    type: sqlite
    path: /data/sales.db

crons:
  - name: nightly-aggregate
    schedule: "0 2 * * *"
    command: "python aggregate.py"
    timeout: 300

• 日中: 管理画面から売上データを入力
• 深夜: ジョブが集計処理を実行し、レポートデータを更新
• 翌朝: 管理画面で集計結果を閲覧

パターン: アップロード CSV の定期処理

ユーザーが Web UI から CSV をアップロードし、ジョブが定期的に処理する構成です。

crons:
  - name: process-csv
    schedule: "*/30 * * * *"
    command: "python process_inbox.py"
    timeout: 180

• Web アプリ: CSV を /data/inbox/ にアップロード
• ジョブ（30分ごと）: /data/inbox/ の未処理ファイルを処理し、結果を /data/outbox/ に保存
• Web アプリ: 処理結果を一覧表示・ダウンロード

パターン: 日次の AI 要約

ジョブが毎朝データを要約し、Web アプリで閲覧する構成です。

crons:
  - name: daily-summary
    schedule: "0 7 * * *"
    command: "python generate_summary.py"
    timeout: 600

• ジョブ（毎朝 7:00）: 前日のデータを AI API で要約し、結果を SQLite に保存
• Web アプリ: 要約を日付別に閲覧

---

よくある失敗

cron の時刻を勘違いする

cron 式のフィールド順（分・時・日・月・曜日）を間違えやすいです。「9時に実行したい」場合は 0 9 * * * です。9 0 * * * とすると毎日 0:09 に実行されます。

# NG — 毎日 0:09 に実行される
schedule: "9 0 * * *"

# OK — 毎日 9:00 に実行される
schedule: "0 9 * * *"

一時ファイルと永続ファイルを混同する

ジョブの出力を /tmp に保存すると、次回の実行時には消えている場合があります。結果を残したい場合は /data に保存してください。

ローカルでは動くが本番で認証情報が足りない

外部 API を呼び出すジョブで、API キーをローカルの環境変数にだけ設定していると、Keelson 上では認証エラーになります。keelson.yaml の env またはダッシュボードで環境変数を設定してください。

外部 API のレート制限に引っかかる

毎分実行のジョブで外部 API を大量に呼ぶと、レート制限で失敗します。実行間隔を広げるか、ジョブ内でリクエスト数を制御してください。

月間実行回数の上限

Scheduled Jobs にはプランごとに月間の実行回数上限があります（Starter: 3,000回 〜 Business: 80,000回）。上限に達すると、当月の残りの実行はスキップされます。毎分実行（* * * * *）は月間約43,200回になるため、テスト以外では避けてください。

---

サンプル

最小のジョブ定義

Web アプリなしで、ジョブだけを動かす最小構成です。

slug: my-cron
runtime: python-slim
env:
  PYTHONUNBUFFERED: "1"

crons:
  - name: heartbeat
    schedule: "* * * * *"
    command: "python heartbeat.py"
    timeout: 30

heartbeat.py:

import datetime
print(f"OK: {datetime.datetime.now()}")

Python: 売上データを集計して保存する

import sqlite3, json, os, datetime

DB_PATH = os.environ.get("DB_PATH", "/data/main.db")
REPORT_DIR = "/data/reports"
os.makedirs(REPORT_DIR, exist_ok=True)

conn = sqlite3.connect(DB_PATH)
today = datetime.date.today().isoformat()

row = conn.execute(
    "SELECT COUNT(*), SUM(amount) FROM sales WHERE date = ?", (today,)
).fetchone()

report = {"date": today, "count": row[0], "total": row[1] or 0}
report_path = os.path.join(REPORT_DIR, f"daily-{today}.json")

with open(report_path, "w") as f:
    json.dump(report, f, ensure_ascii=False)

print(f"集計完了: {report}")
conn.close()

Node.js: 外部 API からデータを同期する

const Database = require("better-sqlite3");

const DB_PATH = process.env.DB_PATH || "/data/main.db";
const API_URL = process.env.SYNC_API_URL;

async function sync() {
  const res = await fetch(API_URL);
  const items = await res.json();

  const db = new Database(DB_PATH);
  db.exec(`
    CREATE TABLE IF NOT EXISTS synced_items (
      id TEXT PRIMARY KEY,
      data TEXT,
      synced_at DATETIME DEFAULT CURRENT_TIMESTAMP
    )
  `);

  const stmt = db.prepare(`
    INSERT INTO synced_items (id, data) VALUES (?, ?)
    ON CONFLICT(id) DO UPDATE SET data = excluded.data, synced_at = CURRENT_TIMESTAMP
  `);

  for (const item of items) {
    stmt.run(item.id, JSON.stringify(item));
  }

  console.log(`同期完了: ${items.length} 件`);
  db.close();
}

sync().catch((err) => {
  console.error("同期失敗:", err);
  process.exit(1);
});

Python: Slack 通知を送る

import os, json, urllib.request, datetime

SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK_URL"]

message = {
    "text": f"日次レポート準備完了: {datetime.date.today()}"
}

req = urllib.request.Request(
    SLACK_WEBHOOK,
    data=json.dumps(message).encode(),
    headers={"Content-Type": "application/json"},
)
urllib.request.urlopen(req)
print("Slack 通知送信完了")

keelson.yaml:

slug: daily-notifier
runtime: python-slim
env:
  PYTHONUNBUFFERED: "1"
  SLACK_WEBHOOK_URL: "https://hooks.slack.com/services/..."

crons:
  - name: daily-notify
    schedule: "0 9 * * 1-5"
    command: "python notify.py"
    timeout: 30

---

定期実行ジョブが向いているケース

• 定型業務の自動化 — 毎日・毎時間の集計、通知、データ同期
• Web アプリと組み合わせた業務フロー — 日中は入力、夜間にバッチ処理
• 外部サービスとの定期的なデータ連携 — API からの取得や Webhook 送信
• 社内データの定期メンテナンス — 古いデータの削除、レポート生成

向いていないケース

• リアルタイム処理 — イベント駆動で即座に反応する必要がある場合は、Web アプリ内で処理してください
• 1時間を超える長時間処理 — タイムアウトの上限は 3,600 秒です
• 複雑なジョブチェーン — ジョブ間の依存関係やワークフロー制御が必要な場合は、専用のワークフローエンジンを検討してください
• 秒単位の精度が必要 — cron 式は分単位の指定です

---

よくある質問

ジョブが失敗したら自動でリトライされますか？

されません。次のスケジュールで再実行されます。ジョブを冪等に設計しておけば、自然にリカバリされます。

Web アプリなしでジョブだけ動かせますか？

はい。keelson.yaml で command を省略し、crons だけを定義すれば、ジョブ専用のアプリとしてデプロイできます。

ジョブの実行結果をどこで確認できますか？

ダッシュボードから、ジョブごとの実行履歴・成功/失敗のステータス・ログを確認できます。

月間実行回数の上限に達したらどうなりますか？

当月の残りの実行はスキップされます。翌月にリセットされます。上限はプランによって異なります（Starter: 3,000回 〜 Business: 80,000回）。

タイムゾーンはどうなっていますか？

cron 式の時刻はサーバーのタイムゾーンに従います。

---

AI への依頼例

「毎朝9時に売上データを集計するジョブを追加してください。keelson.yaml の crons に定義してください」

「30分ごとに外部 API からデータを取得して SQLite に同期するバッチ処理を作ってください」

「このアプリに、毎日深夜3時に古いデータを削除するクリーンアップジョブを追加してください」