Frappe Syntax Scheduler

Frappe Scheduler & Background Jobs

Deterministic syntax reference for Frappe scheduler events and background job processing via Redis Queue (RQ).

Decision Tree

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()

Quick Reference: Scheduler Events (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"],
    }
}

CRITICAL: ALWAYS run bench migrate after ANY change to scheduler_events. Without it, changes are NOT applied.

Scheduler Event Types

Cron Syntax

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

Common patterns:

• Every 5 min: */5 * * * *
• Weekdays at 9:00: 0 9 * * 1-5
• Monday at 8:00: 0 8 * * 1
• Business hours hourly: 0 9-17 * * 1-5

Quick Reference: 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
)

Queue Types

# 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()

Enqueue a controller method on a specific document.

frappe.enqueue_doc(
    "Sales Invoice",              # DocType
    "SINV-00001",                 # Document name
    "send_notification",          # Controller method name
    queue="long",
    timeout=600,
    recipient="user@example.com"  # kwargs passed to method
)

The controller method MUST be decorated with @frappe.whitelist():

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

self.queue_action()

Alternative from within a controller:

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)

Job Deduplication

[v15+] Recommended Pattern

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] Legacy Pattern (NEVER use in new 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

ALWAYS use try/except with commit/rollback per record in batch jobs:

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}

Retry Pattern

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

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}", "Job Error")

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

Progress Updates

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,
        )

User Context

CRITICAL: Scheduler jobs run as Administrator. ALWAYS set explicit ownership when creating documents:

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

Monitoring

Version Differences

Custom tick in common_site_config.json:

{ "scheduler_tick_interval": 120 }

Critical Rules

1. ALWAYS run bench migrate after any scheduler_events change in hooks.py 2. ALWAYS use job_id + is_job_enqueued() for deduplication [v15+] 3. ALWAYS choose the correct queue: short/default/long based on task duration 4. ALWAYS commit per record and rollback on error in batch jobs 5. ALWAYS remember that scheduler jobs run as Administrator 6. NEVER run heavy logic directly in a scheduler event — enqueue it instead 7. NEVER use job_name for deduplication in new code (v14 legacy)

Reference Files

• [scheduler-events.md](references/scheduler-events.md): All event types, cron syntax, configuration
• [enqueue-api.md](references/enqueue-api.md): Complete frappe.enqueue / enqueue_doc API
• [queues.md](references/queues.md): Queue types, timeouts, custom queues, workers
• [monitoring.md](references/monitoring.md): RQ DocTypes, bench doctor, log files, alerts
• [error-handling.md](references/error-handling.md): Error patterns, retry, batch processing
• [examples.md](references/examples.md): Complete working examples
• [anti-patterns.md](references/anti-patterns.md): Common mistakes and corrections

See Also

• frappe-syntax-hooks — Full hooks.py reference
• frappe-core-background — Background job architecture
• frappe-errors-jobs — Job failure debugging

Scheduler & Background Jobs — Anti-Patterns

1. Forgetting bench migrate After hooks.py Change

# WRONG — changed scheduler_events but did not run bench migrate
scheduler_events = {
    "daily": ["myapp.tasks.new_task"],  # NOT active until bench migrate!
}

Fix: ALWAYS run bench migrate after ANY change to scheduler_events.

---

2. Heavy Logic Directly in Scheduler Event

# WRONG — blocks the scheduler worker for the entire duration
def daily_export():
    for record in frappe.get_all("Sales Invoice", fields=["*"]):
        generate_pdf(record)      # 30+ minutes
        upload_to_cloud(record)

# CORRECT — scheduler event enqueues the heavy work
def daily_export():
    frappe.enqueue(
        "myapp.tasks.do_export",
        queue="long",
        timeout=3600,
    )

---

3. No Error Handling in Background Job

# WRONG — one failure kills the entire job
def process_all():
    for item in frappe.get_all("Item"):
        external_api_call(item.name)  # If this throws, everything stops

# CORRECT — per-item error handling with commit/rollback
def process_all():
    for item in frappe.get_all("Item", fields=["name"]):
        try:
            external_api_call(item.name)
            frappe.db.commit()
        except Exception:
            frappe.db.rollback()
            frappe.log_error(frappe.get_traceback(), f"Process Error: {item.name}")

---

4. Wrong Queue for Task Duration

# WRONG — heavy report on short queue (5 min timeout)
frappe.enqueue("myapp.tasks.annual_report", queue="short")

# CORRECT — use long queue with appropriate timeout
frappe.enqueue("myapp.tasks.annual_report", queue="long", timeout=3600)

Rule: short < 30s, default < 5min, long < 25min, long+timeout for longer.

---

5. No Deduplication

# WRONG — clicking button 5 times = 5 duplicate jobs
def on_click():
    frappe.enqueue("myapp.tasks.process", doc=doc.name)

# CORRECT — deduplicate with job_id (v15+)
from frappe.utils.background_jobs import is_job_enqueued

def on_click():
    job_id = f"process::{doc.name}"
    if not is_job_enqueued(job_id):
        frappe.enqueue("myapp.tasks.process", job_id=job_id, doc=doc.name)
    else:
        frappe.msgprint("Already processing")

---

6. Using job_name Instead of job_id (v15+)

# WRONG — deprecated v14 pattern
frappe.enqueue("myapp.tasks.process", job_name="my_job")

# CORRECT — v15+ deduplication
frappe.enqueue("myapp.tasks.process", job_id="my_job")

---

7. Forgetting Administrator Context

