server-scripts

Frappe Server Scripts Reference

Complete reference for server-side Python development in Frappe Framework.

When to Use This Skill

• Writing document controllers
• Creating whitelisted API endpoints
• Handling document lifecycle events
• Background job processing
• Database operations and queries
• Permission checks and validation
• Email and notification handling

Controller Location

my_app/
└── my_module/
    └── doctype/
        └── my_doctype/
            └── my_doctype.py    # Python controller

Document Controller

Complete Controller Template

# my_doctype.py
import frappe
from frappe import _
from frappe.model.document import Document
from frappe.utils import nowdate, nowtime, flt, cint, getdate, add_days


class MyDocType(Document):
    # ===== NAMING =====

    def autoname(self):
        """Custom naming logic"""
        self.name = f"{self.prefix}-{frappe.generate_hash()[:8].upper()}"

    def before_naming(self):
        """Called before autoname"""
        pass

    # ===== VALIDATION =====

    def before_validate(self):
        """Called before validate"""
        self.set_defaults()

    def validate(self):
        """Main validation - called on insert and update"""
        self.validate_dates()
        self.validate_amounts()
        self.calculate_totals()
        self.set_status()

    def before_save(self):
        """Called after validate, before database write"""
        self.update_modified_info()

    # ===== INSERT =====

    def before_insert(self):
        """Called before new document is inserted"""
        self.set_initial_values()

    def after_insert(self):
        """Called after new document is inserted"""
        self.create_related_documents()
        self.send_notification()

    # ===== UPDATE =====

    def on_update(self):
        """Called after document is saved (insert or update)"""
        self.update_related_documents()
        self.clear_cache()

    def after_save(self):
        """Called after on_update, always runs"""
        pass

    def on_change(self):
        """Called when document changes in database"""
        pass

    # ===== SUBMISSION =====

    def before_submit(self):
        """Called before document is submitted"""
        self.validate_for_submit()

    def on_submit(self):
        """Called after document is submitted"""
        self.create_gl_entries()
        self.update_stock()

    def on_update_after_submit(self):
        """Called when submitted doc is updated (limited fields)"""
        pass

    # ===== CANCELLATION =====

    def before_cancel(self):
        """Called before document is cancelled"""
        self.validate_cancellation()

    def on_cancel(self):
        """Called after document is cancelled"""
        self.reverse_gl_entries()
        self.reverse_stock()

    # ===== DELETION =====

    def before_delete(self):
        """Called before document is deleted"""
        self.check_dependencies()

    def after_delete(self):
        """Called after document is deleted"""
        self.cleanup_related()

    def on_trash(self):
        """Called when document is trashed"""
        pass

    def after_restore(self):
        """Called after document is restored from trash"""
        pass

    # ===== CUSTOM METHODS =====

    def set_defaults(self):
        """Set default values"""
        if not self.posting_date:
            self.posting_date = nowdate()
        if not self.company:
            self.company = frappe.defaults.get_user_default("Company")

    def validate_dates(self):
        """Validate date fields"""
        if self.end_date and getdate(self.start_date) > getdate(self.end_date):
            frappe.throw(_("End Date cannot be before Start Date"))

        if getdate(self.posting_date) > getdate(nowdate()):
            frappe.throw(_("Posting Date cannot be in the future"))

    def validate_amounts(self):
        """Validate amount fields"""
        for item in self.items:
            if flt(item.qty) <= 0:
                frappe.throw(_("Row {0}: Quantity must be greater than 0").format(item.idx))
            if flt(item.rate) < 0:
                frappe.throw(_("Row {0}: Rate cannot be negative").format(item.idx))

    def calculate_totals(self):
        """Calculate document totals"""
        self.total = 0
        for item in self.items:
            item.amount = flt(item.qty) * flt(item.rate)
            self.total += item.amount

        self.tax_amount = flt(self.total) * flt(self.tax_rate) / 100
        self.grand_total = flt(self.total) + flt(self.tax_amount)

    def set_status(self):
        """Set document status based on state"""
        if self.docstatus == 0:
            self.status = "Draft"
        elif self.docstatus == 1:
            if self.is_completed():
                self.status = "Completed"
            else:
                self.status = "Submitted"
        elif self.docstatus == 2:
            self.status = "Cancelled"

    def is_completed(self):
        """Check if document is completed"""
        return all(item.delivered_qty >= item.qty for item in self.items)

Whitelisted APIs

Basic API

