frappe-syntax-scheduler

Frappe スケジューラーとバックグラウンドジョブ

Frappe スケジューラーイベントと Redis Queue（RQ）によるバックグラウンドジョブ処理の決定的な構文リファレンスです。

判断ツリー

Need periodic execution?
├─ Fixed interval (hourly/daily/weekly/monthly) → scheduler_events in hooks.py
├─ Custom cron schedule → scheduler_events.cron in hooks.py
├─ User-configurable interval → Scheduled Job Type DocType
└─ No, triggered by user/event
   ├─ Run method on a specific document → frappe.enqueue_doc()
   ├─ Run standalone function async → frappe.enqueue()
   └─ Run from controller on self → self.queue_action()

クイックリファレンス：スケジューラーイベント（hooks.py）

# hooks.py — ALWAYS run bench migrate after changes
scheduler_events = {
    # Standard events (default queue)
    "all": ["myapp.tasks.every_tick"],           # Every tick [v14: 240s, v15+: 60s]
    "hourly": ["myapp.tasks.hourly_task"],
    "daily": ["myapp.tasks.daily_task"],
    "weekly": ["myapp.tasks.weekly_task"],
    "monthly": ["myapp.tasks.monthly_task"],

    # Long queue events (for heavy processing)
    "hourly_long": ["myapp.tasks.hourly_heavy"],
    "daily_long": ["myapp.tasks.daily_heavy"],
    "weekly_long": ["myapp.tasks.weekly_heavy"],
    "monthly_long": ["myapp.tasks.monthly_heavy"],

    # Cron events (croniter-compatible syntax)
    "cron": {
        "*/15 * * * *": ["myapp.tasks.every_15_min"],
        "0 9 * * 1-5": ["myapp.tasks.weekday_9am"],
        "0 0 1 * *": ["myapp.tasks.first_of_month"],
    }
}

重要：scheduler_events を変更した後は、必ず bench migrate を実行してください。実行しないと、変更は適用されません。

スケジューラーイベントの種類

イベント	頻度	キュー	用途
all	毎ティック［v14：4分、v15+：60秒］	default	頻繁なポーリング
hourly	1時間に1回	default	同期、クリーンアップ
daily	1日に1回	default	レポート、要約
weekly	1週間に1回	default	アーカイブ
monthly	1か月に1回	default	請求、明細書
hourly_long	1時間に1回	long	負荷の高い同期
daily_long	1日に1回	long	大規模なエクスポート
weekly_long	1週間に1回	long	データウェアハウジング
monthly_long	1か月に1回	long	年次レポート
cron	カスタムスケジュール	設定可能	任意のカスタム実行タイミング

Cron 構文

┌───────────── minute (0-59)
│ ┌───────────── hour (0-23)
│ │ ┌───────────── day of month (1-31)
│ │ │ ┌───────────── month (1-12)
│ │ │ │ ┌───────────── day of week (0-6, Sunday=0)
│ │ │ │ │
* * * * *

記号	意味	例
*	任意の値	* * * * * = 毎分
,	リスト	1,15 * * * * = 1分と15分
-	範囲	0 9-17 * * * = 9時から17時まで
/	間隔	*/10 * * * * = 10分ごと

よく使われるパターン：

• 5分ごと：*/5 * * * *
• 平日の9:00：0 9 * * 1-5
• 月曜日の8:00：0 8 * * 1
• 営業時間中は毎時：0 9-17 * * 1-5

クイックリファレンス：frappe.enqueue()

frappe.enqueue(
    method,                      # REQUIRED: function or "dotted.module.path"
    queue="default",             # "short", "default", "long", or custom
    timeout=None,                # Override queue timeout (seconds)
    is_async=True,               # False = run synchronously (skip worker)
    now=False,                   # True = run via frappe.call() directly
    job_id=None,                 # [v15+] Unique ID for deduplication
    enqueue_after_commit=False,  # Wait for DB commit before enqueue
    at_front=False,              # Place at front of queue
    on_success=None,             # Success callback
    on_failure=None,             # Failure callback
    **kwargs                     # Arguments passed to method
)

キューの種類

キュー	デフォルトのタイムアウト	使用するタイミング
short	300s（5分）	タスクが30秒未満
default	300s（5分）	タスクが30秒～5分
long	1500s（25分）	タスクが5～25分
long + カスタムタイムアウト	ユーザー定義	タスクが25分超

# Short queue — quick status update
frappe.enqueue("myapp.tasks.update_status", queue="short", doc=doc.name)

# Long queue — heavy report generation
frappe.enqueue("myapp.tasks.generate_report", queue="long", timeout=3600)

frappe.enqueue_doc()

特定のドキュメント上のコントローラーメソッドをキューに登録します。

frappe.enqueue_doc(
    "Sales Invoice",              # DocType
    "SINV-00001",                 # Document name
    "send_notification",          # Controller method name
    queue="long",
    timeout=600,
    recipient="[email protected]"  # kwargs passed to method
)

コントローラーメソッドには、必ず @frappe.whitelist() デコレーターを付けてください。

class SalesInvoice(Document):
    @frappe.whitelist()
    def send_notification(self, recipient):
        # self is the loaded document
        pass

