Frappe Impl Scheduler

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

---

Which Scheduler Event Type?

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

---

Which Queue for frappe.enqueue?

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

Rule: ALWAYS wrap per-record processing in try-except. NEVER let one failure stop the entire batch.

---

Long-Running Job Patterns

Self-Chaining Pattern (>25 min tasks)

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
        )

Rule: ALWAYS split tasks >25 min into self-chaining batches.

---

Common Implementation Patterns

Email Digest (weekly summary)

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

See references/examples.md Example 4 for complete implementation.

Data Cleanup (daily maintenance)

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

See references/examples.md Example 1 for batch deletion pattern.

Report Generation (user-triggered)

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

See references/workflows.md Workflow 6 for progress reporting.

---

Critical Rules

1. Scheduler tasks receive NO arguments - Use settings or hardcoded values 2. ALWAYS `bench migrate` after hooks.py changes - Required to register events 3. Jobs run as Administrator - ALWAYS commit explicitly 4. Commit in batches - NEVER per-record (every 100-500 records) 5. ALWAYS use `job_id` for user-triggered jobs - Prevents duplicates 6. Use `enqueue_after_commit=True` from document events - Ensures data exists 7. Scheduler events should be thin - Enqueue heavy work to background

Version Differences

---

Reference Files

See Also

• frappe-syntax-scheduler - Exact syntax reference for hooks and enqueue
• frappe-errors-serverscripts - Error handling patterns
• frappe-impl-hooks - Hook configuration patterns
• frappe-ops-bench - Bench commands for scheduler management
• frappe-ops-performance - Performance tuning for background jobs
• frappe-testing-unit - Testing scheduled task logic

Scheduler & Background Jobs - Anti-Patterns

Common mistakes and how to avoid them.

---

Anti-Pattern 1: Passing Arguments to Scheduler Tasks

❌ Wrong

# hooks.py
scheduler_events = {
    "daily": ["myapp.tasks.cleanup(days=30)"]  # Won't work!
}

# Or trying to pass in the function definition
def cleanup(days=30):
    ...

Error: Arguments are ignored; task may fail or use wrong values

✅ Correct

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

# myapp/tasks.py
def cleanup():
    """Scheduler tasks receive NO arguments."""
    days = 30  # Hardcode or get from settings
    # Or:
    days = frappe.db.get_single_value("Cleanup Settings", "retention_days") or 30

Rule

Scheduler tasks CANNOT receive arguments. Use hardcoded values, settings, or database lookups.

---

Anti-Pattern 2: Forgetting to Migrate After hooks.py Changes

❌ Wrong

# Add new scheduler event to hooks.py
scheduler_events = {
    "daily": ["myapp.tasks.new_task"]
}

# Restart bench and expect it to work
# bench restart

Error: Task never runs because it's not registered

✅ Correct

# ALWAYS migrate after hooks.py changes
bench --site sitename migrate

# Then optionally restart
bench restart

Rule

ALWAYS run `bench migrate` after ANY hooks.py change. This registers scheduler events in the database.

---

Anti-Pattern 3: Using Wrong Queue for Task Duration

❌ Wrong

# 10-minute task in default queue (5 min timeout)
frappe.enqueue(
    "myapp.tasks.heavy_import",
    queue="default"  # Will timeout!
)

# Or quick task in long queue (wastes resources)
frappe.enqueue(
    "myapp.tasks.send_notification",
    queue="long"  # Overkill for 1-second task
)

Error: Task times out or wastes worker resources

✅ Correct

# Match queue to task duration
# < 30 sec: short
frappe.enqueue("myapp.tasks.quick_task", queue="short")

# < 5 min: default
frappe.enqueue("myapp.tasks.medium_task", queue="default")

# 5-25 min: long
frappe.enqueue("myapp.tasks.heavy_task", queue="long")

# > 25 min: split into chunks!

Rule

---

Anti-Pattern 4: No Deduplication for User-Triggered Jobs

❌ Wrong

@frappe.whitelist()
def start_export():
    # User clicks 5 times = 5 identical jobs!
    frappe.enqueue("myapp.tasks.export_data")

Error: Multiple identical jobs queue up, waste resources, may cause data issues

✅ Correct

from frappe.utils.background_jobs import is_job_enqueued

@frappe.whitelist()
def start_export():
    job_id = f"export::{frappe.session.user}"
    
    if is_job_enqueued(job_id):
        return {"message": "Export already in progress"}
    
    frappe.enqueue(
        "myapp.tasks.export_data",
        job_id=job_id,
        user=frappe.session.user
    )
    return {"message": "Export started"}

Rule

Always deduplicate user-triggered background jobs using job_id and is_job_enqueued().

---

Anti-Pattern 5: Committing After Every Record

❌ Wrong

def process_records():
    records = get_all_records()  # 10,000 records
    
    for record in records:
        process(record)
        frappe.db.commit()  # 10,000 commits = SLOW!

Error: Extremely slow due to database overhead per commit

✅ Correct

def process_records():
    records = get_all_records()
    
    for i, record in enumerate(records):
        process(record)
        
        # Commit every 100 records
        if i % 100 == 0:
            frappe.db.commit()
    
    # Final commit
    frappe.db.commit()

Rule

Commit in batches, not per record. Every 100-500 records is typically optimal.

---

Anti-Pattern 6: No Error Handling in Scheduler Tasks

❌ Wrong

def sync_all_orders():
    orders = get_pending_orders()
    
    for order in orders:
        sync_to_external(order)  # If one fails, entire task fails!

Error: One failure stops all processing; partial data may be inconsistent

✅ Correct

def sync_all_orders():
    orders = get_pending_orders()
    success = 0
    errors = 0
    
    for order in orders:
        try:
            sync_to_external(order)
            success += 1
            frappe.db.commit()  # Commit successful ones
        except Exception as e:
            errors += 1
            frappe.db.rollback()  # Rollback failed one
            frappe.log_error(
                f"Sync failed for {order}: {e}",
                "Order Sync Error"
            )
    
    frappe.logger("sync").info(f"Sync: {success} success, {errors} errors")

Rule

Always wrap record processing in try-except. Log errors and continue with remaining records.

---

Anti-Pattern 7: Assuming User Context in Scheduler Tasks

❌ Wrong

def scheduled_task():
    # Creates document owned by Administrator!
    doc = frappe.get_doc({"doctype": "Task", "subject": "Auto-created"})
    doc.insert()
    
    # Permission check will use Administrator permissions!
    frappe.get_list("Private Doc")  # May return ALL records

Error: Wrong ownership, unexpected permission behavior

✅ Correct

def scheduled_task():
    # Explicitly set user if needed
    frappe.set_user("system@example.com")
    
    # Or explicitly set owner
    doc = frappe.get_doc({
        "doctype": "Task",
        "subject": "Auto-created",
        "owner": "target_user@example.com"
    })
    doc.insert(ignore_permissions=True)

Rule

Scheduler tasks run as Administrator. Explicitly set user context or owner when needed.

---

Anti-Pattern 8: Long-Running Task in Short/Default Queue

❌ Wrong

# hooks.py
scheduler_events = {
    "hourly": ["myapp.tasks.heavy_report"]  # Uses default queue!
}

# This will timeout after 5 minutes
def heavy_report():
    # 30-minute report generation
    ...

Error: Task times out, incomplete processing

✅ Correct

# hooks.py
scheduler_events = {
    "hourly_long": ["myapp.tasks.heavy_report"]  # Uses long queue
}

# Or for user-triggered
frappe.enqueue(
    "myapp.tasks.heavy_report",
    queue="long",
    timeout=3600  # 1 hour
)

Rule

*Use `_long scheduler events or queue="long"` for tasks over 5 minutes.**

---

Anti-Pattern 9: No Progress Feedback for User-Triggered Tasks

❌ Wrong

@frappe.whitelist()
def start_import():
    frappe.enqueue("myapp.tasks.import_10000_records")
    return {"message": "Started"}
    # User has no idea what's happening for 10 minutes!