@frappe.whitelist()
def get_customer_details(customer):
    """Get customer details

    Args:
        customer (str): Customer ID

    Returns:
        dict: Customer details with outstanding amount
    """
    if not customer:
        frappe.throw(_("Customer is required"))

    doc = frappe.get_doc("Customer", customer)

    return {
        "customer_name": doc.customer_name,
        "customer_type": doc.customer_type,
        "territory": doc.territory,
        "credit_limit": flt(doc.credit_limit),
        "outstanding": get_customer_outstanding(customer)
    }


@frappe.whitelist()
def create_invoice(customer, items):
    """Create sales invoice from data

    Args:
        customer (str): Customer ID
        items (str): JSON string of items

    Returns:
        str: Invoice name
    """
    items = frappe.parse_json(items)

    doc = frappe.get_doc({
        "doctype": "Sales Invoice",
        "customer": customer,
        "items": [{
            "item_code": item.get("item_code"),
            "qty": flt(item.get("qty")),
            "rate": flt(item.get("rate"))
        } for item in items]
    })

    doc.insert()
    doc.submit()

    return doc.name

Guest API

@frappe.whitelist(allow_guest=True)
def get_public_data():
    """Public API - no login required"""
    return {
        "status": "ok",
        "message": "This is public data"
    }

Method-Restricted API

@frappe.whitelist(methods=["POST"])
def create_record(data):
    """Only accepts POST requests"""
    data = frappe.parse_json(data)
    doc = frappe.get_doc(data)
    doc.insert()
    return {"name": doc.name}


@frappe.whitelist(methods=["GET", "POST"])
def flexible_endpoint(**kwargs):
    """Accepts GET and POST"""
    return kwargs

Permission-Checked API

@frappe.whitelist()
def sensitive_operation(doctype, name):
    """API with permission check"""
    # Check permission
    if not frappe.has_permission(doctype, "write", name):
        frappe.throw(_("Not permitted"), frappe.PermissionError)

    # Proceed with operation
    doc = frappe.get_doc(doctype, name)
    # ... do something

    return {"status": "success"}

Database Operations

Reading Data

# Get single document
doc = frappe.get_doc("Customer", "CUST-001")

# Get with filters
doc = frappe.get_doc("Customer", {"customer_name": "John Corp"})

# Get single value
name = frappe.db.get_value("Customer", "CUST-001", "customer_name")

# Get multiple values
values = frappe.db.get_value("Customer", "CUST-001",
    ["customer_name", "territory"], as_dict=True)

# Get list
customers = frappe.db.get_all("Customer",
    filters={"status": "Active"},
    fields=["name", "customer_name", "territory"],
    order_by="customer_name asc",
    limit=10
)

# Complex filters
invoices = frappe.db.get_all("Sales Invoice",
    filters={
        "status": ["in", ["Paid", "Unpaid"]],
        "grand_total": [">", 1000],
        "posting_date": [">=", "2024-01-01"],
        "customer": ["like", "%Corp%"]
    },
    fields=["name", "customer", "grand_total"]
)

# Pluck single field
names = frappe.db.get_all("Customer",
    filters={"status": "Active"},
    pluck="name"
)

# Count
count = frappe.db.count("Customer", {"status": "Active"})

# Exists check
exists = frappe.db.exists("Customer", "CUST-001")

Raw SQL

# Simple query
result = frappe.db.sql("""
    SELECT name, customer_name, grand_total
    FROM `tabSales Invoice`
    WHERE status = %s AND grand_total > %s
    ORDER BY creation DESC
    LIMIT 10
""", ("Paid", 1000), as_dict=True)

# Named parameters
result = frappe.db.sql("""
    SELECT * FROM `tabCustomer`
    WHERE territory = %(territory)s
    AND status = %(status)s
""", {"territory": "West", "status": "Active"}, as_dict=True)

# Aggregation
total = frappe.db.sql("""
    SELECT SUM(grand_total) as total
    FROM `tabSales Invoice`
    WHERE status = 'Paid'
""")[0][0] or 0

Writing Data

# Create document
doc = frappe.get_doc({
    "doctype": "Customer",
    "customer_name": "New Customer",
    "customer_type": "Company"
})
doc.insert()

# Update document
doc = frappe.get_doc("Customer", "CUST-001")
doc.customer_name = "Updated Name"
doc.save()

# Quick update (bypasses controller)
frappe.db.set_value("Customer", "CUST-001", "status", "Inactive")