# WRONG — document created with no explicit owner
def scheduled_task():
    doc = frappe.new_doc("ToDo")
    doc.description = "Auto-created task"
    doc.insert()
    # doc.owner = "Administrator" — probably not intended

# CORRECT — set explicit owner
def scheduled_task():
    doc = frappe.new_doc("ToDo")
    doc.description = "Auto-created task"
    doc.owner = "actual-user@example.com"
    doc.allocated_to = "actual-user@example.com"
    doc.insert(ignore_permissions=True)

---

8. No Commit in Batch Processing

# WRONG — all changes lost on any error (implicit rollback)
def process_batch():
    for item in large_list:
        frappe.db.set_value("Item", item, "status", "Processed")
    # No commit — if worker crashes, everything is lost

# CORRECT — commit per batch
def process_batch():
    for i, item in enumerate(large_list):
        frappe.db.set_value("Item", item, "status", "Processed")
        if i % 100 == 0:
            frappe.db.commit()
    frappe.db.commit()

---

9. Assuming Execution Order of Multiple Methods

# WRONG — assuming step_1 runs before step_2
scheduler_events = {
    "daily": [
        "myapp.tasks.step_1",  # Execution order is NOT guaranteed!
        "myapp.tasks.step_2",
    ]
}

# CORRECT — chain explicitly if order matters
scheduler_events = {
    "daily": ["myapp.tasks.run_steps"],
}

def run_steps():
    step_1()
    step_2()

---

10. Using enqueue_after_commit and Expecting Return Value

# WRONG — returns None when enqueue_after_commit=True
job = frappe.enqueue("myapp.tasks.process", enqueue_after_commit=True)
print(job.id)  # AttributeError: NoneType has no attribute 'id'

# CORRECT — do not use return value with enqueue_after_commit
frappe.enqueue("myapp.tasks.process", enqueue_after_commit=True)
# No return value available

---

Summary

frappe.enqueue API Reference

frappe.enqueue

Full Signature

frappe.enqueue(
    method,                      # Python function or module path (REQUIRED)
    queue="default",             # Queue: "short", "default", "long", or custom
    timeout=None,                # Custom timeout in seconds
    is_async=True,               # False = execute directly (not in worker)
    now=False,                   # True = execute via frappe.call() directly
    job_name=None,               # [DEPRECATED v15] Name for identification
    job_id=None,                 # [v15+] Unique ID for deduplication
    enqueue_after_commit=False,  # Wait for DB commit before enqueue
    at_front=False,              # Place job at front of queue
    on_success=None,             # Callback on success
    on_failure=None,             # Callback on failure
    **kwargs                     # Arguments for the method
)

Parameter Details

Return Value

# Returns RQ Job object (if enqueue_after_commit=False)
job = frappe.enqueue("myapp.tasks.process", param="value")
print(job.id)      # Job ID
print(job.status)  # Job status

# With enqueue_after_commit=True returns None
job = frappe.enqueue(..., enqueue_after_commit=True)
# job is None!

Examples

# Basic - module path
frappe.enqueue("myapp.tasks.process_data", customer="CUST-001")

# Basic - function object
def my_task(name, value):
    pass

frappe.enqueue(my_task, name="test", value=123)

# With custom timeout on long queue
frappe.enqueue(
    "myapp.tasks.heavy_report",
    queue="long",
    timeout=3600,  # 1 hour
    report_type="annual"
)

# Priority job (front of queue)
frappe.enqueue(
    "myapp.tasks.urgent_task",
    at_front=True,
    priority="high"
)

# After database commit
frappe.enqueue(
    "myapp.tasks.send_notification",
    enqueue_after_commit=True,
    user=frappe.session.user
)

---

frappe.enqueue_doc

Enqueue a controller method of a specific document.

Signature

frappe.enqueue_doc(
    doctype,           # DocType name (REQUIRED)
    name=None,         # Document name
    method=None,       # Controller method name
    queue="default",   # Queue name
    timeout=300,       # Timeout in seconds
    now=False,         # Execute directly
    **kwargs           # Extra arguments
)

Example

# Controller method
class SalesInvoice(Document):
    @frappe.whitelist()
    def send_notification(self, recipient, message):
        # Long-running operation
        pass

# Call it
frappe.enqueue_doc(
    "Sales Invoice",
    "SINV-00001",
    "send_notification",
    queue="long",
    timeout=600,
    recipient="user@example.com",
    message="Invoice ready"
)

---

Document.queue_action

Alternative to enqueue_doc from controller:

class SalesOrder(Document):
    def on_submit(self):
        # Queue heavy processing
        self.queue_action("send_emails", emails=email_list)
    
    def send_emails(self, emails):
        # Heavy operation
        for email in emails:
            send_mail(email)

---

Callbacks

Success Callback

def on_success_handler(job, connection, result, *args, **kwargs):
    """
    Args:
        job: RQ Job object
        connection: Redis connection
        result: Return value of job method
    """
    frappe.publish_realtime(
        "show_alert",
        {"message": f"Job {job.id} completed!"}
    )

Failure Callback

def on_failure_handler(job, connection, type, value, traceback):
    """
    Args:
        job: RQ Job object
        connection: Redis connection
        type: Exception type
        value: Exception value
        traceback: Traceback object
    """
    frappe.log_error(
        f"Job {job.id} failed: {value}",
        "Background Job Error"
    )

Usage

frappe.enqueue(
    "myapp.tasks.risky_operation",
    on_success=on_success_handler,
    on_failure=on_failure_handler,
    data=my_data
)

---

Job Deduplication

v15+ Pattern (Recommended)

from frappe.utils.background_jobs import is_job_enqueued