Error: Poor UX, user may restart task thinking it failed

✅ Correct

def import_records(user):
    total = 10000
    
    for i, record in enumerate(records):
        process(record)
        
        if i % 100 == 0:
            frappe.publish_progress(
                percent=int((i / total) * 100),
                title="Importing records..."
            )
    
    frappe.publish_realtime(
        "msgprint",
        {"message": "Import complete!", "indicator": "green"},
        user=user
    )

Rule

Provide progress feedback for long user-triggered tasks using publish_progress and publish_realtime.

---

Anti-Pattern 10: Infinite Retry Without Limit

❌ Wrong

def sync_with_retry(record):
    try:
        sync_external(record)
    except Exception:
        # Retry forever if external system is down!
        frappe.enqueue("myapp.tasks.sync_with_retry", record=record)

Error: Infinite loop if external system is permanently down

✅ Correct

def sync_with_retry(record, attempt=1, max_attempts=3):
    try:
        sync_external(record)
    except Exception as e:
        if attempt < max_attempts:
            frappe.enqueue(
                "myapp.tasks.sync_with_retry",
                record=record,
                attempt=attempt + 1,
                max_attempts=max_attempts
            )
        else:
            frappe.log_error(
                f"Sync failed permanently: {record}",
                "Max Retries Exceeded"
            )

Rule

Always limit retry attempts. Log permanently failing tasks for manual review.

---

Anti-Pattern 11: Using "all" Event for Heavy Tasks

❌ Wrong

# hooks.py
scheduler_events = {
    "all": ["myapp.tasks.sync_external"]  # Runs every 60 seconds!
}

def sync_external():
    # Takes 2 minutes to complete
    ...

Error: Task can't complete before next run, queue backup

✅ Correct

# For tasks that take time, use appropriate frequency
scheduler_events = {
    "hourly": ["myapp.tasks.sync_external"]
}

# Or use cron for specific intervals
scheduler_events = {
    "cron": {
        "*/5 * * * *": ["myapp.tasks.sync_external"]  # Every 5 minutes
    }
}

Rule

"all" event is for quick tasks only (<60 seconds). Use hourly or cron for longer tasks.

---

Anti-Pattern 12: Not Using enqueue_after_commit

❌ Wrong

def on_submit(self):
    # Job might start before this transaction commits!
    frappe.enqueue("myapp.tasks.process", doc_name=self.name)

Error: Background job may not find the document (not committed yet)

✅ Correct

def on_submit(self):
    frappe.enqueue(
        "myapp.tasks.process",
        enqueue_after_commit=True,  # Wait for transaction to commit
        doc_name=self.name
    )

Rule

Use `enqueue_after_commit=True` when enqueueing from document events to ensure data is available.

---

Anti-Pattern 13: Ignoring Scheduler Status

❌ Wrong

# Deploy and assume scheduler is working
# Never check if tasks are actually running

Error: Tasks silently fail, issues discovered too late

✅ Correct

# Regular health checks
bench --site sitename scheduler status

# Check for failed jobs
bench --site sitename show-failed-jobs

# Monitor in UI
# Setup > Scheduled Job Log

# Add monitoring task
def scheduler_health_check():
    failed_count = frappe.db.count(
        "Scheduled Job Log",
        {"status": "Failed", "creation": [">=", frappe.utils.add_days(None, -1)]}
    )
    
    if failed_count > 10:
        alert_admin("Many scheduler failures!")

Rule

Actively monitor scheduler health. Check Scheduled Job Log regularly.

---

Anti-Pattern 14: Heavy Computation in Scheduler Event Handler

❌ Wrong

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

def generate_all_reports():
    # 2-hour task directly in scheduler event
    for report in get_all_reports():
        generate_report(report)  # Timeout!

Error: Scheduler event handler times out

✅ Correct

# Scheduler event just enqueues the actual work
def generate_all_reports():
    """Scheduler event handler - enqueues actual work."""
    reports = get_all_reports()
    
    for report in reports:
        frappe.enqueue(
            "myapp.tasks.generate_single_report",
            queue="long",
            report=report
        )

def generate_single_report(report):
    """Actual work in background job."""
    # Heavy processing here
    ...

Rule

Scheduler events should be thin wrappers. Enqueue heavy work to appropriate queues.

---

Quick Reference: Anti-Pattern Summary

Scheduler & Background Jobs - Complete Decision Trees

Detailed flowcharts for selecting the right scheduling approach.

---

Decision Tree: Task Type Selection

WHAT TRIGGERS THE TASK?
│
├─► Time-based (runs automatically)?
│   │
│   │ IS IT A FIXED SCHEDULE?
│   │
│   ├─► Yes (hourly, daily, weekly, monthly)
│   │   └─► scheduler_events in hooks.py
│   │       │
│   │       │ WHICH EVENT TYPE?
│   │       │
│   │       ├─► Every scheduler tick
│   │       │   └─► "all"
│   │       │       ⚠️ Must complete in <60 seconds
│   │       │
│   │       ├─► Once per hour
│   │       │   ├─► < 5 min → "hourly"
│   │       │   └─► 5-25 min → "hourly_long"
│   │       │
│   │       ├─► Once per day
│   │       │   ├─► < 5 min → "daily"
│   │       │   └─► 5-25 min → "daily_long"
│   │       │
│   │       ├─► Once per week
│   │       │   ├─► < 5 min → "weekly"
│   │       │   └─► 5-25 min → "weekly_long"
│   │       │
│   │       └─► Once per month
│   │           ├─► < 5 min → "monthly"
│   │           └─► 5-25 min → "monthly_long"
│   │
│   └─► Specific time/day?
│       └─► cron scheduler_events
│           - "0 9 * * 1-5" = 9am weekdays
│           - "0 0 1 * *" = 1st of month midnight
│           - "*/30 * * * *" = every 30 minutes
│
├─► User action (button click, form submit)?
│   └─► frappe.enqueue() in your code
│       │
│       │ HOW HEAVY IS THE TASK?
│       │
│       ├─► Quick (<30s) → queue="short"
│       ├─► Medium (<5min) → queue="default"
│       └─► Heavy (5-25min) → queue="long"
│
├─► System event (doc save, submit)?
│   │
│   │ SHOULD IT BLOCK THE USER?
│   │
│   ├─► No (heavy processing)
│   │   └─► frappe.enqueue() from doc_event/controller
│   │
│   └─► Yes (quick, must complete before response)
│       └─► Direct execution in controller
│
└─► One-time task (run once, not recurring)?
    └─► frappe.enqueue() directly
        - No hooks.py needed
        - Can be called from console or script

---

Decision Tree: Queue Selection

HOW LONG WILL THE TASK RUN?
│
├─► < 30 seconds?
│   │
│   │ IS QUICK RESPONSE NEEDED?
│   │
│   ├─► Yes (UI waiting)
│   │   └─► queue="short" (5 min timeout)
│   │       - Button callbacks
│   │       - Quick validations
│   │       - Small updates
│   │
│   └─► No (can wait)
│       └─► queue="default" (5 min timeout)
│
├─► 30 seconds - 5 minutes?
│   └─► queue="default" (5 min timeout)
│       - Most common tasks
│       - Standard scheduler events
│       - Medium datasets
│
├─► 5 - 25 minutes?
│   └─► queue="long" (25 min timeout)
│       - Large imports/exports
│       - Report generation
│       - Bulk operations
│       - Use *_long scheduler events
│
└─► > 25 minutes?
    └─► MUST split into chunks
        │
        │ CHUNKING STRATEGY:
        │
        ├─► Process batch, enqueue next batch
        │   - Self-chaining pattern
        │   - Track offset in arguments
        │
        ├─► Split by date range
        │   - Process one month at a time
        │   - Enqueue next month when done
        │
        └─► Split by record count
            - Process 1000 records per job
            - Use limit_start/limit_page_length

