frappe-impl-scheduler

Frappe Scheduler 与后台作业 - 实现

实现定时任务和后台作业的工作流。准确语法请参见 frappe-syntax-scheduler。

版本：兼容 v14/v15/v16

---

主要选择：scheduler_events 还是 frappe.enqueue

WHAT ARE YOU BUILDING?
|
+-- Runs at fixed intervals/times?
|   +-- YES --> scheduler_events (hooks.py)
|   |           Task receives NO arguments
|   |           See: Workflow 1-2
|   |
|   +-- NO --> Triggered by user action or code?
|              +-- YES --> frappe.enqueue()
|              |           Pass any serializable data
|              |           See: Workflow 3-4
|              |
|              +-- NO --> Reconsider requirements

方面	scheduler_events	frappe.enqueue
触发方式	时间／间隔	代码执行
定义位置	hooks.py	Python 代码
参数	无（必须不带参数）	任意可序列化数据
适用场景	每日清理、每小时同步	用户触发的长时间任务
队列控制	事件后缀（_long）	queue= 参数
重启行为	按计划运行	Worker 重启后会丢失

---

应使用哪种 Scheduler 事件类型？

需求	事件键	队列
每次 scheduler tick	all	short（绝不可超过 60 秒）
每小时（<5 分钟）	hourly	short
每小时（5-25 分钟）	hourly_long	long
每日（<5 分钟）	daily	short
每日（5-25 分钟）	daily_long	long
每周（<5 分钟）	weekly	short
每周（5-25 分钟）	weekly_long	long
每月（<5 分钟）	monthly	short
每月（5-25 分钟）	monthly_long	long
自定义计划	cron["expr"]	short

规则：任务超过 5 分钟时，务必使用 *_long 后缀。

---

frappe.enqueue 应使用哪个队列？

队列	默认超时	适用场景
short	300 秒（5 分钟）	快速操作（<1 分钟）
default	300 秒（5 分钟）	标准任务（1-5 分钟）
long	1500 秒（25 分钟）	繁重处理（>5 分钟）

规则：务必显式指定 queue=。绝不要依赖默认值。

---

实现步骤 1：Scheduler 事件

# myapp/tasks.py
import frappe

def daily_cleanup():
    """Daily cleanup - NO parameters allowed."""
    cutoff = frappe.utils.add_days(frappe.utils.nowdate(), -30)
    frappe.db.delete("Error Log", {"creation": ("<", cutoff)})
    frappe.db.commit()

# hooks.py
scheduler_events = {
    "daily": ["myapp.tasks.daily_cleanup"]
}

编辑 hooks.py 后：务必运行 bench migrate。

---

实现步骤 2：后台作业（frappe.enqueue）

# myapp/api.py
import frappe
from frappe.utils.background_jobs import is_job_enqueued

@frappe.whitelist()
def process_documents(doctype, filters):
    job_id = f"process_{doctype}_{frappe.session.user}"

    if is_job_enqueued(job_id):
        return {"message": "Already in progress"}

    frappe.enqueue(
        "myapp.tasks.process_batch",
        queue="long",
        timeout=1800,
        job_id=job_id,
        enqueue_after_commit=True,
        doctype=doctype,
        filters=filters
    )
    return {"status": "queued"}

---

测试定时任务

方法 1：bench execute（直接执行）

# Run the function directly (no queue involved)
bench --site mysite execute myapp.tasks.daily_cleanup

方法 2：bench scheduler（完整的 scheduler 测试）

# Check scheduler status
bench --site mysite scheduler status

# Enable scheduler
bench --site mysite scheduler enable

# Trigger all pending scheduler events NOW
bench --site mysite scheduler trigger

# Run specific event type
bench --site mysite execute frappe.utils.scheduler.trigger --args "['daily']"

方法 3：bench console（交互式）

bench --site mysite console
>>> frappe.enqueue("myapp.tasks.my_task", queue="short", now=True)
# now=True executes synchronously for testing

方法 4：检查 Scheduled Job Type

1. Go to: Setup > Scheduled Job Type
2. Find: myapp.tasks.daily_cleanup
3. Verify: Frequency correct, Stopped = No
4. Click "Run Now" to trigger manually

---

监控

Scheduled Job Log（UI）