job_id = f"data_import::{self.name}"

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

v14 Pattern (Deprecated)

# ONLY for legacy v14 code
from frappe.core.page.background_jobs.background_jobs import get_info

enqueued_jobs = [d.get("job_name") for d in get_info()]
if self.name not in enqueued_jobs:
    frappe.enqueue(..., job_name=self.name)

Error Handling Reference

Complete reference for error handling in background jobs.

What Happens on Job Failure

1. Exception is logged:

• Scheduler Log DocType (visible in desk)
• logs/worker.error.log file

2. Lock file mechanism:

• Scheduler maintains lock file
• On crash, lock file remains
• LockTimeoutError after 10 minutes of inactive lock

3. Job status becomes "failed" in RQ

Basic Error Handling Pattern

def process_records(records):
    """Process records with error handling per item."""
    success_count = 0
    error_count = 0
    
    for record in records:
        try:
            process_single(record)
            frappe.db.commit()  # Commit per success
            success_count += 1
        except Exception:
            frappe.db.rollback()  # Rollback on error
            frappe.log_error(
                frappe.get_traceback(),
                f"Process Error for {record}"
            )
            error_count += 1
    
    return {"success": success_count, "errors": error_count}

frappe.log_error

Basic Usage

frappe.log_error(
    message="Error processing record",
    title="Background Job Error"
)

With Traceback

try:
    risky_operation()
except Exception:
    frappe.log_error(
        message=frappe.get_traceback(),
        title="Process Failed"
    )

With Context

frappe.log_error(
    message=f"Failed for {doc.name}: {frappe.get_traceback()}",
    title=f"Process Error: {doc.doctype}"
)

On-Failure Callbacks

def on_failure_handler(job, connection, type, value, traceback):
    """Callback on job failure."""
    frappe.log_error(
        message=f"Job {job.id} failed with {type.__name__}: {value}",
        title="Background Job Failed"
    )
    
    # Optional: send notification
    frappe.sendmail(
        recipients=["admin@example.com"],
        subject=f"Job Failed: {job.id}",
        message=f"Error: {value}"
    )

frappe.enqueue(
    'myapp.tasks.risky_operation',
    on_failure=on_failure_handler
)

On-Success Callbacks

def on_success_handler(job, connection, result, *args, **kwargs):
    """Callback on job success."""
    frappe.publish_realtime(
        'show_alert',
        {'message': 'Processing complete!', 'indicator': 'green'},
        user=frappe.session.user
    )

frappe.enqueue(
    'myapp.tasks.process',
    on_success=on_success_handler
)

Retry Pattern (Manual)

def task_with_retry(data, retry_count=0, max_retries=3):
    """Task with exponential backoff retry."""
    try:
        external_api_call(data)
    except Exception as e:
        if retry_count < max_retries:
            # Exponential backoff: 60s, 120s, 240s
            delay = 60 * (2 ** retry_count)
            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} scheduled",
                f"Task Retry: {data}"
            )
        else:
            frappe.log_error(
                frappe.get_traceback(),
                f"Task Failed after {max_retries} retries: {data}"
            )
            raise

Batch Processing with Graceful Degradation

def process_batch(items, notify_user=None):
    """Process batch with individual error handling."""
    results = {"success": [], "failed": []}
    
    for item in items:
        try:
            result = process_item(item)
            results["success"].append({"item": item, "result": result})
            frappe.db.commit()
        except frappe.ValidationError as e:
            # Known validation error - log and continue
            results["failed"].append({"item": item, "error": str(e)})
            frappe.db.rollback()
        except Exception:
            # Unknown error - log with traceback
            results["failed"].append({
                "item": item, 
                "error": frappe.get_traceback()
            })
            frappe.log_error(
                frappe.get_traceback(),
                f"Batch Item Failed: {item}"
            )
            frappe.db.rollback()
    
    # Report results
    if notify_user:
        frappe.publish_realtime(
            'batch_complete',
            results,
            user=notify_user
        )
    
    return results

Email Notifications for Failed Jobs

In sites/common_site_config.json:

{
    "celery_error_emails": {
        "ADMINS": [
            ["Admin Name", "admin@example.com"]
        ],
        "SERVER_EMAIL": "errors@example.com"
    }
}

Note: Uses local mail server on port 25.

Viewing Error Logs

Via Desk

Search for "Error Log" in the searchbar.

Via CLI

# Recent errors
bench --site mysite.local execute frappe.get_list \
    --kwargs '{"doctype": "Error Log", "limit": 10}'

# Worker log
tail -f logs/worker.error.log

Common Errors and Solutions

LockTimeoutError

# Cause: Scheduler crash with lock file
# Solution:
frappe.utils.scheduler.enable_scheduler()

TimeLimitExceeded

# Cause: Job takes longer than timeout
# Solution: Use long queue or increase timeout
frappe.enqueue(..., queue='long', timeout=3600)

MemoryError

# Cause: Too much data in memory
# Solution: Process in chunks
def process_large_dataset():
    offset = 0
    batch_size = 1000
    while True:
        items = frappe.get_all("Item", limit=batch_size, start=offset)
        if not items:
            break
        process_items(items)
        frappe.db.commit()
        offset += batch_size

Best Practices

1. ALWAYS try/except with commit/rollback per record 2. LOG errors with context (document name, parameters) 3. USE on_failure callback for critical tasks 4. IMPLEMENT retry logic for external API calls 5. NOTIFY users on completion (success or failure) 6. PROCESS in chunks for large datasets

Scheduler Event Types — Detailed Reference

Standard Events

These run on the default queue with a 300-second timeout.

Tick Interval