---

Decision Tree: Deduplication Strategy

CAN DUPLICATE JOBS CAUSE PROBLEMS?
│
├─► Yes (data corruption, duplicate processing)?
│   │
│   │ FRAPPE VERSION?
│   │
│   ├─► V15+ (recommended)
│   │   └─► Use job_id + is_job_enqueued()
│   │       ```python
│   │       from frappe.utils.background_jobs import is_job_enqueued
│   │       
│   │       job_id = f"task::{unique_key}"
│   │       if not is_job_enqueued(job_id):
│   │           frappe.enqueue(..., job_id=job_id)
│   │       ```
│   │
│   └─► V14 (legacy)
│       └─► Use job_name + manual check
│           ```python
│           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)
│           ```
│
├─► No (idempotent operation)?
│   └─► No deduplication needed
│       - Safe to run multiple times
│       - Each run produces same result
│
└─► Partially (some steps need protection)?
    └─► Deduplication at specific points
        - Dedup at enqueue time
        - Also check at execution start
        - Use database flags for critical sections

---

Decision Tree: Error Handling Strategy

WHAT HAPPENS IF THE TASK FAILS?
│
├─► Single record processing?
│   │
│   │ IS IT OKAY TO SKIP FAILURES?
│   │
│   ├─► Yes (log and continue)
│   │   └─► Try-except per record
│   │       - Log error with context
│   │       - Continue to next record
│   │       - Commit successful records
│   │
│   └─► No (must process all or nothing)
│       └─► Transaction-based
│           - Process all in try block
│           - Rollback entire batch on error
│           - Retry or alert on failure
│
├─► External API call?
│   │
│   │ IS RETRY APPROPRIATE?
│   │
│   ├─► Yes (transient errors likely)
│   │   └─► Retry pattern with backoff
│   │       - Track attempt count
│   │       - Exponential delay
│   │       - Max retry limit
│   │
│   └─► No (permanent failure likely)
│       └─► Fail fast, log, alert
│
├─► Critical business process?
│   └─► Alert on failure
│       - Send email to admin
│       - Create ToDo/Issue
│       - Log with high visibility
│
└─► Non-critical cleanup?
    └─► Log and ignore
        - Just log error
        - Continue processing
        - Review logs periodically

---

Decision Tree: Progress Reporting

SHOULD USER SEE PROGRESS?
│
├─► User-triggered task?
│   │
│   │ HOW LONG WILL IT TAKE?
│   │
│   ├─► < 10 seconds
│   │   └─► No progress needed
│   │       - Just show "Processing..."
│   │       - Show result when done
│   │
│   ├─► 10 seconds - 5 minutes
│   │   └─► Progress bar
│   │       ```python
│   │       frappe.publish_progress(
│   │           percent=50,
│   │           title="Processing..."
│   │       )
│   │       ```
│   │
│   └─► > 5 minutes
│       └─► Progress + notification on complete
│           - Progress bar during
│           - Email/realtime alert when done
│           - Link to results
│
├─► Scheduled task?
│   │
│   │ IS IT CRITICAL?
│   │
│   ├─► Yes (must know status)
│   │   └─► Log + optional alert
│   │       - Scheduled Job Log (automatic)
│   │       - Email summary for critical tasks
│   │
│   └─► No (routine maintenance)
│       └─► Scheduled Job Log only
│           - Automatic by scheduler
│           - Review periodically
│
└─► Background task (no user watching)?
    └─► Log only
        - frappe.logger().info()
        - Error Log for failures

---

Decision Tree: User Context

WHO SHOULD THE TASK RUN AS?
│
├─► Scheduled task?
│   └─► Runs as Administrator (default)
│       │
│       │ NEED SPECIFIC USER CONTEXT?
│       │
│       ├─► No (system operations)
│       │   └─► Leave as Administrator
│       │       - Full permissions
│       │       - Can access everything
│       │
│       └─► Yes (user-specific data)
│           └─► Set user explicitly
│               ```python
│               def scheduled_task():
│                   frappe.set_user("user@example.com")
│                   # Now runs as that user
│               ```
│
├─► User-triggered task?
│   │
│   │ SHOULD IT USE USER'S PERMISSIONS?
│   │
│   ├─► Yes (respect permissions)
│   │   └─► Pass user, set in task
│   │       ```python
│   │       frappe.enqueue(..., user=frappe.session.user)
│   │       
│   │       def task(user):
│   │           frappe.set_user(user)
│   │       ```
│   │
│   └─► No (needs elevated permissions)
│       └─► Run as Administrator (default)
│           - Use ignore_permissions=True
│           - Log original user for audit
│
└─► System event task?
    └─► Consider context carefully
        - May need original user for audit
        - May need Administrator for access
        - Document which context is used

---

Decision Tree: Monitoring Needs

HOW CRITICAL IS THE TASK?
│
├─► Business-critical (payments, invoices)?
│   └─► Full monitoring
│       - Scheduled Job Log (automatic)
│       - Custom success/failure logging
│       - Email alerts on failure
│       - Dashboard or report for status
│
├─► Important (sync, reports)?
│   └─► Standard monitoring
│       - Scheduled Job Log
│       - Error Log for failures
│       - Periodic manual review
│
├─► Routine (cleanup, maintenance)?
│   └─► Basic monitoring
│       - Scheduled Job Log only
│       - Review if problems reported
│
└─► Development/testing?
    └─► Debug logging
        - frappe.logger().debug()
        - Console output
        - Temporary, remove in production

---

Quick Reference: Decision Summary

Scheduler & Background Jobs - Complete Examples

Production-ready examples for common scheduling scenarios.

---

Example 1: Daily Database Maintenance

Complete solution for daily database cleanup and optimization.

hooks.py

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

myapp/maintenance.py

import frappe

def daily_database_maintenance():
    """
    Comprehensive daily database maintenance.
    Runs in long queue to allow up to 25 minutes.
    """
    stats = {
        "error_logs_deleted": 0,
        "activity_logs_deleted": 0,
        "email_queue_cleared": 0,
        "versions_cleaned": 0
    }
    
    try:
        # 1. Clean old Error Logs (>30 days)
        stats["error_logs_deleted"] = cleanup_old_records(
            doctype="Error Log",
            date_field="creation",
            days_old=30,
            batch_size=1000
        )
        
        # 2. Clean old Activity Logs (>90 days)
        stats["activity_logs_deleted"] = cleanup_old_records(
            doctype="Activity Log",
            date_field="creation",
            days_old=90,
            batch_size=1000
        )
        
        # 3. Clear sent/expired email queue (>7 days)
        stats["email_queue_cleared"] = cleanup_email_queue(days_old=7)
        
        # 4. Clean old document versions (>180 days, keep latest 5)
        stats["versions_cleaned"] = cleanup_document_versions(
            days_old=180,
            keep_latest=5
        )
        
        # Log summary
        frappe.logger("maintenance").info(
            f"Daily maintenance completed: {stats}"
        )
        
    except Exception:
        frappe.log_error(
            frappe.get_traceback(),
            "Daily Maintenance Failed"
        )

def cleanup_old_records(doctype, date_field, days_old, batch_size=500):
    """Generic cleanup for old records."""
    cutoff = frappe.utils.add_days(frappe.utils.nowdate(), -days_old)
    deleted = 0
    
    while True:
        records = frappe.get_all(
            doctype,
            filters={date_field: ["<", cutoff]},
            pluck="name",
            limit=batch_size
        )
        
        if not records:
            break
        
        for name in records:
            try:
                frappe.delete_doc(doctype, name, ignore_permissions=True)
                deleted += 1
            except Exception:
                pass  # Continue with next
        
        frappe.db.commit()
    
    return deleted

def cleanup_email_queue(days_old=7):
    """Clear old email queue entries."""
    cutoff = frappe.utils.add_days(frappe.utils.nowdate(), -days_old)
    
    deleted = frappe.db.sql("""
        DELETE FROM `tabEmail Queue`
        WHERE status IN ('Sent', 'Expired', 'Error')
        AND creation < %s
        LIMIT 5000
    """, cutoff)
    
    frappe.db.commit()
    return deleted