Setup > Scheduled Job Log
- Shows every scheduler run with status
- Filter by: status (Success/Failed), creation date
- Check execution time to detect slow tasks

RQ Dashboard

# Start RQ monitor (development)
bench --site mysite rq-dashboard
# Opens at http://localhost:9181

# Show background job status
bench --site mysite show-pending-jobs
bench --site mysite show-failed-jobs

程序化健康检查

def scheduler_health_check():
    failed = frappe.db.count("Scheduled Job Log", {
        "status": "Failed",
        "creation": [">=", frappe.utils.add_to_date(None, hours=-1)]
    })
    if failed > 5:
        frappe.sendmail(
            recipients=["[email protected]"],
            subject="Scheduler Alert: Many failures",
            message=f"{failed} scheduler jobs failed in last hour"
        )

---

定时任务中的错误处理

按记录隔离错误

def sync_all_orders():
    orders = get_pending_orders()
    success, errors = 0, 0

    for order in orders:
        try:
            sync_to_external(order)
            success += 1
        except Exception as e:
            errors += 1
            frappe.db.rollback()
            frappe.log_error(
                f"Sync failed for {order}: {e}",
                "Order Sync Error"
            )
    frappe.db.commit()
    frappe.logger("sync").info(f"{success} ok, {errors} errors")

规则：务必使用 try-except 包裹每条记录的处理。绝不要让单条记录的失败中断整个批次。

---

长时间运行作业模式

自我链式模式（任务超过 25 分钟）

def process_batch(offset=0, batch_size=500, total=None):
    if total is None:
        total = frappe.db.count("Sales Invoice", {"custom_processed": 0})

    records = frappe.get_all("Sales Invoice",
        filters={"custom_processed": 0},
        pluck="name", limit=batch_size)

    if not records:
        return  # Done

    for name in records:
        process_single(name)
    frappe.db.commit()

    remaining = frappe.db.count("Sales Invoice", {"custom_processed": 0})
    if remaining > 0:
        frappe.enqueue(
            "myapp.tasks.process_batch",
            queue="long",
            offset=offset + batch_size,
            batch_size=batch_size,
            total=total
        )

规则：任务超过 25 分钟时，务必拆分为自我链式批次。

---

常见实现模式

电子邮件摘要（每周汇总）

# hooks.py
scheduler_events = {
    "cron": {
        "0 8 * * 1": ["myapp.newsletter.send_weekly_digest"]
    }
}

完整实现请参见 references/examples.md 中的示例 4。

数据清理（每日维护）

scheduler_events = {
    "daily_long": ["myapp.maintenance.daily_database_maintenance"]
}

批量删除模式请参见 references/examples.md 中的示例 1。

报告生成（用户触发）

frappe.enqueue(
    "myapp.tasks.generate_report",
    queue="long",
    timeout=3600,
    job_id=f"report::{frappe.session.user}",
    user=frappe.session.user
)

进度报告请参见 references/workflows.md 中的工作流 6。

---

关键规则

1. Scheduler 任务不接收任何参数——使用设置或硬编码值
2. 修改 hooks.py 后务必运行 bench migrate——注册事件的必要步骤
3. 作业以 Administrator 身份运行——务必显式提交
4. 分批提交——绝不要逐条记录提交（每 100-500 条记录提交一次）
5. 用户触发的作业务必使用 job_id——防止重复
6. 在文档事件中使用 enqueue_after_commit=True——确保数据已存在
7. Scheduler 事件应保持精简——将繁重工作放入后台队列

版本差异

方面	v14	v15	v16
Tick 间隔	240 秒	60 秒	60 秒
作业去重参数	job_name	job_id	job_id
enqueue_doc()	是	是	是
自定义队列	否	是	是

---

参考文件

文件	内容
workflows.md	8 种分步实现模式
decision-tree.md	详细的决策流程图
examples.md	5 个完整可运行的示例
anti-patterns.md	14 个需要避免的常见错误

另请参阅

• frappe-syntax-scheduler - hooks 和 enqueue 的精确语法参考
• frappe-errors-serverscripts - 错误处理模式
• frappe-impl-hooks - Hook 配置模式
• frappe-ops-bench - 用于管理 scheduler 的 Bench 命令
• frappe-ops-performance - 后台作业性能调优
• frappe-testing-unit - 测试计划任务逻辑