The all event fires on every tick. Other events are tracked by last execution time.

---

Long Queue Events

These run on the long queue with a 1500-second (25 min) timeout.

ALWAYS use _long variants when the task takes more than 5 minutes.

---

Cron Events

Custom schedules using croniter-compatible strings.

scheduler_events = {
    "cron": {
        "*/5 * * * *": ["myapp.tasks.every_5_minutes"],
        "0 9 * * 1-5": ["myapp.tasks.weekday_9am"],
        "0 8 * * 1": ["myapp.tasks.monday_morning"],
        "0 0 1 * *": ["myapp.tasks.first_of_month"],
        "15 18 * * *": ["myapp.tasks.evening_summary"],
        "0 9-17 * * 1-5": ["myapp.tasks.business_hours"],
    }
}

Cron Syntax Reference

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

---

Runtime-Configurable Events

For events that need to be adjusted without code deployment, use Scheduled Job Type DocType:

# Create via code (or via desk UI)
job = frappe.new_doc("Scheduled Job Type")
job.method = "myapp.tasks.configurable_task"
job.frequency = "Cron"
job.cron_format = "0/5 * * * *"
job.save()

This is useful when administrators need to adjust scheduling without developer intervention.

---

Multiple Methods Per Event

scheduler_events = {
    "daily": [
        "myapp.tasks.cleanup_logs",
        "myapp.tasks.send_daily_report",
        "myapp.tasks.sync_external_data",
    ]
}

IMPORTANT: Execution order of multiple methods within the same event is NOT guaranteed. If order matters, use a single function that calls them sequentially.

---

Event Function Requirements

Every scheduler event function:

• Takes NO arguments
• Runs as Administrator user
• Has access to frappe.db, frappe.get_all(), etc.
• Should handle its own errors (unhandled exceptions are logged to Error Log)

def my_scheduler_task():
    """CORRECT — no parameters."""
    records = frappe.get_all("MyDocType")
    for record in records:
        process(record)

---

Scheduled Job Log

Every execution of a scheduler event creates a Scheduled Job Log record:

Access via desk: Search > Scheduled Job Log

Scheduler & Background Jobs — Complete Examples

Example 1: Daily Email Summary

hooks.py

scheduler_events = {
    "daily": ["myapp.tasks.send_daily_summary"],
}

myapp/tasks.py

import frappe

def send_daily_summary():
    """Send daily summary email to all active managers."""
    managers = frappe.get_all(
        "User",
        filters={"enabled": 1, "role_profile_name": "Manager"},
        fields=["name", "email"],
    )

    for manager in managers:
        try:
            summary = build_summary(manager.name)
            frappe.sendmail(
                recipients=[manager.email],
                subject=f"Daily Summary — {frappe.utils.today()}",
                message=summary,
            )
            frappe.db.commit()
        except Exception:
            frappe.db.rollback()
            frappe.log_error(
                frappe.get_traceback(),
                f"Daily Summary Error: {manager.name}",
            )

def build_summary(user):
    open_tasks = frappe.db.count("ToDo", {"allocated_to": user, "status": "Open"})
    return f"You have {open_tasks} open tasks."

---

Example 2: Cron-Based Data Sync

hooks.py

scheduler_events = {
    "cron": {
        "*/15 * * * *": ["myapp.integrations.sync.run_sync"],
    }
}

myapp/integrations/sync.py

import frappe
from frappe.utils.background_jobs import is_job_enqueued

def run_sync():
    """Enqueue sync job with deduplication (v15+)."""
    job_id = "external_api_sync"

    if is_job_enqueued(job_id):
        return  # Already running

    frappe.enqueue(
        "myapp.integrations.sync.do_sync",
        job_id=job_id,
        queue="long",
        timeout=1800,
    )

def do_sync():
    """Actual sync logic — runs in background worker."""
    records = fetch_from_external_api()

    for record in records:
        try:
            update_or_create(record)
            frappe.db.commit()
        except Exception:
            frappe.db.rollback()
            frappe.log_error(
                frappe.get_traceback(),
                f"Sync Error: {record.get('id')}",
            )

---

Example 3: frappe.enqueue_doc with Progress

Controller

class DataImport(Document):
    @frappe.whitelist()
    def start_import(self):
        frappe.enqueue_doc(
            self.doctype,
            self.name,
            "run_import",
            queue="long",
            timeout=3600,
        )
        frappe.msgprint("Import started in background")

    def run_import(self):
        rows = get_import_rows(self.name)
        total = len(rows)

        for i, row in enumerate(rows):
            try:
                process_row(row)
                frappe.db.commit()
            except Exception:
                frappe.db.rollback()
                frappe.log_error(frappe.get_traceback(), f"Import Row Error: {i}")

            frappe.publish_realtime(
                "import_progress",
                {"progress": (i + 1) / total * 100, "current": i + 1, "total": total},
                doctype=self.doctype,
                docname=self.name,
            )

---

Example 4: Retry with Exponential Backoff

def call_external_api(data, retry_count=0, max_retries=3):
    """Call external API with retry on failure."""
    try:
        response = make_api_request(data)
        process_response(response)
    except Exception as e:
        if retry_count < max_retries:
            frappe.enqueue(
                "myapp.tasks.call_external_api",
                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}: {str(e)}",
                "API Retry",
            )
        else:
            frappe.log_error(
                frappe.get_traceback(),
                f"API Failed after {max_retries} retries",
            )
            raise

---

Example 5: Heavy Scheduler Event Delegating to Enqueue

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