def cleanup_document_versions(days_old=180, keep_latest=5):
    """Clean old document versions, keeping recent ones."""
    cutoff = frappe.utils.add_days(frappe.utils.nowdate(), -days_old)
    
    # Get documents with many versions
    result = frappe.db.sql("""
        SELECT ref_doctype, docname, COUNT(*) as version_count
        FROM `tabVersion`
        WHERE creation < %s
        GROUP BY ref_doctype, docname
        HAVING version_count > %s
    """, (cutoff, keep_latest), as_dict=True)
    
    deleted = 0
    for row in result:
        # Get versions to delete (all except latest N)
        versions = frappe.db.sql("""
            SELECT name FROM `tabVersion`
            WHERE ref_doctype = %s AND docname = %s
            ORDER BY creation DESC
            LIMIT 999999 OFFSET %s
        """, (row.ref_doctype, row.docname, keep_latest), as_dict=True)
        
        for v in versions:
            frappe.delete_doc("Version", v.name, ignore_permissions=True)
            deleted += 1
        
        frappe.db.commit()
    
    return deleted

---

Example 2: External System Sync

Complete bidirectional sync with external REST API.

hooks.py

scheduler_events = {
    "hourly": [
        "myapp.sync.sync_from_external"
    ],
    "cron": {
        # Push updates every 15 minutes
        "*/15 * * * *": ["myapp.sync.push_to_external"]
    }
}

myapp/sync.py

import frappe
import requests
from datetime import datetime

API_BASE = "https://api.external.com/v1"
API_KEY = frappe.conf.get("external_api_key")

def sync_from_external():
    """
    Pull new/updated records from external system.
    Runs hourly.
    """
    last_sync = get_last_sync_time("pull")
    
    try:
        # Fetch updates from API
        response = requests.get(
            f"{API_BASE}/orders",
            headers={"Authorization": f"Bearer {API_KEY}"},
            params={"updated_since": last_sync.isoformat() if last_sync else None},
            timeout=60
        )
        response.raise_for_status()
        
        orders = response.json().get("data", [])
        
        synced = 0
        errors = 0
        
        for order_data in orders:
            try:
                sync_single_order(order_data)
                synced += 1
            except Exception as e:
                errors += 1
                frappe.log_error(
                    f"Sync error for order {order_data.get('id')}: {e}",
                    "External Sync Error"
                )
        
        frappe.db.commit()
        
        # Update sync timestamp
        set_last_sync_time("pull")
        
        frappe.logger("sync").info(
            f"Pull sync completed: {synced} synced, {errors} errors"
        )
        
    except requests.exceptions.RequestException as e:
        frappe.log_error(f"API request failed: {e}", "External Sync Failed")

def push_to_external():
    """
    Push pending local changes to external system.
    Runs every 15 minutes.
    """
    pending = frappe.get_all(
        "Sales Order",
        filters={
            "custom_external_sync_pending": 1,
            "docstatus": 1
        },
        fields=["name", "custom_external_id"],
        limit=50
    )
    
    for order in pending:
        try:
            doc = frappe.get_doc("Sales Order", order.name)
            
            # Prepare payload
            payload = {
                "order_id": doc.name,
                "customer": doc.customer,
                "total": doc.grand_total,
                "items": [
                    {"sku": i.item_code, "qty": i.qty, "rate": i.rate}
                    for i in doc.items
                ]
            }
            
            # Create or update
            if order.custom_external_id:
                response = requests.put(
                    f"{API_BASE}/orders/{order.custom_external_id}",
                    headers={"Authorization": f"Bearer {API_KEY}"},
                    json=payload,
                    timeout=30
                )
            else:
                response = requests.post(
                    f"{API_BASE}/orders",
                    headers={"Authorization": f"Bearer {API_KEY}"},
                    json=payload,
                    timeout=30
                )
            
            response.raise_for_status()
            external_id = response.json().get("id")
            
            # Update local record
            frappe.db.set_value(
                "Sales Order", order.name,
                {
                    "custom_external_id": external_id,
                    "custom_external_sync_pending": 0,
                    "custom_last_synced": frappe.utils.now()
                }
            )
            
        except Exception as e:
            frappe.log_error(
                f"Push sync failed for {order.name}: {e}",
                "External Push Error"
            )
    
    frappe.db.commit()

def sync_single_order(order_data):
    """Create or update local order from external data."""
    external_id = order_data.get("id")
    
    # Check if exists
    existing = frappe.db.get_value(
        "Sales Order",
        {"custom_external_id": external_id},
        "name"
    )
    
    if existing:
        # Update existing
        doc = frappe.get_doc("Sales Order", existing)
        doc.custom_external_status = order_data.get("status")
        doc.save(ignore_permissions=True)
    else:
        # Create new
        doc = frappe.get_doc({
            "doctype": "Sales Order",
            "customer": order_data.get("customer"),
            "custom_external_id": external_id,
            "custom_external_sync_pending": 0,
            # ... map other fields
        })
        doc.insert(ignore_permissions=True)

def get_last_sync_time(sync_type):
    """Get last sync timestamp."""
    return frappe.db.get_value(
        "Singles",
        {"doctype": "Sync Settings", "field": f"last_{sync_type}_sync"},
        "value"
    )

def set_last_sync_time(sync_type):
    """Update last sync timestamp."""
    frappe.db.set_value(
        "Sync Settings", None,
        f"last_{sync_type}_sync",
        frappe.utils.now()
    )

---

Example 3: Bulk Data Import with Progress

Complete solution for importing large CSV files.

myapp/api.py

import frappe
from frappe.utils.background_jobs import is_job_enqueued

@frappe.whitelist()
def start_csv_import(file_url, doctype, update_existing=False):
    """
    Start CSV import in background.
    
    Args:
        file_url: URL of uploaded CSV file
        doctype: Target DocType
        update_existing: If True, update existing records
    """
    job_id = f"csv_import::{frappe.session.user}::{file_url}"
    
    if is_job_enqueued(job_id):
        frappe.throw("This file is already being imported")
    
    # Create import log
    import_log = frappe.get_doc({
        "doctype": "Data Import Log",
        "import_file": file_url,
        "reference_doctype": doctype,
        "status": "Queued",
        "started_by": frappe.session.user
    })
    import_log.insert(ignore_permissions=True)
    frappe.db.commit()
    
    frappe.enqueue(
        "myapp.importer.run_csv_import",
        queue="long",
        timeout=7200,  # 2 hours
        job_id=job_id,
        file_url=file_url,
        doctype=doctype,
        update_existing=update_existing,
        user=frappe.session.user,
        import_log_name=import_log.name
    )
    
    return {
        "message": "Import started",
        "import_log": import_log.name
    }

myapp/importer.py

import frappe
import csv
from io import StringIO