# Update multiple fields
frappe.db.set_value("Customer", "CUST-001", {
    "status": "Inactive",
    "disabled": 1
})

# Delete
frappe.delete_doc("Customer", "CUST-001")

# Commit transaction
frappe.db.commit()

# Rollback
frappe.db.rollback()

Background Jobs

Enqueue Jobs

# Basic enqueue
frappe.enqueue(
    "my_app.tasks.process_data",
    queue="default",
    customer="CUST-001"
)

# With options
frappe.enqueue(
    method="my_app.tasks.heavy_task",
    queue="long",        # short, default, long
    timeout=1800,        # 30 minutes
    is_async=True,
    job_name="Heavy Task",
    now=False,           # True to run immediately
    enqueue_after_commit=True,
    # Task arguments
    document_name="DOC-001",
    data={"key": "value"}
)

Task Function

# my_app/tasks.py
import frappe

def process_data(customer):
    """Background task"""
    frappe.init(site=frappe.local.site)
    frappe.connect()

    try:
        # Process logic
        doc = frappe.get_doc("Customer", customer)
        doc.last_processed = frappe.utils.now()
        doc.save()
        frappe.db.commit()
    except Exception:
        frappe.log_error(title="Process Data Failed")
        raise
    finally:
        frappe.destroy()

Scheduled Jobs (hooks.py)

scheduler_events = {
    "all": [
        "my_app.tasks.every_minute"
    ],
    "daily": [
        "my_app.tasks.daily_report"
    ],
    "hourly": [
        "my_app.tasks.hourly_sync"
    ],
    "cron": {
        "0 9 * * 1": [
            "my_app.tasks.monday_morning"
        ],
        "*/15 * * * *": [
            "my_app.tasks.every_15_min"
        ]
    }
}

Error Handling

from frappe import _
from frappe.exceptions import ValidationError, PermissionError

def my_function():
    # Throw with message
    frappe.throw(_("Invalid data"))

    # Throw with title
    frappe.throw(_("Cannot proceed"), title=_("Error"))

    # Throw with exception type
    frappe.throw(_("Permission denied"), exc=PermissionError)

    # Message without stopping
    frappe.msgprint(_("Warning: Check your data"))

    # Log error
    frappe.log_error(
        title="My Error",
        message=frappe.get_traceback()
    )

    # Try-except
    try:
        risky_operation()
    except Exception:
        frappe.log_error("Operation failed")
        frappe.throw(_("Something went wrong"))

Utilities

from frappe.utils import (
    nowdate, nowtime, now_datetime, today,
    getdate, get_datetime,
    add_days, add_months, add_years,
    date_diff, time_diff_in_seconds,
    flt, cint, cstr,
    fmt_money, rounded,
    strip_html, escape_html
)

# Date operations
today = nowdate()  # "2024-01-15"
week_later = add_days(nowdate(), 7)
month_end = frappe.utils.get_last_day(nowdate())

# Number operations
amount = flt(value, 2)  # Float with precision
count = cint(value)     # Integer

# Formatting
money = fmt_money(1234.56, currency="USD")

# Current user
user = frappe.session.user
roles = frappe.get_roles()

Email & Notifications

# Send email
frappe.sendmail(
    recipients=["user@example.com"],
    subject="Subject",
    message="Email body",
    reference_doctype="Sales Invoice",
    reference_name="SINV-00001"
)

# With template
frappe.sendmail(
    recipients=["user@example.com"],
    subject="Order Confirmation",
    template="order_confirmation",
    args={
        "customer_name": "John",
        "order_id": "ORD-001"
    }
)

# Real-time notification
frappe.publish_realtime(
    "msgprint",
    {"message": "Task completed"},
    user="user@example.com"
)

Document Events via Hooks

# hooks.py
doc_events = {
    "Sales Invoice": {
        "validate": "my_app.overrides.validate_invoice",
        "on_submit": "my_app.overrides.on_submit_invoice",
        "on_cancel": "my_app.overrides.on_cancel_invoice"
    },
    "*": {
        "on_update": "my_app.overrides.log_all_changes"
    }
}

# my_app/overrides.py
import frappe

def validate_invoice(doc, method):
    """Called during Sales Invoice validation"""
    if doc.grand_total > 100000:
        if not doc.manager_approval:
            frappe.throw(_("Manager approval required for orders above 100,000"))

def on_submit_invoice(doc, method):
    """Called when Sales Invoice is submitted"""
    create_delivery_note(doc)
    notify_warehouse(doc)