# myapp/tasks.py
def daily_cleanup():
    """Scheduler event that delegates to enqueue for heavy work."""
    sites = get_sites_needing_cleanup()

    for site in sites:
        frappe.enqueue(
            "myapp.tasks.cleanup_site",
            queue="long",
            timeout=1800,
            site_name=site,
        )

def cleanup_site(site_name):
    """Actual cleanup — runs as separate background job."""
    batch_size = 500
    offset = 0

    while True:
        old_logs = frappe.get_all(
            "Error Log",
            filters={"creation": ["<", frappe.utils.add_days(None, -30)]},
            fields=["name"],
            limit_page_length=batch_size,
            limit_start=offset,
        )
        if not old_logs:
            break

        for log in old_logs:
            frappe.delete_doc("Error Log", log.name, force=True)

        frappe.db.commit()
        offset += batch_size

Scheduler Hooks Reference

scheduler_events Hook

Location: {app}/{app}/hooks.py

The scheduler_events dictionary maps event frequencies to lists of Python module paths that execute at the specified intervals.

Complete Syntax

scheduler_events = {
    # Standard events — run on default queue
    "all": [
        "myapp.tasks.every_tick",
    ],
    "hourly": [
        "myapp.tasks.hourly_sync",
        "myapp.tasks.hourly_cleanup",
    ],
    "daily": [
        "myapp.tasks.daily_report",
    ],
    "weekly": [
        "myapp.tasks.weekly_summary",
    ],
    "monthly": [
        "myapp.tasks.monthly_billing",
    ],

    # Long queue events — run on long queue (higher timeout)
    "hourly_long": [
        "myapp.tasks.hourly_heavy_sync",
    ],
    "daily_long": [
        "myapp.tasks.daily_large_export",
    ],
    "weekly_long": [
        "myapp.tasks.weekly_archive",
    ],
    "monthly_long": [
        "myapp.tasks.monthly_full_report",
    ],

    # Cron events — croniter-compatible schedule strings
    "cron": {
        "*/5 * * * *": [
            "myapp.tasks.every_5_minutes",
        ],
        "0 9 * * 1-5": [
            "myapp.tasks.weekday_morning",
        ],
        "0 0 1 * *": [
            "myapp.tasks.first_of_month",
        ],
    },
}

How Scheduler Events Are Registered

1. Developer adds/modifies scheduler_events in hooks.py 2. Developer runs bench migrate 3. Frappe reads hooks from all installed apps 4. Creates/updates Scheduled Job Type records in the database 5. The scheduler process checks these records at each tick

Activation Requirement

After EVERY change to scheduler_events:

bench migrate

Without this step, changes to scheduler_events are NOT active.

Multiple Apps

When multiple apps define scheduler events for the same frequency, ALL registered methods execute. There is no override — events accumulate across apps.

# App A hooks.py
scheduler_events = {"daily": ["app_a.tasks.daily"]}

# App B hooks.py
scheduler_events = {"daily": ["app_b.tasks.daily"]}

# Result: BOTH app_a.tasks.daily AND app_b.tasks.daily run daily

Event Function Signature

All scheduler event functions take NO arguments:

def my_scheduled_task():
    """Scheduler event — no arguments."""
    # Use frappe API to access data
    records = frappe.get_all("DocType", filters={...})

Enabling/Disabling Scheduler

# Disable scheduler for a site
bench --site mysite disable-scheduler

# Enable scheduler for a site
bench --site mysite enable-scheduler

# Check scheduler status
bench doctor

Site Config Options

In common_site_config.json:

{
    "scheduler_tick_interval": 60,
    "pause_scheduler": 0
}

Scheduler & Background Jobs — Method Reference

frappe.enqueue()

frappe.enqueue(
    method,                      # str or callable — REQUIRED
    queue="default",             # "short" | "default" | "long" | custom
    timeout=None,                # int — override queue timeout (seconds)
    is_async=True,               # bool — False = synchronous (skip worker)
    now=False,                   # bool — True = run via frappe.call() directly
    job_id=None,                 # str — [v15+] unique ID for deduplication
    enqueue_after_commit=False,  # bool — wait for DB commit before enqueue
    at_front=False,              # bool — place at front of queue
    on_success=None,             # callable — success callback
    on_failure=None,             # callable — failure callback
    **kwargs                     # passed to method as arguments
)
# Returns: RQ Job object (or None if enqueue_after_commit=True)

Method Parameter Formats

# Module path string (recommended for cross-module calls)
frappe.enqueue("myapp.tasks.process_data", customer="CUST-001")

# Function object (for same-module calls)
def my_task(name, value):
    pass

frappe.enqueue(my_task, name="test", value=123)

Return Value

# Normal enqueue — returns RQ Job object
job = frappe.enqueue("myapp.tasks.process", param="value")
print(job.id)       # Job ID string
print(job.status)   # "queued", "started", "finished", "failed"

# With enqueue_after_commit — returns None
job = frappe.enqueue("myapp.tasks.process", enqueue_after_commit=True)
# job is None — NEVER access attributes on it

---

frappe.enqueue_doc()

frappe.enqueue_doc(
    doctype,           # str — DocType name (REQUIRED)
    name=None,         # str — document name
    method=None,       # str — controller method name
    queue="default",   # str — queue name
    timeout=300,       # int — timeout in seconds
    now=False,         # bool — run directly
    **kwargs           # passed to controller method
)

The controller method MUST be decorated with @frappe.whitelist().

---

Document.queue_action()

# From within a controller
self.queue_action(
    method_name,       # str — name of method on this controller
    **kwargs           # passed to method
)

---

frappe.utils.background_jobs.is_job_enqueued()