def run_csv_import(file_url, doctype, update_existing, user, import_log_name):
    """
    Main import function.
    """
    frappe.set_user(user)
    
    # Update status
    update_import_log(import_log_name, status="Running")
    
    try:
        # Read file
        file_content = frappe.get_doc("File", {"file_url": file_url}).get_content()
        
        if isinstance(file_content, bytes):
            file_content = file_content.decode("utf-8")
        
        reader = csv.DictReader(StringIO(file_content))
        rows = list(reader)
        total = len(rows)
        
        if total == 0:
            update_import_log(
                import_log_name,
                status="Error",
                error_message="No data rows found in file"
            )
            return
        
        # Process rows
        success = 0
        failed = 0
        errors = []
        
        for i, row in enumerate(rows):
            try:
                import_single_row(doctype, row, update_existing)
                success += 1
                
                # Commit every 100 rows
                if i % 100 == 0:
                    frappe.db.commit()
                    
                    # Update progress
                    percent = int((i / total) * 100)
                    frappe.publish_progress(
                        percent=percent,
                        title="Importing...",
                        description=f"Row {i + 1} of {total}"
                    )
                    
                    update_import_log(
                        import_log_name,
                        processed=i + 1,
                        success_count=success,
                        error_count=failed
                    )
                    
            except Exception as e:
                failed += 1
                errors.append({
                    "row": i + 2,  # +2 for header and 0-index
                    "error": str(e),
                    "data": row
                })
        
        frappe.db.commit()
        
        # Final status
        status = "Success" if failed == 0 else "Completed with Errors"
        
        update_import_log(
            import_log_name,
            status=status,
            processed=total,
            success_count=success,
            error_count=failed,
            errors=frappe.as_json(errors[:100])  # Store first 100 errors
        )
        
        # Notify user
        notify_import_complete(user, import_log_name, success, failed)
        
    except Exception as e:
        frappe.db.rollback()
        frappe.log_error(frappe.get_traceback(), f"Import Failed: {import_log_name}")
        
        update_import_log(
            import_log_name,
            status="Error",
            error_message=str(e)
        )
        
        frappe.publish_realtime(
            "msgprint",
            {"message": "Import failed. Check Data Import Log.", "indicator": "red"},
            user=user
        )

def import_single_row(doctype, row, update_existing):
    """Import or update single row."""
    # Get identifying field (usually name or custom identifier)
    identifier = row.get("name") or row.get("id")
    
    if update_existing and identifier:
        existing = frappe.db.exists(doctype, identifier)
        if existing:
            doc = frappe.get_doc(doctype, identifier)
            doc.update(row)
            doc.save(ignore_permissions=True)
            return
    
    # Create new
    doc = frappe.get_doc({
        "doctype": doctype,
        **row
    })
    doc.insert(ignore_permissions=True)

def update_import_log(name, **kwargs):
    """Update import log record."""
    frappe.db.set_value("Data Import Log", name, kwargs)
    frappe.db.commit()

def notify_import_complete(user, import_log_name, success, failed):
    """Send completion notification."""
    indicator = "green" if failed == 0 else "orange"
    message = f"Import complete: {success} success, {failed} failed"
    
    frappe.publish_realtime(
        "msgprint",
        {
            "message": f"{message}. <a href='/app/data-import-log/{import_log_name}'>View Log</a>",
            "indicator": indicator
        },
        user=user
    )

---

Example 4: Email Digest/Newsletter

Automated weekly summary email to users.

hooks.py

scheduler_events = {
    "cron": {
        # Every Monday at 8:00 AM
        "0 8 * * 1": ["myapp.newsletter.send_weekly_digest"]
    }
}

myapp/newsletter.py

import frappe

def send_weekly_digest():
    """
    Send weekly digest email to all subscribed users.
    """
    # Get digest data
    digest_data = compile_weekly_digest()
    
    # Get subscribers
    subscribers = frappe.get_all(
        "User",
        filters={
            "enabled": 1,
            "custom_weekly_digest": 1  # Custom checkbox field
        },
        fields=["name", "email", "full_name", "language"]
    )
    
    sent = 0
    for subscriber in subscribers:
        try:
            send_digest_to_user(subscriber, digest_data)
            sent += 1
        except Exception:
            frappe.log_error(
                f"Failed to send digest to {subscriber.email}",
                "Digest Send Error"
            )
    
    frappe.logger("newsletter").info(f"Weekly digest sent to {sent} subscribers")

def compile_weekly_digest():
    """Compile data for weekly digest."""
    last_week = frappe.utils.add_days(frappe.utils.nowdate(), -7)
    
    return {
        "period_start": last_week,
        "period_end": frappe.utils.nowdate(),
        
        "new_orders": frappe.db.count(
            "Sales Order",
            {"creation": [">=", last_week], "docstatus": 1}
        ),
        
        "total_revenue": get_weekly_revenue(last_week),
        
        "top_items": get_top_selling_items(last_week, limit=5),
        
        "new_customers": frappe.db.count(
            "Customer",
            {"creation": [">=", last_week]}
        ),
        
        "open_issues": frappe.db.count(
            "Issue",
            {"status": "Open"}
        )
    }

def get_weekly_revenue(since_date):
    """Get total revenue for the week."""
    result = frappe.db.sql("""
        SELECT COALESCE(SUM(grand_total), 0)
        FROM `tabSales Invoice`
        WHERE docstatus = 1
        AND posting_date >= %s
    """, since_date)
    return result[0][0] if result else 0

def get_top_selling_items(since_date, limit=5):
    """Get top selling items for the week."""
    return frappe.db.sql("""
        SELECT 
            sii.item_code,
            sii.item_name,
            SUM(sii.qty) as total_qty,
            SUM(sii.amount) as total_amount
        FROM `tabSales Invoice Item` sii
        JOIN `tabSales Invoice` si ON si.name = sii.parent
        WHERE si.docstatus = 1
        AND si.posting_date >= %s
        GROUP BY sii.item_code
        ORDER BY total_amount DESC
        LIMIT %s
    """, (since_date, limit), as_dict=True)

def send_digest_to_user(subscriber, digest_data):
    """Send personalized digest to user."""
    # Render template
    message = frappe.render_template(
        "myapp/templates/emails/weekly_digest.html",
        {
            "user": subscriber,
            "data": digest_data
        }
    )
    
    frappe.sendmail(
        recipients=[subscriber.email],
        subject=f"Weekly Digest - {digest_data['period_end']}",
        message=message,
        reference_doctype="User",
        reference_name=subscriber.name
    )

myapp/templates/emails/weekly_digest.html

<div style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto;">
    <h1 style="color: #2c3e50;">Weekly Digest</h1>
    <p>Hi {{ user.full_name }},</p>
    <p>Here's your weekly summary for {{ data.period_start }} to {{ data.period_end }}:</p>
    
    <table style="width: 100%; border-collapse: collapse; margin: 20px 0;">
        <tr style="background: #3498db; color: white;">
            <td style="padding: 15px; text-align: center;">
                <h2 style="margin: 0;">{{ data.new_orders }}</h2>
                <p style="margin: 5px 0 0;">New Orders</p>
            </td>
            <td style="padding: 15px; text-align: center;">
                <h2 style="margin: 0;">{{ frappe.format(data.total_revenue, {'fieldtype': 'Currency'}) }}</h2>
                <p style="margin: 5px 0 0;">Revenue</p>
            </td>
            <td style="padding: 15px; text-align: center;">
                <h2 style="margin: 0;">{{ data.new_customers }}</h2>
                <p style="margin: 5px 0 0;">New Customers</p>
            </td>
        </tr>
    </table>
    
    {% if data.top_items %}
    <h3>Top Selling Items</h3>
    <table style="width: 100%; border-collapse: collapse;">
        <tr style="background: #f5f5f5;">
            <th style="padding: 10px; text-align: left;">Item</th>
            <th style="padding: 10px; text-align: right;">Qty</th>
            <th style="padding: 10px; text-align: right;">Amount</th>
        </tr>
        {% for item in data.top_items %}
        <tr>
            <td style="padding: 10px; border-bottom: 1px solid #eee;">{{ item.item_name }}</td>
            <td style="padding: 10px; border-bottom: 1px solid #eee; text-align: right;">{{ item.total_qty }}</td>
            <td style="padding: 10px; border-bottom: 1px solid #eee; text-align: right;">
                {{ frappe.format(item.total_amount, {'fieldtype': 'Currency'}) }}
            </td>
        </tr>
        {% endfor %}
    </table>
    {% endif %}
    
    <p style="color: #666; margin-top: 30px;">
        <a href="{{ frappe.utils.get_url() }}">Go to Dashboard</a>
    </p>
</div>

---

Example 5: Document Expiry Notifications

Notify users about expiring documents (contracts, licenses, etc.).

hooks.py