self.queue_action()

コントローラー内から使用する代替方法：

class SalesOrder(Document):
    def on_submit(self):
        self.queue_action("send_emails", emails=email_list)

    def send_emails(self, emails):
        for email in emails:
            send_mail(email)

ジョブの重複排除

［v15+］推奨パターン

from frappe.utils.background_jobs import is_job_enqueued

job_id = f"import::{doc.name}"
if not is_job_enqueued(job_id):
    frappe.enqueue(
        "myapp.tasks.import_data",
        job_id=job_id,
        doc_name=doc.name
    )
else:
    frappe.msgprint("Import already in progress")

［v14］従来のパターン（新しいコードでは絶対に使用しないでください）

from frappe.core.page.background_jobs.background_jobs import get_info
enqueued = [d.get("job_name") for d in get_info()]
if name not in enqueued:
    frappe.enqueue(..., job_name=name)

エラー処理パターン

バッチジョブでは、レコードごとに必ず try/except を使ってコミットまたはロールバックしてください。

def process_records(records):
    success, errors = 0, 0
    for record in records:
        try:
            process_single(record)
            frappe.db.commit()
            success += 1
        except Exception:
            frappe.db.rollback()
            frappe.log_error(
                frappe.get_traceback(),
                f"Process Error: {record}"
            )
            errors += 1
    return {"success": success, "errors": errors}

再試行パターン

def task_with_retry(data, retry_count=0, max_retries=3):
    try:
        external_api_call(data)
    except Exception:
        if retry_count < max_retries:
            frappe.enqueue(
                "myapp.tasks.task_with_retry",
                queue="default",
                data=data,
                retry_count=retry_count + 1,
                max_retries=max_retries,
                enqueue_after_commit=True
            )
            frappe.log_error(f"Retry {retry_count+1}/{max_retries}", "Task Retry")
        else:
            frappe.log_error(frappe.get_traceback(), f"Failed after {max_retries} retries")
            raise

コールバック

def on_success_handler(job, connection, result, *args, **kwargs):
    frappe.publish_realtime("show_alert", {"message": "Done!"})

def on_failure_handler(job, connection, type, value, traceback):
    frappe.log_error(f"Job {job.id} failed: {value}", "Job Error")

frappe.enqueue(
    "myapp.tasks.risky_task",
    on_success=on_success_handler,
    on_failure=on_failure_handler,
)

進捗状況の更新

def long_task(items, user):
    total = len(items)
    for i, item in enumerate(items):
        process_item(item)
        frappe.publish_realtime(
            "task_progress",
            {"progress": (i + 1) / total * 100, "current": i + 1, "total": total},
            user=user,
        )

ユーザーコンテキスト

重要：スケジューラージョブは Administrator として実行されます。ドキュメントを作成する際は、必ず所有者を明示的に設定してください。

def scheduled_task():
    doc = frappe.new_doc("ToDo")
    doc.owner = "[email protected]"
    doc.insert(ignore_permissions=True)

監視

ツール	目的
bench doctor	スケジューラーの状態、ワーカーの健全性
RQ Worker（DocType）	ワーカーの状態：ビジー／アイドル
RQ Job（DocType）	ジョブの状態、キューによるフィルタリング
Scheduled Job Log（DocType）	実行履歴、エラー
logs/worker.error.log	ワーカーの例外
logs/scheduler.log	スケジューラーの動作

バージョンによる違い

機能	v14	v15+
ティック間隔（all イベント）	約240秒（4分）	約60秒
ティックの設定キー	scheduler_interval	scheduler_tick_interval
重複排除	job_name（非推奨）	job_id + is_job_enqueued()

common_site_config.json でのカスタムティック間隔：

{ "scheduler_tick_interval": 120 }

重要なルール

1. hooks.py の scheduler_events を変更した後は、必ず bench migrate を実行してください
2. 重複排除には、必ず job_id + is_job_enqueued() を使用してください［v15+］
3. タスクの所要時間に応じて、short/default/long のうち適切なキューを必ず選択してください
4. バッチジョブでは、必ずレコードごとにコミットし、エラー時にロールバックしてください
5. スケジューラージョブは Administrator として実行されることを必ず覚えておいてください
6. スケジューラーイベント内で負荷の高い処理を直接実行するのは絶対に避けてください。代わりにキューに登録してください
7. 新しいコードで重複排除に job_name を使用するのは絶対に避けてください（v14 の従来方式）

参考ファイル

• scheduler-events.md：すべてのイベントタイプ、cron 構文、設定
• enqueue-api.md：frappe.enqueue / enqueue_doc API の完全なリファレンス
• queues.md：キューのタイプ、タイムアウト、カスタムキュー、ワーカー
• monitoring.md：RQ DocType、bench doctor、ログファイル、アラート
• error-handling.md：エラーパターン、再試行、バッチ処理
• examples.md：動作する完全な例
• anti-patterns.md：よくある間違いと修正方法

関連項目

• frappe-syntax-hooks — hooks.py の完全なリファレンス
• frappe-core-background — バックグラウンドジョブのアーキテクチャ
• frappe-errors-jobs — ジョブ失敗のデバッグ