frappe-errors-database

Frappe 数据库错误诊断与解决

交叉参考：frappe-core-database（API 语法）、frappe-errors-controllers（控制器错误）。

---

错误与修复对照表

错误 / 异常	HTTP	原因	修复
DuplicateEntryError	409	插入或重命名时违反唯一约束	先检查是否已存在，或捕获异常并返回现有记录
DoesNotExistError	404	对不存在的记录调用 get_doc()	先使用 frappe.db.exists()，或捕获异常
LinkValidationError	417	链接字段指向不存在的记录	保存前验证链接目标是否存在
LinkExistsError	N/A	存在关联文档，阻止删除	向用户显示关联文档；谨慎使用 force=True
MandatoryError	417	保存时必填字段为空	插入或保存前设置所有必填字段
TimestampMismatchError	N/A	检测到并发编辑（modified 已变更）	重新加载文档并重试，或告知用户刷新
CharacterLengthExceededError	417	字符串超过字段最大长度或数据库列大小	截断输入，或增加字段长度
DataTooLongException	417	值超过数据库列的存储容量	同 CharacterLengthExceededError
InReadOnlyMode	503	在只读模式下尝试写入	检查 frappe.flags.in_import 或站点配置
QueryTimeoutError	N/A	查询超出时间限制 [v15+]	添加索引、缩小结果集、使用分页
QueryDeadlockError	N/A	两个事务相互等待	采用退避策略重试；缩小事务范围
TooManyWritesError	N/A	单个请求中写入过多	批量操作；使用后台任务
InternalError (gone away)	N/A	MariaDB 连接已断开	使用 frappe.db.connect() 重新连接
InternalError (too many)	N/A	连接池已耗尽	检查 max_connections；关闭空闲连接
ValidationError	417	保存时发生常规验证失败	查看错误消息；修正字段值
SQL 语法错误	N/A	frappe.db.sql() 参数格式错误	使用 %(name)s 和字典，而非 %s 和元组

---

异常层级

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() 参数格式

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

规则：始终使用 %(name)s 和字典参数。绝不要对 SQL 值使用字符串格式化。

---

get_value 返回 None：这不是异常

# ❌ 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"))

各方法的关键行为：

方法	未找到记录	字段为空
get_doc()	引发 DoesNotExistError	返回字段默认值
get_value()	返回 None	返回 None 或 ""
get_all()	返回 []	包含在结果中
exists()	返回 False	N/A
set_value()	静默不执行任何操作	N/A
db.sql()	返回 [] 或 ()	包含在结果中

---

处理各类异常

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

---

事务死锁

# ❌ 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 连接断开 / 连接过多

# 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

预防措施：

• 在 MariaDB 配置中设置 wait_timeout（默认 28800s）
• 检查 max_connections 设置是否符合工作负载
• 在生产环境中使用连接池（Gunicorn 工作进程）

---

事务规则

何时提交

上下文	自动提交？	手动提交？
Web 请求（POST/PUT）	是	绝不
控制器钩子（validate、on_update）	是	绝不
doc_events 钩子	是	绝不
调度器任务	否	始终
后台任务（frappe.enqueue）	否	始终
bench execute	否	始终

使用保存点进行部分回滚

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 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 静默失败

# ❌ 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

---

关键规则

始终

1. 在 frappe.db.sql() 中使用 %(name)s 字典参数——绝不使用字符串格式化
2. 在调用 get_doc() 前检查 frappe.db.exists()——或捕获 DoesNotExistError
3. 在每次调用 insert() 时处理 DuplicateEntryError
4. 在 API 中每次调用 save() 时处理 TimestampMismatchError
5. 在调度器和后台任务中调用 frappe.db.commit()
6. 对大型查询进行分页——使用 limit 参数
7. 使用 get_value() 的结果前，检查其是否为 None
8. 在动态 SQL 字符串中使用 frappe.db.escape()

绝不

1. 使用字符串格式化（f""、.format()、%）处理 SQL 值
2. 在控制器钩子或 doc_events 中调用 frappe.db.commit()
3. 捕获裸 Exception 后直接 pass——记录日志或重新抛出特定类型的异常
4. 假定 db.set_value() 一定成功——记录不存在时它会静默失败
5. 向用户公开原始数据库错误消息——记录详细信息，并显示通用消息
6. 不加 limit 运行无界查询——存在内存或超时风险

---

快速参考：异常处理

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

---

参考文件

文件	内容
references/patterns.md	所有数据库操作的完整错误处理模式
references/examples.md	包含错误处理的完整可运行示例
references/anti-patterns.md	常见错误及错误/正确示例

---

另请参阅

• frappe-core-database — 数据库 API 语法和查询构建器
• frappe-errors-controllers — 控制器错误处理
• frappe-errors-hooks — 钩子错误处理
• frappe-core-permissions — 权限模式