scheduler_events = {
    "daily": [
        "myapp.expiry.check_expiring_documents"
    ]
}

myapp/expiry.py

import frappe

EXPIRY_CONFIGS = [
    {
        "doctype": "Contract",
        "date_field": "end_date",
        "days_before": [30, 7, 1],  # Notify 30, 7, 1 days before
        "notify_field": "contract_owner"  # User field to notify
    },
    {
        "doctype": "License",
        "date_field": "valid_till",
        "days_before": [60, 30, 7],
        "notify_field": "assigned_to"
    },
    {
        "doctype": "Insurance Policy",
        "date_field": "expiry_date",
        "days_before": [30, 14, 7],
        "notify_field": "owner"
    }
]

def check_expiring_documents():
    """
    Check for expiring documents and send notifications.
    """
    today = frappe.utils.nowdate()
    
    for config in EXPIRY_CONFIGS:
        check_doctype_expiry(config, today)

def check_doctype_expiry(config, today):
    """Check expiry for a specific DocType."""
    doctype = config["doctype"]
    date_field = config["date_field"]
    notify_field = config["notify_field"]
    
    for days in config["days_before"]:
        target_date = frappe.utils.add_days(today, days)
        
        # Find documents expiring on target date
        expiring = frappe.get_all(
            doctype,
            filters={
                date_field: target_date,
                "docstatus": ["!=", 2]  # Not cancelled
            },
            fields=["name", date_field, notify_field, "owner"]
        )
        
        for doc in expiring:
            # Check if notification already sent
            if was_notified(doctype, doc.name, days):
                continue
            
            # Get user to notify
            notify_user = doc.get(notify_field) or doc.get("owner")
            
            if notify_user:
                send_expiry_notification(
                    doctype=doctype,
                    doc_name=doc.name,
                    expiry_date=doc.get(date_field),
                    days_remaining=days,
                    user=notify_user
                )
                
                # Mark as notified
                mark_notified(doctype, doc.name, days)

def send_expiry_notification(doctype, doc_name, expiry_date, days_remaining, user):
    """Send expiry notification to user."""
    # Create system notification
    frappe.get_doc({
        "doctype": "Notification Log",
        "for_user": user,
        "type": "Alert",
        "document_type": doctype,
        "document_name": doc_name,
        "subject": f"{doctype} {doc_name} expires in {days_remaining} days",
        "email_content": f"""
            <p>The following {doctype} will expire on {expiry_date}:</p>
            <p><strong>{doc_name}</strong></p>
            <p><a href="/app/{frappe.scrub(doctype)}/{doc_name}">View Document</a></p>
        """
    }).insert(ignore_permissions=True)
    
    # Also send email
    user_email = frappe.db.get_value("User", user, "email")
    if user_email:
        frappe.sendmail(
            recipients=[user_email],
            subject=f"⚠️ {doctype} Expiring: {doc_name}",
            message=f"""
                <p>This is a reminder that the following {doctype} will expire in {days_remaining} days:</p>
                <p><strong>{doc_name}</strong></p>
                <p>Expiry Date: {expiry_date}</p>
                <p><a href="{frappe.utils.get_url()}/app/{frappe.scrub(doctype)}/{doc_name}">
                    View Document
                </a></p>
            """
        )
    
    frappe.db.commit()

def was_notified(doctype, doc_name, days):
    """Check if notification was already sent."""
    return frappe.db.exists(
        "Expiry Notification Log",
        {
            "reference_doctype": doctype,
            "reference_name": doc_name,
            "days_before": days
        }
    )

def mark_notified(doctype, doc_name, days):
    """Record that notification was sent."""
    frappe.get_doc({
        "doctype": "Expiry Notification Log",
        "reference_doctype": doctype,
        "reference_name": doc_name,
        "days_before": days,
        "sent_on": frappe.utils.nowdate()
    }).insert(ignore_permissions=True)

---

Quick Reference: Example Summary

Scheduler & Background Jobs - Implementation Workflows

Step-by-step implementation patterns for all scheduling scenarios.

---

Workflow 1: Basic Scheduled Task

Goal: Run a cleanup task daily

Step 1: Create Task Module

# myapp/tasks.py
import frappe

def daily_cleanup():
    """
    Clean up old temporary files daily.
    
    IMPORTANT:
    - NO arguments allowed for scheduler tasks!
    - Runs as Administrator user
    - Must be fast enough (<5 min for daily, <25 min for daily_long)
    """
    # Get cutoff date
    cutoff = frappe.utils.add_days(frappe.utils.nowdate(), -7)
    
    # Find old files
    old_files = frappe.get_all(
        "File",
        filters={
            "is_private": 1,
            "attached_to_doctype": "",
            "creation": ["<", cutoff]
        },
        pluck="name",
        limit=500  # Process in manageable batches
    )
    
    deleted = 0
    for file_name in old_files:
        try:
            frappe.delete_doc("File", file_name, ignore_permissions=True)
            deleted += 1
        except Exception:
            frappe.log_error(f"Could not delete file: {file_name}")
    
    frappe.db.commit()
    frappe.logger("scheduler").info(f"Daily cleanup: deleted {deleted} files")

Step 2: Register in hooks.py

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

Step 3: Deploy and Enable

# REQUIRED: Migrate to register scheduler events
bench --site sitename migrate

# Enable scheduler if not already enabled
bench --site sitename scheduler enable

# Verify registration
bench --site sitename scheduler status

Step 4: Verify in UI

1. Go to: Setup > Scheduled Job Type
2. Find: myapp.tasks.daily_cleanup
3. Check: Frequency = Daily, Stopped = No

4. Go to: Scheduled Job Log
5. After task runs, verify success/failure

---

Workflow 2: Cron-Based Scheduled Task

Goal: Send summary email at 9am on weekdays

Step 1: Create Task

# myapp/tasks.py
import frappe

def weekday_summary_email():
    """Send daily summary at 9am weekdays."""
    
    # Generate summary data
    yesterday = frappe.utils.add_days(frappe.utils.nowdate(), -1)
    
    summary = {
        "new_orders": frappe.db.count(
            "Sales Order",
            {"creation": [">=", yesterday], "docstatus": 1}
        ),
        "new_invoices": frappe.db.count(
            "Sales Invoice",
            {"creation": [">=", yesterday], "docstatus": 1}
        ),
        "total_revenue": get_yesterday_revenue()
    }
    
    # Get recipients
    recipients = frappe.get_all(
        "User",
        filters={
            "enabled": 1,
            "user_type": "System User"
        },
        pluck="email"
    )
    
    # Send email
    frappe.sendmail(
        recipients=recipients,
        subject=f"Daily Summary - {yesterday}",
        message=frappe.render_template(
            "myapp/templates/emails/daily_summary.html",
            summary
        )
    )

def get_yesterday_revenue():
    result = frappe.db.sql("""
        SELECT COALESCE(SUM(grand_total), 0)
        FROM `tabSales Invoice`
        WHERE docstatus = 1
        AND posting_date = DATE_SUB(CURDATE(), INTERVAL 1 DAY)
    """)
    return result[0][0] if result else 0

Step 2: Register with Cron Expression

# myapp/hooks.py
scheduler_events = {
    "cron": {
        # At 9:00 AM, Monday through Friday
        "0 9 * * 1-5": [
            "myapp.tasks.weekday_summary_email"
        ]
    }
}

Cron Expression Cheat Sheet

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

Examples:
"0 9 * * 1-5"     → 9:00 AM, Monday-Friday
"0 0 * * *"       → Midnight daily
"0 0 1 * *"       → Midnight, 1st of month
"*/15 * * * *"    → Every 15 minutes
"0 */2 * * *"     → Every 2 hours
"30 8 * * 1"      → 8:30 AM every Monday
"0 18 * * 5"      → 6:00 PM every Friday
"0 0 15 * *"      → Midnight, 15th of month

Step 3: Deploy

bench --site sitename migrate

---

Workflow 3: User-Triggered Background Job

