Frappe Errors Database

Frappe Database Error Diagnosis & Resolution

Cross-ref: frappe-core-database (API syntax), frappe-errors-controllers (controller errors).

---

Error-to-Fix Mapping Table

---

Exception Hierarchy

Exception
├── frappe.ValidationError (HTTP 417)
│   ├── frappe.MandatoryError
│   ├── frappe.LinkValidationError
│   ├── frappe.CharacterLengthExceededError
│   ├── frappe.DataTooLongException
│   ├── frappe.UniqueValidationError
│   ├── frappe.UpdateAfterSubmitError
│   └── frappe.DataError
├── frappe.DoesNotExistError (HTTP 404)
├── frappe.DuplicateEntryError (HTTP 409)  ← inherits NameError
├── frappe.TimestampMismatchError
├── frappe.LinkExistsError
├── frappe.QueryTimeoutError
├── frappe.QueryDeadlockError
├── frappe.TooManyWritesError
├── frappe.InReadOnlyMode (HTTP 503)
└── frappe.db.InternalError  ← MariaDB/Postgres driver error

---

frappe.db.sql() Parameter Format

# ❌ WRONG — %s with positional tuple (works but fragile)
frappe.db.sql("SELECT * FROM `tabItem` WHERE name = %s", ("ITEM-001",))

# ❌ WRONG — f-string or .format() — SQL INJECTION!
frappe.db.sql(f"SELECT * FROM `tabItem` WHERE name = '{item_name}'")
frappe.db.sql("SELECT * FROM `tabItem` WHERE name = '{}'".format(item_name))

# ❌ WRONG — bare % operator
frappe.db.sql("SELECT * FROM `tabItem` WHERE name = '%s'" % item_name)

# ✅ CORRECT — named parameters with dict (ALWAYS use this)
frappe.db.sql(
    "SELECT * FROM `tabItem` WHERE name = %(name)s AND warehouse = %(wh)s",
    {"name": item_name, "wh": warehouse},
    as_dict=True
)

# ✅ CORRECT — frappe.qb (query builder, no injection risk)
Item = frappe.qb.DocType("Item")
result = (
    frappe.qb.from_(Item)
    .select(Item.name, Item.item_name)
    .where(Item.warehouse == warehouse)
    .run(as_dict=True)
)

Rule: ALWAYS use %(name)s with a dict parameter. NEVER use string formatting for SQL values.

---

get_value Returns None: Not an Exception

# ❌ DANGEROUS — get_value returns None, not raises
credit = frappe.db.get_value("Customer", "CUST-001", "credit_limit")
if credit > 1000:  # TypeError: '>' not supported between NoneType and int
    pass

# ✅ CORRECT — handle None explicitly
credit = frappe.db.get_value("Customer", "CUST-001", "credit_limit")
if credit is None:
    frappe.throw(_("Customer not found"))
credit = credit or 0  # Default to 0 if field is empty

# ✅ CORRECT — get_value with as_dict for multiple fields
data = frappe.db.get_value("Customer", "CUST-001",
    ["credit_limit", "disabled"], as_dict=True)
if not data:  # None when record not found
    frappe.throw(_("Customer not found"))
if data.disabled:
    frappe.throw(_("Customer is disabled"))

Key behavior by method:

---

Handling Each Exception Type

DuplicateEntryError

# Pattern: Insert with duplicate handling
def create_or_get(doctype, data):
    try:
        doc = frappe.get_doc({"doctype": doctype, **data})
        doc.insert()
        return doc
    except frappe.DuplicateEntryError:
        # Race condition safe: someone else created it
        name = frappe.db.get_value(doctype, data, "name")
        return frappe.get_doc(doctype, name)

TimestampMismatchError

# Pattern: Concurrent edit detection
try:
    doc = frappe.get_doc("Sales Invoice", name)
    doc.update(updates)
    doc.save()
except frappe.TimestampMismatchError:
    frappe.throw(
        _("Document modified by another user. Please refresh and try again."),
        title=_("Concurrent Edit")
    )

LinkValidationError & MandatoryError

# Pattern: Pre-validate before save
def safe_create_invoice(data):
    errors = []

    # Check mandatory fields
    if not data.get("customer"):
        errors.append(_("Customer is required"))
    if not data.get("items"):
        errors.append(_("At least one item is required"))

    # Check link validity
    if data.get("customer"):
        if not frappe.db.exists("Customer", data["customer"]):
            errors.append(_("Customer '{0}' not found").format(data["customer"]))

    if errors:
        frappe.throw("<br>".join(errors))

    doc = frappe.get_doc({"doctype": "Sales Invoice", **data})
    doc.insert()
    return doc

CharacterLengthExceededError

# Pattern: Truncate before save
def safe_set_description(doc, description):
    max_len = 140  # Match field length in DocType
    if len(description) > max_len:
        description = description[:max_len - 3] + "..."
        frappe.msgprint(_("Description truncated to {0} characters").format(max_len))
    doc.description = description

QueryTimeoutError [v15+]

# Pattern: Paginated query to avoid timeout
def get_large_report(filters):
    try:
        return frappe.db.sql(query, filters, as_dict=True)
    except frappe.QueryTimeoutError:
        frappe.log_error(frappe.get_traceback(), "Report Query Timeout")
        frappe.throw(
            _("Report too large. Please narrow your date range or add filters."),
            title=_("Query Timeout")
        )

InReadOnlyMode

