ERPNext Syntax: Scheduler & Background Jobs

Deterministic syntax reference for Frappe scheduler events and background job processing.

Quick Reference

Scheduler Events (hooks.py)

# hooks.py
scheduler_events = {
    "all": ["myapp.tasks.every_tick"],
    "hourly": ["myapp.tasks.hourly_task"],
    "daily": ["myapp.tasks.daily_task"],
    "weekly": ["myapp.tasks.weekly_task"],
    "monthly": ["myapp.tasks.monthly_task"],
    "daily_long": ["myapp.tasks.heavy_daily"],  # Long queue
    "cron": {
        "0 9 * * 1-5": ["myapp.tasks.weekday_9am"],
        "*/15 * * * *": ["myapp.tasks.every_15_min"]
    }
}

CRITICAL: After EVERY change to scheduler_events: bench migrate

frappe.enqueue Basics

# Simple
frappe.enqueue("myapp.tasks.process", customer="CUST-001")

# With queue and timeout
frappe.enqueue(
    "myapp.tasks.heavy_task",
    queue="long",
    timeout=3600,
    param="value"
)

# With deduplication (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=doc.name)

Scheduler Event Types

Event	Frequency	Queue
all	Every tick (v14: 4min, v15: 60s)	default
hourly	Per hour	default
daily	Per day	default
weekly	Per week	default
monthly	Per month	default
hourly_long	Per hour	long
daily_long	Per day	long
weekly_long	Per week	long
monthly_long	Per month	long
cron	Custom schedule	configurable

Version difference scheduler tick:

• v14: ~240 seconds (4 min)
• v15: ~60 seconds

Queue Types

Queue	Timeout	Usage
short	300s (5 min)	Quick tasks, UI responses
default	300s (5 min)	Standard tasks
long	1500s (25 min)	Heavy processing, imports

frappe.enqueue Parameters

frappe.enqueue(
    method,                      # REQUIRED: function or module path
    queue="default",             # Queue name
    timeout=None,                # Override timeout (seconds)
    is_async=True,               # False = execute directly
    now=False,                   # True = via frappe.call()
    job_id=None,                 # v15: unique ID for deduplication
    enqueue_after_commit=False,  # Wait for DB commit
    at_front=False,              # Place at front of queue
    on_success=None,             # Success callback
    on_failure=None,             # Failure callback
    **kwargs                     # Arguments for method
)

Job Deduplication

v15+ (Recommended)

from frappe.utils.background_jobs import is_job_enqueued

job_id = f"process::{doc.name}"
if not is_job_enqueued(job_id):
    frappe.enqueue(
        "myapp.tasks.process",
        job_id=job_id,
        doc_name=doc.name
    )

v14 (Deprecated)

# DO NOT USE - only for legacy code
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)

Error Handling Pattern

def process_records(records):
    for record in records:
        try:
            process_single(record)
            frappe.db.commit()  # Commit per success
        except Exception:
            frappe.db.rollback()  # Rollback on error
            frappe.log_error(
                frappe.get_traceback(),
                f"Process Error: {record}"
            )

Callbacks

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}")

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

User Context

IMPORTANT: Scheduler jobs run as Administrator!

def scheduled_task():
    # frappe.session.user = "Administrator"
    
    # Set explicit owner:
    doc = frappe.new_doc("ToDo")
    doc.owner = "user@example.com"
    doc.insert(ignore_permissions=True)

Monitoring

Tool	Description
RQ Worker (DocType)	Worker status, busy/idle
RQ Job (DocType)	Job status, queue filter
bench doctor	Scheduler status overview
Scheduled Job Log	Execution history

Version Differences v14 vs v15

Feature	v14	v15
Tick interval	4 min	60 sec
Config key	scheduler_interval	scheduler_tick_interval
Deduplication	job_name	job_id + is_job_enqueued()

Reference Files

• scheduler-events.md: All event types, cron syntax, configuration
• enqueue-api.md: Complete frappe.enqueue/enqueue_doc API
• queues.md: Queue types, timeouts, custom queues, workers
• examples.md: Complete working examples
• anti-patterns.md: Common mistakes and corrections

Critical Rules

1. ALWAYS bench migrate after hooks.py scheduler_events changes
2. USE job_id + is_job_enqueued() for deduplication (v15)
3. CHOOSE correct queue: short/default/long based on duration
4. COMMIT per successful record, rollback on error
5. REMEMBER that jobs run as Administrator
6. ENQUEUE heavy tasks from scheduler events, don't execute directly