Goal: Export large dataset when user clicks button

Step 1: Create Export Task

# myapp/tasks.py
import frappe
import csv
from io import StringIO

def export_customers(user, filters=None):
    """
    Export customers to CSV file.
    
    Args:
        user: User who requested export (for notifications)
        filters: Optional dict of filters
    """
    # Set user context for permissions
    frappe.set_user(user)
    
    try:
        # Get data
        customers = frappe.get_all(
            "Customer",
            filters=filters or {},
            fields=["name", "customer_name", "customer_group", "territory", "email_id"],
            limit_page_length=0  # All records
        )
        
        if not customers:
            notify_user(user, "No customers found matching filters", "orange")
            return
        
        # Create CSV
        output = StringIO()
        writer = csv.DictWriter(output, fieldnames=customers[0].keys())
        writer.writeheader()
        writer.writerows(customers)
        
        # Save file
        file_doc = frappe.get_doc({
            "doctype": "File",
            "file_name": f"customers_export_{frappe.utils.nowdate()}.csv",
            "content": output.getvalue(),
            "is_private": 1
        })
        file_doc.insert(ignore_permissions=True)
        
        frappe.db.commit()
        
        # Notify user
        notify_user(
            user,
            f"Export complete! <a href='{file_doc.file_url}'>Download CSV</a> ({len(customers)} records)",
            "green"
        )
        
    except Exception:
        frappe.db.rollback()
        frappe.log_error(frappe.get_traceback(), "Customer Export Failed")
        notify_user(user, "Export failed. Check Error Log.", "red")

def notify_user(user, message, indicator="blue"):
    """Send realtime notification to user."""
    frappe.publish_realtime(
        "msgprint",
        {"message": message, "indicator": indicator},
        user=user
    )

Step 2: Create API Endpoint

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

@frappe.whitelist()
def start_customer_export(filters=None):
    """
    Start customer export in background.
    
    Args:
        filters: JSON string of filter dict
    """
    job_id = f"customer_export::{frappe.session.user}"
    
    # Prevent duplicate exports
    if is_job_enqueued(job_id):
        frappe.throw("An export is already in progress. Please wait.")
    
    # Parse filters
    if filters:
        filters = frappe.parse_json(filters)
    
    # Enqueue export
    frappe.enqueue(
        "myapp.tasks.export_customers",
        queue="long",
        timeout=1800,  # 30 minutes
        job_id=job_id,
        user=frappe.session.user,
        filters=filters
    )
    
    return {"message": "Export started. You'll be notified when complete."}

Step 3: Create Client Interface

// In Client Script or custom page
frappe.ui.form.on("Customer", {
    refresh: function(frm) {
        frm.add_custom_button(__("Export All Customers"), function() {
            frappe.call({
                method: "myapp.api.start_customer_export",
                args: {
                    filters: JSON.stringify({
                        "customer_group": frm.doc.customer_group
                    })
                },
                callback: function(r) {
                    if (r.message) {
                        frappe.show_alert({
                            message: r.message.message,
                            indicator: "blue"
                        });
                    }
                }
            });
        });
    }
});

---

Workflow 4: Long-Running Task with Chunking

Goal: Process 100,000+ records without timeout

Step 1: Create Chunked Task

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

def process_invoices_batch(
    offset=0,
    batch_size=500,
    total=None,
    processed=0,
    errors=0,
    job_id=None
):
    """
    Process invoices in batches, chaining to next batch.
    
    Pattern:
    1. Process current batch
    2. If more records, enqueue next batch
    3. If done, send completion notification
    """
    
    # First call: count total
    if total is None:
        total = frappe.db.count(
            "Sales Invoice",
            {"status": "Draft", "custom_processed": 0}
        )
        job_id = f"process_invoices::{frappe.utils.now()}"
    
    # Get current batch
    invoices = frappe.get_all(
        "Sales Invoice",
        filters={"status": "Draft", "custom_processed": 0},
        fields=["name"],
        limit_start=0,  # Always start at 0 since we mark processed
        limit_page_length=batch_size
    )
    
    if not invoices:
        # All done!
        frappe.logger("scheduler").info(
            f"Invoice processing complete: {processed} processed, {errors} errors"
        )
        return
    
    # Process batch
    batch_errors = 0
    for inv in invoices:
        try:
            process_single_invoice(inv.name)
            processed += 1
        except Exception:
            errors += 1
            batch_errors += 1
            frappe.log_error(
                frappe.get_traceback(),
                f"Invoice Processing Error: {inv.name}"
            )
    
    frappe.db.commit()
    
    # Log progress
    progress = int((processed / total) * 100) if total else 0
    frappe.logger("scheduler").info(
        f"Invoice processing: {progress}% ({processed}/{total})"
    )
    
    # Check if more to process
    remaining = frappe.db.count(
        "Sales Invoice",
        {"status": "Draft", "custom_processed": 0}
    )
    
    if remaining > 0:
        # Enqueue next batch
        frappe.enqueue(
            "myapp.tasks.process_invoices_batch",
            queue="long",
            job_id=job_id,
            offset=offset + batch_size,
            batch_size=batch_size,
            total=total,
            processed=processed,
            errors=errors
        )

def process_single_invoice(invoice_name):
    """Process a single invoice."""
    doc = frappe.get_doc("Sales Invoice", invoice_name)
    
    # Your processing logic here
    doc.custom_processed = 1
    doc.save(ignore_permissions=True)

Step 2: Start Processing

# myapp/api.py
@frappe.whitelist()
def start_invoice_processing():
    """Start batch invoice processing."""
    job_id = "process_invoices::batch"
    
    if is_job_enqueued(job_id):
        return {"message": "Processing already in progress"}
    
    frappe.enqueue(
        "myapp.tasks.process_invoices_batch",
        queue="long",
        job_id=job_id
    )
    
    return {"message": "Invoice processing started"}

---

Workflow 5: External API Sync with Retry

Goal: Sync with external system, handle failures gracefully

Step 1: Create Sync Task with Retry Logic

# myapp/tasks.py
import frappe
import requests

def sync_to_external_system():
    """
    Sync pending records to external API.
    Scheduler task - runs hourly.
    """
    pending = frappe.get_all(
        "Sales Order",
        filters={
            "custom_sync_status": ["in", ["Pending", "Failed"]],
            "docstatus": 1
        },
        fields=["name", "custom_sync_attempts"],
        limit=100
    )
    
    for record in pending:
        sync_single_record_with_retry(
            record.name,
            attempt=record.custom_sync_attempts or 0
        )

def sync_single_record_with_retry(order_name, attempt=0, max_attempts=3):
    """
    Sync single record with retry on failure.
    """
    try:
        doc = frappe.get_doc("Sales Order", order_name)
        
        # Call external API
        response = requests.post(
            "https://api.external.com/orders",
            json=doc.as_dict(),
            timeout=30
        )
        response.raise_for_status()
        
        # Success
        frappe.db.set_value(
            "Sales Order", order_name,
            {
                "custom_sync_status": "Synced",
                "custom_sync_date": frappe.utils.now(),
                "custom_sync_attempts": attempt + 1
            }
        )
        frappe.db.commit()
        
    except requests.exceptions.Timeout:
        handle_sync_failure(order_name, "Timeout", attempt, max_attempts)
        
    except requests.exceptions.HTTPError as e:
        if e.response.status_code >= 500:
            # Server error - retry
            handle_sync_failure(order_name, f"Server Error: {e}", attempt, max_attempts)
        else:
            # Client error - don't retry
            frappe.db.set_value(
                "Sales Order", order_name,
                {
                    "custom_sync_status": "Error",
                    "custom_sync_error": str(e)
                }
            )
            frappe.db.commit()
            
    except Exception as e:
        handle_sync_failure(order_name, str(e), attempt, max_attempts)

