/frappe-impl-scheduler

name: frappe-impl-scheduler
description: >
  Use when implementing scheduled tasks and background jobs in Frappe
  v14/v15/v16. Covers hooks.py scheduler_events, frappe.enqueue, queue
  selection, job deduplication, testing with bench execute/scheduler,
  monitoring via Scheduled Job Log and RQ Dashboard, error handling,
  long-running job patterns, email digest, data cleanup, and report
  generation. Keywords: schedule task, background job, cron job, async
  processing, queue selection, job deduplication, scheduler implementation,
  run task automatically, background process, scheduled task not running, async task.
license: MIT
compatibility: "Claude Code, Claude.ai Projects, Claude API. Frappe v14-v16."
metadata:
  author: OpenAEC-Foundation
  version: "2.0"

Frappe Scheduler & Background Jobs - Implementation

Workflow for implementing scheduled tasks and background jobs. For exact syntax, see `frappe-syntax-scheduler`.

**Version**: v14/v15/v16 compatible

---

Main Decision: scheduler_events vs 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

| Aspect | scheduler_events | frappe.enqueue | |--------|------------------|----------------| | Triggered by | Time/interval | Code execution | | Defined in | hooks.py | Python code | | Arguments | NONE (must be parameterless) | Any serializable data | | Use case | Daily cleanup, hourly sync | User-triggered long task | | Queue control | Event suffix (_long) | queue= parameter | | Restart behavior | Runs on schedule | Lost if worker restarts |

---

Which Scheduler Event Type?

| Need | Event Key | Queue | |------|-----------|-------| | Every scheduler tick | `all` | short (NEVER >60s) | | Hourly (<5 min) | `hourly` | short | | Hourly (5-25 min) | `hourly_long` | long | | Daily (<5 min) | `daily` | short | | Daily (5-25 min) | `daily_long` | long | | Weekly (<5 min) | `weekly` | short | | Weekly (5-25 min) | `weekly_long` | long | | Monthly (<5 min) | `monthly` | short | | Monthly (5-25 min) | `monthly_long` | long | | Custom schedule | `cron["expr"]` | short |

**Rule**: ALWAYS use `*_long` suffix for tasks exceeding 5 minutes.

---

Which Queue for frappe.enqueue?

| Queue | Default Timeout | Use For | |-------|-----------------|---------| | `short` | 300s (5 min) | Quick operations (<1 min) | | `default` | 300s (5 min) | Standard tasks (1-5 min) | | `long` | 1500s (25 min) | Heavy processing (>5 min) |

**Rule**: ALWAYS specify `queue=` explicitly. NEVER rely on the default.

---

Implementation Step 1: Scheduler Event

# 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"]
}

**After editing hooks.py**: ALWAYS run `bench migrate`.

---

Implementation Step 2: Background Job (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"}

---

Testing Scheduled Tasks

Method 1: bench execute (direct)

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

Method 2: bench scheduler (full scheduler test)

# 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']"

Method 3: bench console (interactive)

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

Method 4: Check 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

---

Monitoring

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

Programmatic Health Check

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=["admin@example.com"],
            subject="Scheduler Alert: Many failures",
            message=f"{failed} scheduler jobs failed in last hour"
        )

---

Error Handling in Scheduled Tasks

Per-Record Error Isolation

def sync_all_orders():
    orders = get_pending_orders()
    success, er