# Pattern: Check before write
def safe_write(doctype, name, field, value):
    if frappe.flags.in_import:
        frappe.db.set_value(doctype, name, field, value)
        return
    try:
        frappe.db.set_value(doctype, name, field, value)
    except frappe.InReadOnlyMode:
        frappe.log_error(f"Write blocked: {doctype}/{name}", "Read-Only Mode")
        frappe.throw(_("System is in read-only mode. Please try again later."))

---

Transaction Deadlocks

# ❌ CAUSES DEADLOCKS — long transaction with many writes
def process_all():
    for inv in frappe.get_all("Sales Invoice", limit=10000):
        doc = frappe.get_doc("Sales Invoice", inv.name)
        doc.custom_field = "value"
        doc.save()  # Each save locks rows; other processes wait

# ✅ CORRECT — batch with commits to release locks
def process_all():
    invoices = frappe.get_all("Sales Invoice", limit=10000)
    BATCH = 100
    for i in range(0, len(invoices), BATCH):
        for inv in invoices[i:i + BATCH]:
            frappe.db.set_value("Sales Invoice", inv.name, "custom_field", "value")
        frappe.db.commit()  # Release locks after each batch

# ✅ CORRECT — retry on deadlock
import time
def with_deadlock_retry(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func()
        except frappe.QueryDeadlockError:
            if attempt < max_retries - 1:
                frappe.db.rollback()
                time.sleep(0.5 * (attempt + 1))
            else:
                raise

---

MariaDB Gone Away / Too Many Connections

# Pattern: Connection recovery
def reliable_operation():
    try:
        return frappe.db.sql("SELECT 1")
    except frappe.db.InternalError as e:
        msg = str(e).lower()
        if "gone away" in msg or "lost connection" in msg:
            frappe.db.connect()  # Reconnect
            return frappe.db.sql("SELECT 1")
        if "too many connections" in msg:
            frappe.log_error("Too many DB connections", "Connection Pool")
            frappe.throw(_("Server busy. Please try again in a moment."))
        raise  # Unknown InternalError — re-raise

Prevention:

• Set wait_timeout in MariaDB config (default 28800s)
• Check max_connections setting matches your workload
• Use connection pooling in production (Gunicorn workers)

---

Transaction Rules

When to Commit

Savepoints for Partial Rollback

def complex_operation():
    frappe.db.savepoint("before_risky")
    try:
        risky_database_operation()
    except Exception:
        frappe.db.rollback(save_point="before_risky")
        safe_alternative()  # Continue with fallback

# Transaction hooks [v15+]
frappe.db.after_commit.add(lambda: send_notification())
frappe.db.after_rollback.add(lambda: cleanup_files())

---

SQL Injection Prevention

# ❌ INJECTION VULNERABLE — all of these
frappe.db.sql(f"SELECT * FROM `tabItem` WHERE name = '{user_input}'")
frappe.db.sql("SELECT * FROM `tabItem` WHERE name = '%s'" % user_input)
frappe.db.sql("SELECT * FROM `tabItem` WHERE name = '{}'".format(user_input))

# ❌ ALSO VULNERABLE — in permission_query_conditions
def query_conditions(user):
    return f"owner = '{user}'"  # Unescaped!

# ✅ SAFE — parameterized query
frappe.db.sql("SELECT * FROM `tabItem` WHERE name = %(name)s", {"name": user_input})

# ✅ SAFE — frappe.db.escape() for dynamic SQL (permission hooks)
def query_conditions(user):
    return f"owner = {frappe.db.escape(user)}"

# ✅ SAFE — query builder
Item = frappe.qb.DocType("Item")
frappe.qb.from_(Item).where(Item.name == user_input).run()

# ✅ SAFE — ORM methods
frappe.get_all("Item", filters={"name": user_input})
frappe.db.get_value("Item", user_input, "item_name")

---

db.set_value Silent Failure

# ❌ DANGEROUS — no error if record doesn't exist
frappe.db.set_value("Customer", "NONEXISTENT", "status", "Active")
# Returns without error! No rows updated.

# ✅ ALWAYS verify existence before set_value
if not frappe.db.exists("Customer", customer_name):
    frappe.throw(_("Customer '{0}' not found").format(customer_name))
frappe.db.set_value("Customer", customer_name, "status", "Active")

# Note: set_value skips validate/on_update hooks
# Use doc.save() when you need validation to run

---

Critical Rules

ALWAYS

1. Use %(name)s dict params in frappe.db.sql() — NEVER string formatting 2. Check frappe.db.exists() before get_doc() — or catch DoesNotExistError 3. Handle DuplicateEntryError on every insert() call 4. Handle TimestampMismatchError on every save() in APIs 5. Call frappe.db.commit() in scheduler and background jobs 6. Paginate large queries — use limit parameter 7. Check get_value() result for None before using it 8. Use frappe.db.escape() in dynamic SQL strings

NEVER

1. Use string formatting (f"", .format(), %) for SQL values 2. Call frappe.db.commit() in controller hooks or doc_events 3. Catch bare Exception and pass — log or re-raise specific types 4. Assume db.set_value() succeeded — it fails silently on missing records 5. Expose raw database error messages to users — log details, show generic message 6. Run unbounded queries without limit — memory/timeout risk

---

Quick Reference: Exception Handling

try:
    doc = frappe.get_doc("Customer", name)
except frappe.DoesNotExistError:
    frappe.throw(_("Not found"))

try:
    doc.insert()
except frappe.DuplicateEntryError:
    existing = frappe.db.get_value("Customer", filters, "name")
except frappe.MandatoryError as e:
    frappe.throw(_("Missing required field: {0}").format(e))

try:
    doc.save()
except frappe.TimestampMismatchError:
    frappe.throw(_("Document modified. Please refresh."))
except frappe.CharacterLengthExceededError:
    frappe.throw(_("Text too long for field"))

try:
    frappe.delete_doc("Customer", name)
except frappe.LinkExistsError:
    frappe.throw(_("Cannot delete — linked documents exist"))

try:
    frappe.db.sql(query, values)
except frappe.QueryTimeoutError:          # [v15+]
    frappe.throw(_("Query too slow. Add filters."))
except frappe.QueryDeadlockError:
    frappe.db.rollback()                   # Retry with backoff
except frappe.db.InternalError as e:
    frappe.log_error(frappe.get_traceback(), "DB Error")

---

Reference Files

---

See Also

• frappe-core-database — Database API syntax and query builder
• frappe-errors-controllers — Controller error handling
• frappe-errors-hooks — Hook error handling
• frappe-core-permissions — Permission patterns

Anti-Patterns — Database Error Handling

Common mistakes to avoid when handling database errors in Frappe/ERPNext.

---

1. SQL Injection via String Formatting

Problem

# ALL of these are SQL INJECTION vulnerabilities
query = f"SELECT * FROM `tabCustomer` WHERE name = '{customer_name}'"
query = "SELECT * FROM `tabCustomer` WHERE name = '{}'".format(customer_name)
query = "SELECT * FROM `tabCustomer` WHERE name = '%s'" % customer_name

Fix

# ALWAYS use named parameters with dict
frappe.db.sql(
    "SELECT * FROM `tabCustomer` WHERE name = %(name)s",
    {"name": customer_name},
    as_dict=True
)

# Or use ORM — no injection risk
frappe.db.get_value("Customer", customer_name, "*", as_dict=True)

Why: String formatting allows SQL injection. ALWAYS use %(name)s with dict params.

---

2. Using %s Instead of %(name)s

Problem

# Positional %s is fragile and error-prone
frappe.db.sql(
    "SELECT * FROM `tabItem` WHERE name = %s AND warehouse = %s",
    ("ITEM-001", "Main")
)
# Easy to get parameter order wrong!

Fix

# Named parameters are self-documenting and order-independent
frappe.db.sql(
    "SELECT * FROM `tabItem` WHERE name = %(name)s AND warehouse = %(wh)s",
    {"name": "ITEM-001", "wh": "Main"},
    as_dict=True
)

Why: Named parameters prevent order-dependent bugs and are clearer to read.

---

3. Not Checking get_value() for None

Problem

credit = frappe.db.get_value("Customer", "CUST-001", "credit_limit")
if credit > 1000:  # TypeError if customer not found (None > 1000)
    apply_discount()

Fix

credit = frappe.db.get_value("Customer", "CUST-001", "credit_limit")
if credit is None:
    frappe.throw(_("Customer not found"))
credit = credit or 0
if credit > 1000:
    apply_discount()

Why: get_value() returns None when record not found. ALWAYS check before using.

---

4. Not Checking Existence Before get_doc

Problem

# Crashes with DoesNotExistError
doc = frappe.get_doc("Customer", customer_name)
doc.update(data)
doc.save()

Fix

if not frappe.db.exists("Customer", customer_name):
    frappe.throw(_("Customer not found"))
doc = frappe.get_doc("Customer", customer_name)
doc.update(data)
doc.save()

Why: get_doc() raises DoesNotExistError. Check first or catch the exception.

---

5. Ignoring DuplicateEntryError on Insert

Problem

doc = frappe.get_doc({"doctype": "Customer", **data})
doc.insert()  # Crashes on duplicate!

Fix

try:
    doc = frappe.get_doc({"doctype": "Customer", **data})
    doc.insert()
except frappe.DuplicateEntryError:
    existing = frappe.db.get_value("Customer", {"customer_name": data.get("customer_name")})
    return {"name": existing, "existing": True}

Why: Unique constraints cause DuplicateEntryError. Handle it on every insert.

---

6. Assuming db.set_value Always Works

Problem

frappe.db.set_value("Customer", "NONEXISTENT", "synced", 1)
return "Success"  # LIE — nothing was updated

Fix

if not frappe.db.exists("Customer", customer_name):
    frappe.throw(_("Customer not found"))
frappe.db.set_value("Customer", customer_name, "synced", 1)

Why: db.set_value() silently does nothing if record doesn't exist.

---

7. Committing in Controller Hooks

Problem

class SalesOrder(Document):
    def validate(self):
        self.calculate_totals()
        frappe.db.commit()  # BREAKS transaction!

    def on_update(self):
        frappe.db.commit()  # ALSO WRONG

Fix

class SalesOrder(Document):
    def validate(self):
        self.calculate_totals()
        # Framework handles commit

    def on_update(self):
        self.update_linked()
        # Framework handles commit

Why: Manual commits break transaction management. Frappe auto-commits after web requests.

---

8. Missing Commit in Background Jobs

Problem

def background_sync():
    for item in frappe.get_all("Item", limit=100):
        frappe.db.set_value("Item", item.name, "synced", 1)
    # ALL CHANGES LOST — no auto-commit in background!

Fix

def background_sync():
    for item in frappe.get_all("Item", limit=100):
        frappe.db.set_value("Item", item.name, "synced", 1)
    frappe.db.commit()  # REQUIRED

Why: Background jobs and scheduler tasks have no auto-commit.

---

9. Swallowing Database Errors

Problem

try:
    doc = frappe.get_doc("Customer", name)
    doc.save()
except Exception:
    pass  # Silent failure — impossible to debug

Fix

try:
    doc = frappe.get_doc("Customer", name)
    doc.save()
except frappe.DoesNotExistError:
    frappe.throw(_("Customer not found"))
except frappe.TimestampMismatchError:
    frappe.throw(_("Modified by another user. Refresh and retry."))
except frappe.ValidationError as e:
    frappe.throw(str(e))
except Exception:
    frappe.log_error(frappe.get_traceback(), "Customer Save Error")
    frappe.throw(_("An error occurred"))

Why: Catch specific exceptions first. NEVER silently swallow errors.

---

10. Not Handling Empty Query Results

Problem

result = frappe.db.sql("SELECT name FROM `tabSales Invoice` WHERE customer = %s LIMIT 1", customer)
return result[0][0]  # IndexError if no results!

Fix

result = frappe.db.sql("SELECT name FROM `tabSales Invoice` WHERE customer = %(c)s LIMIT 1",
                       {"c": customer})
return result[0][0] if result else None

Why: Empty result sets cause IndexError. ALWAYS check before accessing.

---

11. N+1 Query Pattern

Problem

def get_details(names):
    details = []
    for name in names:
        doc = frappe.get_doc("Customer", name)  # N queries!
        details.append(doc.as_dict())
    return details

Fix

def get_details(names):
    return frappe.get_all(
        "Customer",
        filters={"name": ["in", names]},
        fields=["name", "customer_name", "credit_limit"]
    )  # 1 query!

Why: N+1 queries are extremely slow. ALWAYS batch fetch.

---

12. No Limit on Queries

Problem

all_invoices = frappe.get_all("Sales Invoice")  # Could return millions!

Fix

invoices = frappe.get_all("Sales Invoice", limit=100)
# Or paginate:
invoices = frappe.get_all("Sales Invoice", limit_start=0, limit_page_length=100)

Why: Unbounded queries cause memory exhaustion and timeouts.

---

13. Exposing Database Errors to Users

Problem

try:
    return frappe.db.sql(query, filters)
except Exception as e:
    frappe.throw(str(e))  # Exposes SQL details!

Fix

try:
    return frappe.db.sql(query, filters)
except frappe.db.InternalError:
    frappe.log_error(frappe.get_traceback(), "Query Error")
    frappe.throw(_("Database error. Please contact support."))

Why: Database error messages can expose table names, column names, and SQL structure.

---

14. Race Condition on Get-or-Create

Problem

def get_or_create(name):
    if not frappe.db.exists("Customer", name):
        doc = frappe.get_doc({"doctype": "Customer", "customer_name": name})
        doc.insert()  # DuplicateEntryError — someone else created it!
    return frappe.get_doc("Customer", name)

Fix

def get_or_create(name):
    if not frappe.db.exists("Customer", name):
        try:
            doc = frappe.get_doc({"doctype": "Customer", "customer_name": name})
            doc.insert()
        except frappe.DuplicateEntryError:
            pass  # Race condition — someone else created it
    return frappe.get_doc("Customer", name)

Why: Between exists() and insert(), another process can create the record.

---

15. Not Handling Concurrent Edits

Problem

doc = frappe.get_doc("Customer", name)
doc.update(data)
doc.save()  # TimestampMismatchError if modified by another user!

Fix

try:
    doc = frappe.get_doc("Customer", name)
    doc.update(data)
    doc.save()
except frappe.TimestampMismatchError:
    frappe.throw(_("Document modified. Please refresh and try again."))

Why: Concurrent edits cause TimestampMismatchError. Handle gracefully in APIs.

---

16. Using get_doc When get_value Suffices

Problem

# Loads ENTIRE document just for one field
doc = frappe.get_doc("Customer", name)
return doc.credit_limit

Fix

# Only fetches the needed field
return frappe.db.get_value("Customer", name, "credit_limit") or 0

Why: get_doc() loads entire document with children. Use get_value() for single fields.

---

17. Catching Generic Exception for All DB Errors

Problem

try:
    frappe.delete_doc("Customer", name)
except Exception as e:
    frappe.throw(str(e))  # No specific handling

Fix

try:
    frappe.delete_doc("Customer", name)
except frappe.DoesNotExistError:
    frappe.throw(_("Customer not found"))
except frappe.LinkExistsError:
    frappe.throw(_("Cannot delete — linked documents exist"))
except Exception:
    frappe.log_error(frappe.get_traceback(), "Delete Error")
    frappe.throw(_("Delete failed. Contact support."))

Why: Specific exceptions allow specific error messages and recovery strategies.

---

Quick Checklist: Database Code Review

Before deploying:

• [ ] All SQL uses %(name)s with dict (no string formatting)
• [ ] get_value() results checked for None
• [ ] Existence checked before get_doc() (or exception caught)
• [ ] DuplicateEntryError handled on every insert()
• [ ] TimestampMismatchError handled on save() in APIs
• [ ] db.set_value() preceded by existence check
• [ ] No frappe.db.commit() in controller hooks / doc_events
• [ ] frappe.db.commit() in background/scheduler tasks
• [ ] Database errors logged, not swallowed
• [ ] Empty results handled (no blind array access)
• [ ] Queries have limits / pagination
• [ ] Specific exceptions caught before generic Exception
• [ ] Database errors not exposed to users
• [ ] Race conditions handled on get-or-create
• [ ] get_value() used instead of get_doc() for single fields

Examples — Database Error Handling

Complete working examples of error handling for Frappe/ERPNext database operations.

---

Example 1: Customer API with Full Error Handling

# myapp/api/customer.py
import frappe
from frappe import _

@frappe.whitelist()
def get_customer(customer_name):
    """Get customer with proper error handling."""
    if not customer_name:
        frappe.throw(_("Customer name is required"))

    data = frappe.db.get_value(
        "Customer", customer_name,
        ["name", "customer_name", "customer_type", "credit_limit", "disabled"],
        as_dict=True
    )

    if not data:
        frappe.throw(_("Customer '{0}' not found").format(customer_name),
                     exc=frappe.DoesNotExistError)

    if data.disabled:
        frappe.throw(_("Customer '{0}' is disabled").format(data.customer_name))

    return data


@frappe.whitelist()
def create_customer(customer_name, customer_type="Company"):
    """Create customer with duplicate and validation handling."""
    if not customer_name:
        frappe.throw(_("Customer name is required"))

    if frappe.db.exists("Customer", {"customer_name": customer_name}):
        frappe.throw(_("Customer '{0}' already exists").format(customer_name),
                     exc=frappe.DuplicateEntryError)

    try:
        doc = frappe.get_doc({
            "doctype": "Customer",
            "customer_name": customer_name,
            "customer_type": customer_type
        })
        doc.insert()
        return {"success": True, "name": doc.name}

    except frappe.DuplicateEntryError:
        # Race condition — another user created it
        existing = frappe.db.get_value("Customer", {"customer_name": customer_name}, "name")
        frappe.throw(_("Customer was just created by another user"))

    except frappe.MandatoryError as e:
        frappe.throw(_("Missing required field: {0}").format(e))

    except frappe.ValidationError as e:
        frappe.throw(str(e))


@frappe.whitelist()
def update_customer(customer_name, updates):
    """Update customer with concurrent edit handling."""
    if not customer_name:
        frappe.throw(_("Customer name is required"))

    if not frappe.db.exists("Customer", customer_name):
        frappe.throw(_("Customer not found"), exc=frappe.DoesNotExistError)

    try:
        doc = frappe.get_doc("Customer", customer_name)
        if isinstance(updates, str):
            updates = frappe.parse_json(updates)
        doc.update(updates)
        doc.save()
        return {"success": True, "name": doc.name}

    except frappe.TimestampMismatchError:
        frappe.throw(_("Modified by another user. Please refresh."))
    except frappe.CharacterLengthExceededError:
        frappe.throw(_("One or more fields exceed maximum length"))
    except frappe.ValidationError as e:
        frappe.throw(str(e))


@frappe.whitelist()
def delete_customer(customer_name):
    """Delete customer with link checking."""
    if not customer_name:
        frappe.throw(_("Customer name is required"))

    if not frappe.db.exists("Customer", customer_name):
        return {"success": True, "message": "Already deleted"}

    # Pre-check for common linked documents
    linked = frappe.db.count("Sales Invoice", {"customer": customer_name, "docstatus": 1})
    if linked:
        frappe.throw(_("Cannot delete — {0} submitted invoice(s) exist").format(linked))

    try:
        frappe.delete_doc("Customer", customer_name)
        return {"success": True}
    except frappe.LinkExistsError:
        frappe.throw(_("Cannot delete — linked documents exist"))

---

Example 2: Data Import with Error Tracking

# myapp/imports/item_import.py
import frappe
from frappe import _

@frappe.whitelist()
def import_items(items_json):
    """Import items with per-record error tracking."""
    items = frappe.parse_json(items_json)
    if not items:
        frappe.throw(_("No items provided"))

    results = {
        "total": len(items), "created": 0, "updated": 0,
        "failed": 0, "errors": []
    }

    for idx, item_data in enumerate(items, 1):
        code = item_data.get("item_code")
        if not code:
            results["failed"] += 1
            results["errors"].append({"row": idx, "error": "Item code required"})
            continue

        try:
            if frappe.db.exists("Item", code):
                doc = frappe.get_doc("Item", code)
                doc.update(item_data)
                doc.save()
                results["updated"] += 1
            else:
                doc = frappe.get_doc({"doctype": "Item", **item_data})
                doc.insert()
                results["created"] += 1

        except frappe.DuplicateEntryError:
            results["errors"].append({"row": idx, "item": code, "error": "Duplicate"})
        except frappe.MandatoryError as e:
            results["failed"] += 1
            results["errors"].append({"row": idx, "item": code, "error": f"Missing: {e}"})
        except frappe.CharacterLengthExceededError:
            results["failed"] += 1
            results["errors"].append({"row": idx, "item": code, "error": "Field too long"})
        except frappe.ValidationError as e:
            results["failed"] += 1
            results["errors"].append({"row": idx, "item": code, "error": str(e)[:200]})
        except Exception:
            results["failed"] += 1
            frappe.log_error(frappe.get_traceback(), f"Import error: {code}")
            results["errors"].append({"row": idx, "item": code, "error": "Unexpected"})

        if idx % 50 == 0:
            frappe.db.commit()

    frappe.db.commit()
    return results

---

Example 3: Report Query with Error Handling

# myapp/reports/sales_report.py
import frappe
from frappe import _

def execute(filters=None):
    """Sales report with comprehensive query error handling."""
    columns = [
        {"label": _("Invoice"), "fieldname": "name", "fieldtype": "Link",
         "options": "Sales Invoice", "width": 120},
        {"label": _("Customer"), "fieldname": "customer", "fieldtype": "Link",
         "options": "Customer", "width": 150},
        {"label": _("Date"), "fieldname": "posting_date", "fieldtype": "Date", "width": 100},
        {"label": _("Total"), "fieldname": "grand_total", "fieldtype": "Currency", "width": 120},
    ]

    try:
        data = get_report_data(filters)
    except frappe.QueryTimeoutError:
        frappe.throw(_("Report too large. Please narrow your date range."))
    except frappe.db.InternalError:
        frappe.log_error(frappe.get_traceback(), "Sales Report Query Error")
        frappe.throw(_("Database error. Please try again."))

    return columns, data


def get_report_data(filters):
    """Build and execute report query with parameterized SQL."""
    conditions = ["si.docstatus = 1"]
    values = {}

    if filters.get("customer"):
        conditions.append("si.customer = %(customer)s")
        values["customer"] = filters["customer"]

    if filters.get("from_date"):
        conditions.append("si.posting_date >= %(from_date)s")
        values["from_date"] = filters["from_date"]

    if filters.get("to_date"):
        conditions.append("si.posting_date <= %(to_date)s")
        values["to_date"] = filters["to_date"]

    where = " AND ".join(conditions)

    return frappe.db.sql(f"""
        SELECT si.name, si.customer, si.posting_date, si.grand_total
        FROM `tabSales Invoice` si
        WHERE {where}
        ORDER BY si.posting_date DESC
        LIMIT 1000
    """, values, as_dict=True)

---

Example 4: Background Sync with Database Error Recovery

# myapp/tasks/sync.py
import frappe
from frappe import _

def sync_customers():
    """Background sync with connection recovery and per-record error handling."""
    results = {"synced": 0, "failed": 0, "errors": []}

    try:
        customers = frappe.get_all(
            "Customer",
            filters={"sync_status": "Pending"},
            fields=["name", "customer_name"],
            limit=200  # ALWAYS limit
        )

        if not customers:
            frappe.db.commit()
            return

        for customer in customers:
            try:
                external_id = call_external_api(customer)
                frappe.db.set_value("Customer", customer.name, {
                    "sync_status": "Synced",
                    "external_id": external_id
                })
                results["synced"] += 1

            except frappe.db.InternalError as e:
                msg = str(e).lower()
                if "gone away" in msg or "lost connection" in msg:
                    frappe.db.connect()  # Reconnect
                    frappe.log_error("Connection lost during sync", "Sync Recovery")
                else:
                    raise

            except Exception as e:
                results["failed"] += 1
                frappe.db.set_value("Customer", customer.name, {
                    "sync_status": "Failed",
                    "sync_error": str(e)[:500]
                })
                frappe.log_error(frappe.get_traceback(), f"Sync: {customer.name}")

        frappe.db.commit()  # REQUIRED in background job

    except frappe.QueryDeadlockError:
        frappe.db.rollback()
        frappe.log_error("Deadlock during sync", "Sync Deadlock")

    except Exception:
        frappe.db.rollback()
        frappe.log_error(frappe.get_traceback(), "Sync Fatal Error")

    if results["failed"]:
        frappe.log_error(frappe.as_json(results), "Sync Summary")


def call_external_api(customer):
    """Call external API — stub."""
    return "EXT-001"

---

Example 5: Controller with Complete Database Error Handling

# myapp/doctype/custom_order/custom_order.py
import frappe
from frappe import _
from frappe.model.document import Document

class CustomOrder(Document):
    def validate(self):
        self.validate_customer()
        self.validate_items()
        self.calculate_totals()

    def validate_customer(self):
        if not self.customer:
            frappe.throw(_("Customer is required"))

        data = frappe.db.get_value(
            "Customer", self.customer,
            ["customer_name", "disabled", "credit_limit"],
            as_dict=True
        )

        if not data:
            frappe.throw(_("Customer '{0}' not found").format(self.customer))
        if data.disabled:
            frappe.throw(_("Customer is disabled"))

    def validate_items(self):
        if not self.items:
            frappe.throw(_("At least one item is required"))

        errors = []

        # Batch fetch for efficiency — avoids N+1 queries
        codes = [r.item_code for r in self.items if r.item_code]
        existing = {
            d.name: d for d in frappe.get_all(
                "Item",
                filters={"name": ["in", codes]},
                fields=["name", "item_name", "disabled", "is_sales_item"]
            )
        } if codes else {}

        for idx, row in enumerate(self.items, 1):
            if not row.item_code:
                errors.append(_("Row {0}: Item code required").format(idx))
                continue
            item = existing.get(row.item_code)
            if not item:
                errors.append(_("Row {0}: Item '{1}' not found").format(idx, row.item_code))
            elif item.disabled:
                errors.append(_("Row {0}: Item '{1}' disabled").format(idx, item.item_name))

        if errors:
            frappe.throw("<br>".join(errors), title=_("Item Errors"))

    def calculate_totals(self):
        self.total = sum(r.amount or 0 for r in self.items)
        self.grand_total = self.total - (self.discount_amount or 0)

    def on_submit(self):
        try:
            self.create_linked_records()
        except frappe.DuplicateEntryError:
            frappe.throw(_("Linked records already exist"))
        except Exception:
            frappe.log_error(frappe.get_traceback(), f"Submit: {self.name}")
            frappe.throw(_("Error creating linked records"))

    def create_linked_records(self):
        pass

---

Example 6: Transaction Hooks [v15+]

import frappe

def create_with_side_effects(data):
    """Use transaction hooks to coordinate DB changes with external systems."""

    doc = frappe.get_doc({"doctype": "Sales Order", **data})
    doc.insert()

    # File will be created ONLY if DB commit succeeds
    frappe.db.after_commit.add(
        lambda: create_export_file(doc.name)
    )

    # Cleanup file if transaction rolls back
    frappe.db.after_rollback.add(
        lambda: remove_temp_file(doc.name)
    )

    return doc.name


def create_export_file(name):
    """Create export file — only called after successful commit."""
    pass

def remove_temp_file(name):
    """Remove temp file — only called after rollback."""
    pass

---

Quick Reference: Database Error Handling

# Check before get_doc
if frappe.db.exists("Customer", name):
    doc = frappe.get_doc("Customer", name)

# Handle None from get_value
val = frappe.db.get_value("Customer", name, "credit_limit")
val = val or 0  # Default if None

# Safe insert
try:
    doc.insert()
except frappe.DuplicateEntryError:
    pass  # Handle duplicate
except frappe.MandatoryError as e:
    frappe.throw(_("Missing: {0}").format(e))

# Safe save
try:
    doc.save()
except frappe.TimestampMismatchError:
    frappe.throw(_("Modified. Refresh and retry."))

# Safe delete
try:
    frappe.delete_doc("Customer", name)
except frappe.LinkExistsError:
    frappe.throw(_("Linked documents exist"))

# Safe query — ALWAYS use %(name)s format
frappe.db.sql("SELECT * FROM `tabItem` WHERE name = %(n)s", {"n": name})

# Background job — ALWAYS commit
frappe.db.commit()

Error Handling Patterns — Database Operations

Complete error handling patterns for Frappe/ERPNext database operations.

---

Pattern 1: Document CRUD with Full Error Handling

import frappe
from frappe import _

class SafeDocumentManager:
    """Reusable document manager with error handling."""

    @staticmethod
    def get(doctype, name, fields=None):
        """Get document — returns None if not found."""
        if not name:
            return None
        if not frappe.db.exists(doctype, name):
            return None
        if fields:
            return frappe.db.get_value(doctype, name, fields, as_dict=True)
        return frappe.get_doc(doctype, name)

    @staticmethod
    def get_or_throw(doctype, name):
        """Get document or throw user-friendly error."""
        if not name:
            frappe.throw(_("{0} name is required").format(doctype))
        try:
            return frappe.get_doc(doctype, name)
        except frappe.DoesNotExistError:
            frappe.throw(_("{0} '{1}' not found").format(doctype, name))

    @staticmethod
    def create(doctype, data, ignore_duplicates=False):
        """Create document with duplicate handling."""
        try:
            doc = frappe.get_doc({"doctype": doctype, **data})
            doc.insert()
            return {"success": True, "name": doc.name}
        except frappe.DuplicateEntryError:
            if ignore_duplicates:
                existing = frappe.db.get_value(doctype, data, "name")
                if existing:
                    return {"success": True, "name": existing, "existing": True}
            frappe.throw(_("{0} already exists").format(doctype))
        except frappe.MandatoryError as e:
            return {"success": False, "error": f"Missing field: {e}"}
        except frappe.ValidationError as e:
            return {"success": False, "error": str(e)}

    @staticmethod
    def update(doctype, name, updates):
        """Update with concurrent edit handling."""
        if not frappe.db.exists(doctype, name):
            frappe.throw(_("{0} '{1}' not found").format(doctype, name))
        try:
            doc = frappe.get_doc(doctype, name)
            doc.update(updates)
            doc.save()
            return {"success": True}
        except frappe.TimestampMismatchError:
            frappe.throw(_("Modified by another user. Please refresh."))
        except frappe.ValidationError as e:
            return {"success": False, "error": str(e)}

    @staticmethod
    def delete(doctype, name, force=False):
        """Delete with link checking."""
        if not frappe.db.exists(doctype, name):
            return {"success": True, "message": "Already deleted"}
        try:
            frappe.delete_doc(doctype, name, force=force)
            return {"success": True}
        except frappe.LinkExistsError:
            frappe.throw(_("Cannot delete — linked documents exist"))

---

Pattern 2: Safe Get-or-Create (Race Condition Safe)

def get_or_create(doctype, filters, defaults=None):
    """Get existing or create new — handles race conditions."""
    existing = frappe.db.get_value(doctype, filters, "name")
    if existing:
        return frappe.get_doc(doctype, existing)

    doc_data = {"doctype": doctype}
    doc_data.update(filters)
    if defaults:
        doc_data.update(defaults)

    try:
        doc = frappe.get_doc(doc_data)
        doc.insert()
        return doc
    except frappe.DuplicateEntryError:
        # Race condition: another process created it between check and insert
        existing = frappe.db.get_value(doctype, filters, "name")
        if existing:
            return frappe.get_doc(doctype, existing)
        raise  # Unexpected duplicate — re-raise

---

Pattern 3: Batch Operations with Error Isolation

def batch_create(doctype, records, batch_size=100):
    """Create documents in batches with per-record error handling."""
    results = {"created": 0, "duplicates": 0, "failed": 0, "errors": []}

    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        for idx, record in enumerate(batch, i + 1):
            try:
                doc = frappe.get_doc({"doctype": doctype, **record})
                doc.insert()
                results["created"] += 1
            except frappe.DuplicateEntryError:
                results["duplicates"] += 1
            except frappe.MandatoryError as e:
                results["failed"] += 1
                results["errors"].append({"row": idx, "error": f"Missing: {e}"})
            except frappe.ValidationError as e:
                results["failed"] += 1
                results["errors"].append({"row": idx, "error": str(e)[:200]})
            except Exception:
                results["failed"] += 1
                frappe.log_error(frappe.get_traceback(), f"Batch create row {idx}")

        frappe.db.commit()  # Commit per batch

    return results

---

Pattern 4: Safe SQL Query Execution

def safe_query(query, values=None, as_dict=True):
    """Execute SQL with error classification."""
    try:
        return frappe.db.sql(query, values or {}, as_dict=as_dict)
    except frappe.QueryTimeoutError:
        frappe.log_error(f"Timeout: {query[:200]}", "Query Timeout")
        frappe.throw(_("Query too slow. Please add filters."))
    except frappe.QueryDeadlockError:
        frappe.log_error(frappe.get_traceback(), "Deadlock")
        frappe.throw(_("Database busy. Please try again."))
    except frappe.db.InternalError as e:
        msg = str(e).lower()
        if "gone away" in msg or "lost connection" in msg:
            frappe.db.connect()
            return frappe.db.sql(query, values or {}, as_dict=as_dict)
        frappe.log_error(frappe.get_traceback(), "DB Error")
        frappe.throw(_("Database error. Please contact support."))

---

Pattern 5: Query with Retry on Transient Errors

import time

def query_with_retry(func, max_retries=3):
    """Retry on deadlocks and connection errors."""
    for attempt in range(max_retries):
        try:
            return func()
        except frappe.QueryDeadlockError:
            if attempt < max_retries - 1:
                frappe.db.rollback()
                time.sleep(0.5 * (attempt + 1))
            else:
                raise
        except frappe.db.InternalError as e:
            msg = str(e).lower()
            if ("gone away" in msg or "lost connection" in msg) and attempt < max_retries - 1:
                frappe.db.connect()
                time.sleep(0.5)
            else:
                raise

# Usage:
result = query_with_retry(lambda: frappe.get_all("Item", limit=100))

---

Pattern 6: Transaction with Savepoints

def multi_step_operation(data):
    """Multi-step operation with partial rollback."""
    frappe.db.savepoint("step1")
    try:
        parent = frappe.get_doc({"doctype": "Sales Order", **data["parent"]})
        parent.insert()
    except Exception:
        frappe.throw(_("Failed to create order"))

    frappe.db.savepoint("step2")
    try:
        for child in data.get("children", []):
            frappe.get_doc({"doctype": "Task", "order": parent.name, **child}).insert()
    except frappe.ValidationError:
        frappe.db.rollback(save_point="step2")
        # Parent created, children failed — partial success
        return {"parent": parent.name, "partial": True}

    return {"parent": parent.name, "partial": False}

---

Pattern 7: Parameterized SQL (Correct Format)

# ALWAYS use %(name)s with dict — this is the ONLY safe format

# Single parameter
frappe.db.sql(
    "SELECT name FROM `tabItem` WHERE item_code = %(code)s",
    {"code": item_code},
    as_dict=True
)

# Multiple parameters
frappe.db.sql("""
    SELECT name, grand_total
    FROM `tabSales Invoice`
    WHERE customer = %(customer)s
    AND posting_date BETWEEN %(from)s AND %(to)s
    AND docstatus = 1
    ORDER BY posting_date DESC
    LIMIT %(limit)s
""", {
    "customer": customer,
    "from": from_date,
    "to": to_date,
    "limit": 100
}, as_dict=True)

# IN clause — use frappe.db.escape for each value
items = ["ITEM-001", "ITEM-002", "ITEM-003"]
escaped = ", ".join([frappe.db.escape(i) for i in items])
frappe.db.sql(f"""
    SELECT name FROM `tabItem` WHERE name IN ({escaped})
""", as_dict=True)

# Or use query builder for IN clause (safer)
Item = frappe.qb.DocType("Item")
frappe.qb.from_(Item).where(Item.name.isin(items)).run(as_dict=True)

---

Pattern 8: Connection Error Recovery

def with_connection_retry(func):
    """Decorator for automatic reconnection on connection loss."""
    def wrapper(*args, **kwargs):
        max_retries = 3
        for attempt in range(max_retries):
            try:
                return func(*args, **kwargs)
            except frappe.db.InternalError as e:
                msg = str(e).lower()
                is_connection_error = any(x in msg for x in [
                    "gone away", "lost connection", "can't connect", "connection refused"
                ])
                if not is_connection_error:
                    raise
                if attempt < max_retries - 1:
                    frappe.log_error(f"Reconnect attempt {attempt + 1}", "DB Connection")
                    import time
                    time.sleep(1 * (attempt + 1))
                    frappe.db.connect()
                else:
                    frappe.log_error(frappe.get_traceback(), "DB Connection Failed")
                    frappe.throw(_("Database unavailable. Please try again."))
    return wrapper

@with_connection_retry
def reliable_query():
    return frappe.get_all("Sales Invoice", limit=10)

---

Pattern 9: Bulk Update with db.bulk_update [v15+]

# Efficient bulk update — skips ORM, uses direct SQL
updates = {
    "ITEM-001": {"status": "Active", "sync_date": "2025-01-01"},
    "ITEM-002": {"status": "Active", "sync_date": "2025-01-01"},
}

try:
    frappe.db.bulk_update("Item", updates, chunk_size=100)
except frappe.db.InternalError:
    frappe.log_error(frappe.get_traceback(), "Bulk Update Error")
    frappe.throw(_("Bulk update failed"))

# NOTE: bulk_update skips validate/on_update. Use only for mass data fixes.

---

Quick Reference: Database Error Patterns