def handle_sync_failure(order_name, error, attempt, max_attempts):
    """Handle sync failure with optional retry."""
    frappe.db.rollback()
    
    if attempt < max_attempts:
        # Schedule retry with exponential backoff
        frappe.db.set_value(
            "Sales Order", order_name,
            {
                "custom_sync_status": "Retry Pending",
                "custom_sync_attempts": attempt + 1,
                "custom_sync_error": error
            }
        )
        frappe.db.commit()
        
        # Enqueue retry (will be picked up by next scheduler run)
        frappe.logger("sync").warning(
            f"Sync failed for {order_name}, attempt {attempt + 1}/{max_attempts}"
        )
    else:
        # Max attempts reached
        frappe.db.set_value(
            "Sales Order", order_name,
            {
                "custom_sync_status": "Failed",
                "custom_sync_attempts": attempt + 1,
                "custom_sync_error": f"Max attempts reached: {error}"
            }
        )
        frappe.db.commit()
        
        frappe.log_error(
            f"Sync failed permanently for {order_name} after {max_attempts} attempts: {error}",
            "External Sync Failed"
        )

Step 2: Register in hooks.py

# myapp/hooks.py
scheduler_events = {
    "hourly": [
        "myapp.tasks.sync_to_external_system"
    ]
}

---

Workflow 6: Task with Progress and Notification

Goal: Show user progress during long operation

Step 1: Create Task with Progress Reporting

# myapp/tasks.py
import frappe

def generate_report_with_progress(report_type, user, params=None):
    """
    Generate report with progress updates.
    """
    frappe.set_user(user)
    
    try:
        # Initial notification
        frappe.publish_realtime(
            "msgprint",
            {"message": "Starting report generation...", "indicator": "blue"},
            user=user
        )
        
        # Get total records
        total = get_report_record_count(report_type, params)
        
        if total == 0:
            frappe.publish_realtime(
                "msgprint",
                {"message": "No data found for report", "indicator": "orange"},
                user=user
            )
            return
        
        # Process with progress
        results = []
        for i, record in enumerate(get_report_records(report_type, params)):
            results.append(process_record_for_report(record))
            
            # Update progress every 10%
            if i % max(1, total // 10) == 0:
                percent = int((i / total) * 100)
                frappe.publish_progress(
                    percent=percent,
                    title=f"Processing: {percent}%",
                    description=f"Record {i + 1} of {total}"
                )
        
        # Generate final report
        file_url = create_report_file(results, report_type)
        
        # Success notification
        frappe.publish_realtime(
            "msgprint",
            {
                "message": f"Report ready! <a href='{file_url}' target='_blank'>Download Report</a>",
                "indicator": "green"
            },
            user=user
        )
        
        # Also send email
        frappe.sendmail(
            recipients=[user],
            subject=f"Your {report_type} Report is Ready",
            message=f"Download your report: {file_url}"
        )
        
    except Exception:
        frappe.db.rollback()
        frappe.log_error(frappe.get_traceback(), f"Report Generation Failed: {report_type}")
        
        frappe.publish_realtime(
            "msgprint",
            {"message": "Report generation failed. Check Error Log.", "indicator": "red"},
            user=user
        )

Step 2: API to Start Report

# myapp/api.py
@frappe.whitelist()
def generate_report(report_type, params=None):
    """Start report generation in background."""
    from frappe.utils.background_jobs import is_job_enqueued
    
    job_id = f"report::{report_type}::{frappe.session.user}"
    
    if is_job_enqueued(job_id):
        frappe.throw("A report is already being generated. Please wait.")
    
    frappe.enqueue(
        "myapp.tasks.generate_report_with_progress",
        queue="long",
        timeout=3600,
        job_id=job_id,
        report_type=report_type,
        user=frappe.session.user,
        params=frappe.parse_json(params) if params else None
    )
    
    return {"message": "Report generation started. You'll see progress updates."}

---

Workflow 7: Scheduler with Callbacks

Goal: Execute follow-up actions on job completion

Step 1: Define Callbacks

# myapp/tasks.py
import frappe

def on_import_success(job, connection, result, *args, **kwargs):
    """Called when import job succeeds."""
    user = kwargs.get("user")
    import_id = kwargs.get("import_id")
    
    # Update import status
    frappe.db.set_value("Data Import", import_id, "status", "Success")
    frappe.db.commit()
    
    # Notify user
    frappe.publish_realtime(
        "eval_js",
        f'frappe.show_alert({{message: "Import {import_id} completed successfully!", indicator: "green"}})',
        user=user
    )
    
    # Send email
    frappe.sendmail(
        recipients=[user],
        subject=f"Import {import_id} Complete",
        message="Your data import has completed successfully."
    )

def on_import_failure(job, connection, type, value, traceback):
    """Called when import job fails."""
    # Extract kwargs from job
    import_id = job.kwargs.get("import_id")
    user = job.kwargs.get("user")
    
    # Update import status
    frappe.db.set_value(
        "Data Import", import_id,
        {
            "status": "Error",
            "error_message": str(value)
        }
    )
    frappe.db.commit()
    
    # Log error
    frappe.log_error(
        f"Import {import_id} failed: {value}\n{traceback}",
        "Import Failed"
    )
    
    # Notify user
    if user:
        frappe.publish_realtime(
            "eval_js",
            f'frappe.show_alert({{message: "Import {import_id} failed!", indicator: "red"}})',
            user=user
        )

Step 2: Enqueue with Callbacks

# myapp/api.py
@frappe.whitelist()
def start_import(import_id):
    """Start import with success/failure callbacks."""
    frappe.enqueue(
        "myapp.tasks.run_import",
        queue="long",
        timeout=3600,
        on_success="myapp.tasks.on_import_success",
        on_failure="myapp.tasks.on_import_failure",
        import_id=import_id,
        user=frappe.session.user
    )
    
    return {"message": "Import started"}

---

Workflow 8: Monitoring and Alerting

Goal: Monitor scheduler health and alert on failures

Step 1: Create Health Check Task

# myapp/tasks.py
import frappe
from frappe.utils import now_datetime, get_datetime

def scheduler_health_check():
    """
    Check scheduler health and alert if issues found.
    Run every 15 minutes via cron.
    """
    issues = []
    
    # Check for failed jobs in last hour
    failed_jobs = frappe.db.count(
        "Scheduled Job Log",
        {
            "status": "Failed",
            "creation": [">=", frappe.utils.add_to_date(None, hours=-1)]
        }
    )
    
    if failed_jobs > 5:
        issues.append(f"{failed_jobs} scheduler jobs failed in last hour")
    
    # Check for stuck jobs (running > 30 min)
    stuck_jobs = frappe.db.sql("""
        SELECT name, scheduled_job_type, status
        FROM `tabScheduled Job Log`
        WHERE status = 'Running'
        AND creation < DATE_SUB(NOW(), INTERVAL 30 MINUTE)
    """, as_dict=True)
    
    if stuck_jobs:
        issues.append(f"{len(stuck_jobs)} jobs appear stuck")
    
    # Check worker status
    from frappe.utils.background_jobs import get_workers_status
    workers = get_workers_status()
    
    idle_workers = sum(1 for w in workers if w.get("status") == "idle")
    if idle_workers == 0 and len(workers) > 0:
        issues.append("All workers are busy - possible backlog")
    
    # Alert if issues found
    if issues:
        frappe.sendmail(
            recipients=["admin@example.com"],
            subject="⚠️ Scheduler Health Alert",
            message=f"""
            <h3>Scheduler Issues Detected</h3>
            <ul>
            {"".join(f"<li>{issue}</li>" for issue in issues)}
            </ul>
            <p>Check the Background Jobs page for details.</p>
            """
        )
        
        frappe.log_error(
            "\n".join(issues),
            "Scheduler Health Issues"
        )

Step 2: Register Health Check

# myapp/hooks.py
scheduler_events = {
    "cron": {
        "*/15 * * * *": ["myapp.tasks.scheduler_health_check"]
    }
}

---

Quick Reference: Workflow Summary