from frappe.utils.background_jobs import is_job_enqueued

result = is_job_enqueued(job_id)  # str — returns bool

[v15+] Checks if a job with the given job_id is currently queued or running.

---

Callback Signatures

on_success

def on_success_handler(job, connection, result, *args, **kwargs):
    """
    job: RQ Job object
    connection: Redis connection
    result: return value of the executed method
    """
    pass

on_failure

def on_failure_handler(job, connection, type, value, traceback):
    """
    job: RQ Job object
    connection: Redis connection
    type: exception class
    value: exception instance
    traceback: traceback object
    """
    pass

---

frappe.log_error()

frappe.log_error(
    message="Error details or traceback",
    title="Short title for Error Log list"
)

Creates an Error Log document visible in the desk.

---

frappe.publish_realtime()

frappe.publish_realtime(
    event,             # str — event name
    message=None,      # dict — data payload
    user=None,         # str — target user (or broadcast if None)
    doctype=None,      # str — target doctype (for doc-specific updates)
    docname=None,      # str — target document name
    after_commit=False # bool — wait for DB commit
)

---

Scheduler Control

# Enable scheduler
frappe.utils.scheduler.enable_scheduler()

# Disable scheduler
frappe.utils.scheduler.disable_scheduler()

# Check if enabled
is_enabled = frappe.utils.scheduler.is_scheduler_inactive()

---

Queue Utilities

from frappe.utils.background_jobs import get_queue

# Get queue object
queue = get_queue("default")
print(f"Jobs in queue: {len(queue)}")

# Get job status
from rq.job import Job
job = Job.fetch(job_id, connection=frappe.cache())
print(job.get_status())  # "queued", "started", "finished", "failed"

Monitoring Reference

Complete reference for background job monitoring.

RQ Worker DocType (Virtual)

Shows all background workers.

Access: Search > RQ Worker

Fields

RQ Job DocType (Virtual)

Shows all background jobs.

Access: Search > RQ Job

Filters

• Queue: short, default, long, or custom
• Status: queued, started, finished, failed

Fields

Job Statuses

Scheduled Job Log

DocType that tracks scheduler job executions.

Access: Search > Scheduled Job Log

Fields

bench doctor

CLI command for scheduler diagnostics.

bench doctor

Example Output

Scheduler Status for site1.local
   Scheduler is: enabled
   Workers are: running
   Pending tasks: 3

Scheduler Status for site2.local
   Scheduler is: enabled
   Workers are: running
   Pending tasks: 0

Check specific site

bench --site mysite.local doctor

Monitor Feature

Enable

In sites/{site}/site_config.json:

{
    "monitor": 1
}

Log Location

logs/monitor.json.log

Log Format

{
    "duration": 1364,
    "job": {
        "method": "frappe.ping",
        "scheduled": false,
        "wait": 90204
    },
    "site": "frappe.local",
    "timestamp": "2020-03-05 09:37:40.124682",
    "transaction_type": "job",
    "uuid": "8225ab76-8bee-462c-b9fc-a556406b1ee7"
}

Fields

Stuck Worker Debug

If a worker is stuck:

# Find worker PID
ps aux | grep "bench worker"

# Send SIGUSR1 for stack trace
kill -SIGUSR1 <WORKER_PID>

Output goes to logs/worker.error.log.

Log Files

Log Tailing

# Follow worker errors live
tail -f logs/worker.error.log

# Scheduler activity
tail -f logs/scheduler.log

Programmatic Monitoring

Queue Info

from frappe.utils.background_jobs import get_queue_info

info = frappe.call('frappe.utils.background_jobs.get_queue_info')
# Returns queue statistics

Job Status Check

job = frappe.enqueue('myapp.tasks.process')
status = job.get_status()  # 'queued', 'started', 'finished', 'failed'

Pending Jobs Count

from frappe.utils.background_jobs import get_jobs

jobs = get_jobs(site='mysite.local', queue='default', status='queued')
pending_count = len(jobs)

Realtime Progress Updates

def long_running_task(items, user):
    """Task with progress updates."""
    total = len(items)
    
    for i, item in enumerate(items):
        process_item(item)
        
        # Update progress
        frappe.publish_realtime(
            'task_progress',
            {
                'progress': (i + 1) / total * 100,
                'current': i + 1,
                'total': total
            },
            user=user
        )
    
    frappe.publish_realtime(
        'task_complete',
        {'message': f'Processed {total} items'},
        user=user
    )

Alerting Setup

Via Error Log Monitoring

# Scheduled task to check error count
def check_error_rate():
    hour_ago = frappe.utils.add_to_date(
        frappe.utils.now_datetime(),
        hours=-1
    )
    
    errors = frappe.db.count("Error Log", {
        "creation": [">=", hour_ago]
    })
    
    if errors > 100:
        frappe.sendmail(
            recipients=["admin@example.com"],
            subject="High Error Rate Alert",
            message=f"{errors} errors in the last hour"
        )

Best Practices

1. Monitor queue depths via RQ Job doctype 2. Set up alerts for high error rates 3. Use realtime updates for long tasks 4. Check bench doctor after deployments 5. Review Error Log regularly 6. Analyze monitor.json.log for performance insights

Scheduler & Background Jobs — Patterns

Pattern 1: Scheduler Event Delegates to Enqueue

NEVER do heavy work directly in a scheduler event. ALWAYS delegate to frappe.enqueue().

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

# myapp/tasks.py
def daily_export():
    """Lightweight scheduler event — just enqueues the heavy work."""
    frappe.enqueue(
        "myapp.tasks.do_export",
        queue="long",
        timeout=3600,
    )

def do_export():
    """Heavy work runs in background worker, not in scheduler."""
    # Export logic here
    pass

---

Pattern 2: Batch Processing with Progress

def process_large_dataset(user=None):
    """Process records in batches with progress updates."""
    batch_size = 500
    total = frappe.db.count("MyDocType", {"status": "Pending"})
    processed = 0

    while processed < total:
        records = frappe.get_all(
            "MyDocType",
            filters={"status": "Pending"},
            fields=["name"],
            limit_page_length=batch_size,
        )
        if not records:
            break

        for record in records:
            try:
                do_work(record.name)
                frappe.db.commit()
                processed += 1
            except Exception:
                frappe.db.rollback()
                frappe.log_error(frappe.get_traceback(), f"Error: {record.name}")
                processed += 1

        if user:
            frappe.publish_realtime(
                "processing_progress",
                {"progress": processed / total * 100},
                user=user,
            )

---

Pattern 3: Deduplication Guard

from frappe.utils.background_jobs import is_job_enqueued

def safe_enqueue(method, job_id, **kwargs):
    """Enqueue only if not already running (v15+)."""
    if not is_job_enqueued(job_id):
        return frappe.enqueue(method, job_id=job_id, **kwargs)
    return None

---

Pattern 4: Callbacks for User Notification

def start_heavy_task(doc_name, user):
    """Start task with success/failure notifications."""

    def on_success(job, connection, result, *args, **kwargs):
        frappe.publish_realtime(
            "show_alert",
            {"message": f"Task completed for {doc_name}", "indicator": "green"},
            user=user,
        )

    def on_failure(job, connection, type, value, traceback):
        frappe.publish_realtime(
            "show_alert",
            {"message": f"Task failed for {doc_name}", "indicator": "red"},
            user=user,
        )
        frappe.log_error(f"Job {job.id} failed: {value}", "Task Failed")

    frappe.enqueue(
        "myapp.tasks.heavy_task",
        queue="long",
        on_success=on_success,
        on_failure=on_failure,
        doc_name=doc_name,
    )

---

Pattern 5: enqueue_after_commit

Use when the background job needs data that is being saved in the current transaction:

class SalesInvoice(Document):
    def on_submit(self):
        # Invoice data is not yet committed to DB
        frappe.enqueue(
            "myapp.tasks.send_invoice_notification",
            enqueue_after_commit=True,  # Waits for commit
            invoice=self.name,
        )

Without enqueue_after_commit, the worker might start before the document is committed, causing a "not found" error.

---

Pattern 6: Conditional Scheduling

def configurable_sync():
    """Only run if integration is enabled in settings."""
    settings = frappe.get_single("My App Settings")

    if not settings.enable_sync:
        return  # Skip silently

    frappe.enqueue(
        "myapp.integrations.sync.do_sync",
        queue="long",
        timeout=1800,
    )

---

Pattern 7: Multi-Site Awareness

def scheduled_task():
    """Task that respects multi-site setup."""
    site = frappe.local.site

    # Site-specific logic
    config = frappe.get_site_config()
    if config.get("disable_scheduled_exports"):
        return

    # Proceed with task
    pass

Queue Types & Configuration

Default Queues

Queue Selection Guidelines

# SHORT queue - quick operations
frappe.enqueue(
    "myapp.tasks.update_status",
    queue="short",
    doc_name=doc.name
)

# DEFAULT queue - standard tasks
frappe.enqueue(
    "myapp.tasks.send_email",
    # queue="default" is implicit
    recipient=email
)

# LONG queue - heavy processing
frappe.enqueue(
    "myapp.tasks.generate_report",
    queue="long",
    timeout=3600,  # 1 hour
    report_type="annual"
)

Custom Queue Timeout

# Override default timeout
frappe.enqueue(
    "myapp.tasks.medium_task",
    queue="default",
    timeout=900,  # 15 minutes instead of 5
    data=large_data
)

Custom Queues Configuration

In common_site_config.json:

{
    "workers": {
        "priority": {
            "timeout": 60,
            "background_workers": 2
        },
        "reports": {
            "timeout": 7200,
            "background_workers": 1
        },
        "imports": {
            "timeout": 5000,
            "background_workers": 4
        }
    }
}

Using Custom Queue

frappe.enqueue(
    "myapp.tasks.generate_large_report",
    queue="reports",
    report_id=report.name
)

Worker Configuration

Default Procfile

worker_short: bench worker --queue short --quiet
worker_default: bench worker --queue default --quiet
worker_long: bench worker --queue long --quiet

Multi-Queue Worker

# One worker consumes from multiple queues
bench worker --queue short,default
bench worker --queue long

Burst Mode

Temporary worker that stops when queue is empty:

bench worker --queue short --burst

Useful for:

• One-time batch processing
• Development/testing
• Temporary extra capacity

Queue Priority

Jobs are processed in FIFO order within each queue.

# Place at front of queue (priority)
frappe.enqueue(
    "myapp.tasks.urgent_task",
    at_front=True,
    task_id=task.name
)

Queue Monitoring

Bench Commands

# Scheduler status
bench doctor

# View queue status
bench --site mysite show-pending-jobs

# Specific queue
bench --site mysite show-pending-jobs --queue long

Via DocTypes

• RQ Worker: Worker status (busy/idle)
• RQ Job: Job status per queue

Via Code

from frappe.utils.background_jobs import get_queue

# Queue stats
queue = get_queue("default")
print(f"Jobs in queue: {len(queue)}")

# Job status
from rq.job import Job
job = Job.fetch(job_id, connection=frappe.cache())
print(job.get_status())

Queue Best Practices

Choosing the Right Queue

Avoid Queue Blocking

# WRONG - blocks short queue
frappe.enqueue(
    "myapp.tasks.heavy_task",
    queue="short"  # Timeout after 5 min!
)

# RIGHT - use long queue
frappe.enqueue(
    "myapp.tasks.heavy_task",
    queue="long",
    timeout=3600
)

Worker Scaling

# More workers for specific queue
# In supervisor config or Procfile:
worker_long_1: bench worker --queue long --quiet
worker_long_2: bench worker --queue long --quiet
worker_long_3: bench worker --queue long --quiet

Scheduler Events Reference

Event Types Overview

# hooks.py - Complete syntax
scheduler_events = {
    # Standard events (default queue)
    "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"],
    
    # 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 (custom scheduling)
    "cron": {
        "*/15 * * * *": ["myapp.tasks.every_15_minutes"],
        "0 9 * * 1-5": ["myapp.tasks.weekday_9am"],
        "0 0 1 * *": ["myapp.tasks.first_of_month"]
    }
}

Cron Syntax

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

Cron Symbols

Common Cron Patterns

scheduler_events = {
    "cron": {
        # Every 5 minutes
        "*/5 * * * *": ["myapp.tasks.frequent_check"],
        
        # Every weekday at 9:00
        "0 9 * * 1-5": ["myapp.tasks.workday_morning"],
        
        # Every Monday at 8:00
        "0 8 * * 1": ["myapp.tasks.monday_report"],
        
        # First day of month at midnight
        "0 0 1 * *": ["myapp.tasks.monthly_cleanup"],
        
        # Every day at 18:15
        "15 18 * * *": ["myapp.tasks.evening_summary"],
        
        # Every hour from 9-17 on weekdays
        "0 9-17 * * 1-5": ["myapp.tasks.business_hours"],
        
        # Special string (annual)
        "annual": ["myapp.tasks.yearly_archive"]
    }
}

Scheduler Tick Interval

Custom Tick Interval

In common_site_config.json:

{
    "scheduler_tick_interval": 120
}

Multiple Methods Per Event

scheduler_events = {
    "daily": [
        "myapp.tasks.cleanup_logs",
        "myapp.tasks.send_daily_report",
        "myapp.tasks.sync_external_data"
    ]
}

Note: Execution order is NOT guaranteed!

CRITICAL: bench migrate

After EVERY change to scheduler_events:

bench migrate

Without bench migrate changes are NOT applied!

Runtime Configurable Events

For events that need to be adjusted without code deploy:

# Create Scheduler Event record
sch_eve = frappe.new_doc("Scheduler Event")
sch_eve.scheduled_against = "Payment Reconciliation"
sch_eve.save()

# Create Scheduled Job Type
job = frappe.new_doc("Scheduled Job Type")
job.frequency = "Cron"
job.scheduler_event = sch_eve.name
job.cron_format = "0/5 * * * *"  # Every 5 minutes
job.save()

Event Debugging

# Check scheduler status
bench doctor

# View scheduled job log
bench --site mysite execute frappe.utils.scheduler.get_enabled_scheduler_events

Scheduler & Background Jobs — Syntax Quick Reference

hooks.py — scheduler_events

scheduler_events = {
    "all": ["dotted.module.path"],
    "hourly": ["dotted.module.path"],
    "daily": ["dotted.module.path"],
    "weekly": ["dotted.module.path"],
    "monthly": ["dotted.module.path"],
    "hourly_long": ["dotted.module.path"],
    "daily_long": ["dotted.module.path"],
    "weekly_long": ["dotted.module.path"],
    "monthly_long": ["dotted.module.path"],
    "cron": {
        "cron_expression": ["dotted.module.path"],
    },
}

---

frappe.enqueue() — Minimal

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

frappe.enqueue() — Full

frappe.enqueue(
    "myapp.tasks.process",
    queue="long",
    timeout=3600,
    job_id="unique-id",
    enqueue_after_commit=True,
    at_front=False,
    on_success=success_fn,
    on_failure=failure_fn,
    param1="value1",
    param2="value2",
)

---

frappe.enqueue_doc()

frappe.enqueue_doc(
    "Sales Invoice",
    "SINV-00001",
    "controller_method",
    queue="long",
    timeout=600,
    kwarg1="value",
)

---

self.queue_action()

# Inside a Document controller
self.queue_action("method_name", param="value")

---

Deduplication [v15+]

from frappe.utils.background_jobs import is_job_enqueued

if not is_job_enqueued("unique-job-id"):
    frappe.enqueue("myapp.tasks.process", job_id="unique-job-id")

---

Scheduler Event Function

def my_task():
    """No arguments. Runs as Administrator."""
    pass

---

Error Handling Template

def batch_job(items):
    for item in items:
        try:
            process(item)
            frappe.db.commit()
        except Exception:
            frappe.db.rollback()
            frappe.log_error(frappe.get_traceback(), f"Error: {item}")

---

Callback Signatures

def on_success(job, connection, result, *args, **kwargs):
    pass

def on_failure(job, connection, type, value, traceback):
    pass

---

Progress Reporting

frappe.publish_realtime("event_name", {"progress": 50}, user="user@example.com")

---

CLI Commands

bench migrate                          # Activate scheduler_events changes
bench doctor                           # Scheduler status
bench --site mysite enable-scheduler   # Enable scheduler
bench --site mysite disable-scheduler  # Disable scheduler
bench worker --queue short             # Start worker for queue
bench worker --queue short --burst     # Temporary burst worker