Frappe Core Database

Frappe Database Operations

Quick Reference

"Permissions" = Yes means user permission filters are applied automatically.

---

Decision Tree

What do you need?
│
├─ Create / Update / Delete a document?
│  ├─ With validations + hooks → frappe.get_doc() + .insert()/.save()/.delete()
│  └─ Direct DB (no hooks) → frappe.db.set_value() or doc.db_set()
│
├─ Read a single document?
│  ├─ Need full object with methods → frappe.get_doc()
│  ├─ Read-only, rarely changes → frappe.get_cached_doc()
│  └─ Only need 1-2 fields → frappe.db.get_value()
│
├─ List of documents?
│  ├─ Respect user permissions → frappe.db.get_list()
│  └─ System/admin context → frappe.get_all()
│
├─ Single DocType value?
│  └─ frappe.db.get_single_value('Settings', 'field')
│
├─ Check existence?
│  └─ frappe.db.exists() — NEVER use get_doc in try/except
│
├─ Complex query (JOINs, aggregates)?
│  ├─ Cross-DB compatible → frappe.qb (Query Builder)
│  └─ DB-specific SQL → frappe.db.sql() with parameters
│
└─ DB-specific logic?
   └─ frappe.db.multisql({'mariadb': q1, 'postgres': q2})

RULE: ALWAYS use the highest abstraction level: ORM > Database API > Query Builder > Raw SQL.

---

ORM: Document Operations

Get Document

doc = frappe.get_doc('Sales Invoice', 'SINV-00001')

# Single DocType (no name needed)
settings = frappe.get_doc('System Settings')

# Cached (read-only, for rarely-changing docs)
company = frappe.get_cached_doc('Company', 'My Company')

# Last created
last_task = frappe.get_last_doc('Task', filters={'status': 'Open'})

Create Document

doc = frappe.get_doc({
    'doctype': 'Task',
    'subject': 'Review report',
    'status': 'Open'
})
doc.insert()

# Alternative
doc = frappe.new_doc('Task')
doc.subject = 'Review report'
doc.insert()

Update Document

# Via ORM — triggers validate, on_update, etc.
doc = frappe.get_doc('Task', 'TASK-001')
doc.status = 'Completed'
doc.save()

# Direct DB — SKIPS all validations and hooks
frappe.db.set_value('Task', 'TASK-001', 'status', 'Completed')

# Direct DB on loaded doc
doc.db_set('status', 'Completed')
doc.db_set('status', 'Completed', update_modified=False)
doc.db_set({'status': 'Completed', 'priority': 'High'})

Delete Document

frappe.delete_doc('Task', 'TASK-001')
# Also removes linked Communications, Comments, etc.

Insert Flags

doc.insert(
    ignore_permissions=True,     # Bypass permission check
    ignore_links=True,           # Skip link validation
    ignore_if_duplicate=True,    # No error on duplicate
    ignore_mandatory=True        # Skip required field check
)

RULE: NEVER use multiple ignore flags together unless you have a documented reason. Each flag you add weakens data integrity.

---

Database API: Reading

get_value

# Single field → scalar
status = frappe.db.get_value('Task', 'TASK-001', 'status')

# Multiple fields → tuple
subject, status = frappe.db.get_value('Task', 'TASK-001', ['subject', 'status'])

# As dict
data = frappe.db.get_value('Task', 'TASK-001', ['subject', 'status'], as_dict=True)

# With filters instead of name
status = frappe.db.get_value('Task', {'project': 'PROJ-001'}, 'status')

# Cached (for values that rarely change)
country = frappe.db.get_value('Company', 'MyCompany', 'country', cache=True)

get_single_value

timezone = frappe.db.get_single_value('System Settings', 'time_zone')

get_list / get_all

# get_list — applies user permissions
tasks = frappe.db.get_list('Task',
    filters={'status': 'Open'},
    fields=['name', 'subject', 'assigned_to'],
    order_by='creation desc',
    start=0,
    page_length=50
)

# get_all — NO permission check (same API, different default)
all_tasks = frappe.get_all('Task', filters={'status': 'Open'})

# pluck — returns flat list of single field
names = frappe.get_all('Task', filters={'status': 'Open'}, pluck='name')
# Returns: ['TASK-001', 'TASK-002', ...]

exists / count

exists = frappe.db.exists('User', 'admin@example.com')
exists = frappe.db.exists('User', {'email': 'admin@example.com'})

total = frappe.db.count('Task')
open_count = frappe.db.count('Task', {'status': 'Open'})

---

Filter Operators

{'status': 'Open'}                                    # =
{'status': ['!=', 'Cancelled']}                       # !=
{'amount': ['>', 1000]}                               # >
{'amount': ['>=', 1000]}                              # >=
{'status': ['in', ['Open', 'Working']]}               # IN
{'status': ['not in', ['Cancelled', 'Closed']]}       # NOT IN
{'date': ['between', ['2024-01-01', '2024-12-31']]}   # BETWEEN
{'subject': ['like', '%urgent%']}                      # LIKE
{'description': ['is', 'set']}                         # IS NOT NULL
{'description': ['is', 'not set']}                     # IS NULL

Combining Filters

# AND — all conditions in one dict
filters = {'status': 'Open', 'priority': 'High'}

# AND — list format (allows duplicate fields)
filters = [['status', '=', 'Open'], ['priority', '=', 'High']]

# OR — separate parameter
or_filters = {'priority': 'Urgent', 'status': 'Overdue'}

---

Database API: Writing

set_value

# Single field
frappe.db.set_value('Task', 'TASK-001', 'status', 'Closed')

# Multiple fields
frappe.db.set_value('Task', 'TASK-001', {'status': 'Closed', 'priority': 'Low'})

# Without updating modified timestamp
frappe.db.set_value('Task', 'TASK-001', 'status', 'Closed', update_modified=False)

delete / truncate

# Delete with filters (DML — can be rolled back)
frappe.db.delete('Error Log', {'creation': ['<', '2024-01-01']})

# Truncate (DDL — CANNOT be rolled back)
frappe.db.truncate('Error Log')

bulk_update [v15+]

frappe.db.bulk_update('Task', {
    'TASK-001': {'status': 'Closed'},
    'TASK-002': {'status': 'Closed'}
}, chunk_size=100)

---

Raw SQL: ALWAYS Parameterized

# ✅ CORRECT — parameterized query
results = frappe.db.sql("""
    SELECT name, subject FROM `tabTask`
    WHERE status = %(status)s AND owner = %(owner)s
""", {'status': 'Open', 'owner': frappe.session.user}, as_dict=True)

CRITICAL: NEVER use f-strings, % formatting, or string concatenation in SQL. See SQL Injection Prevention.

Return Types

frappe.db.sql(query)                    # Tuple of tuples (default)
frappe.db.sql(query, as_dict=True)      # List of dicts
frappe.db.sql(query, as_list=True)      # List of lists

Table Naming

ALWAYS use backtick-quoted tab prefix: ` tabSales Invoice , tabTask `

Database-Specific SQL

frappe.db.multisql({
    'mariadb': "SELECT IFNULL(field, 0) FROM `tabDoc`",
    'postgres': "SELECT COALESCE(field, 0) FROM `tabDoc`"
})

---

Query Builder (frappe.qb) [v14+]

The Query Builder uses PyPika under the hood. It generates parameterized SQL automatically.

Task = frappe.qb.DocType('Task')

results = (
    frappe.qb.from_(Task)
    .select(Task.name, Task.subject, Task.status)
    .where(Task.status == 'Open')
    .orderby(Task.creation, order='desc')
    .limit(10)
).run(as_dict=True)

JOINs

SI = frappe.qb.DocType('Sales Invoice')
Customer = frappe.qb.DocType('Customer')

results = (
    frappe.qb.from_(SI)
    .inner_join(Customer).on(SI.customer == Customer.name)
    .select(SI.name, SI.grand_total, Customer.customer_name)
    .where(SI.docstatus == 1)
).run(as_dict=True)

Aggregates

from frappe.query_builder.functions import Count, Sum, Avg

stats = (
    frappe.qb.from_(Task)
    .select(Task.status, Count(Task.name).as_('count'))
    .groupby(Task.status)
).run(as_dict=True)

OR Conditions

customers = frappe.qb.DocType('Customer')
results = (
    frappe.qb.from_(customers)
    .select(customers.name)
    .where(
        (customers.territory == 'US') | (customers.territory == 'UK')
    )
).run(as_dict=True)

Inspect Generated SQL

query = frappe.qb.from_(Task).select('*').where(Task.name == 'X')
sql, params = query.walk()   # Returns (sql_string, param_dict)
sql_str = query.get_sql()    # Returns SQL string

See references/query-patterns.md for subqueries, ImportMapper, ConstantColumn, and custom functions.

---

Caching

Document Cache

doc = frappe.get_cached_doc('Company', 'My Company')   # Full document
val = frappe.db.get_value('Company', 'X', 'country', cache=True)  # Single value

Redis Cache

frappe.cache.set_value('key', data, expires_in_sec=3600)
data = frappe.cache.get_value('key')
frappe.cache.delete_value('key')

@redis_cache Decorator

from frappe.utils.caching import redis_cache

@redis_cache(ttl=300)
def get_dashboard_data(user):
    return expensive_calculation(user)

# Invalidate
get_dashboard_data.clear_cache()

See references/caching-patterns.md for hash operations, invalidation strategies, and best practices.

---

Transaction Management

The framework manages transactions automatically:

Manual Transactions (rarely needed)

frappe.db.savepoint('before_payment')
try:
    # operations...
    frappe.db.commit()
except Exception:
    frappe.db.rollback(save_point='before_payment')

Transaction Hooks [v15+]

frappe.db.after_commit.add(sync_to_external_system)
frappe.db.after_rollback.add(cleanup_external_state)

---

SQL Injection Prevention

CRITICAL SECURITY RULE: NEVER interpolate user input into SQL strings.

# ❌ VULNERABLE — SQL injection risk
frappe.db.sql(f"SELECT * FROM `tabUser` WHERE name = '{user_input}'")
frappe.db.sql("SELECT * FROM `tabUser` WHERE name = '%s'" % user_input)
frappe.db.sql("SELECT * FROM `tabUser` WHERE name = " + user_input)

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

# ✅ SAFE — ORM / Query Builder (always parameterized)
frappe.get_all('User', filters={'name': user_input})

User = frappe.qb.DocType('User')
frappe.qb.from_(User).select('*').where(User.name == user_input).run()

RULE: When you MUST use frappe.db.sql(), ALWAYS use %(param)s placeholders with a dict. The Query Builder (frappe.qb) is ALWAYS preferred over raw SQL for new code.

---

Version Differences

v16 Breaking Changes

# v14/v15 — string aggregates
fields=['count(name) as count']

# v16 — dict aggregates
fields=[{'COUNT': 'name', 'as': 'count'}]

# v14/v15 — run=False returns SQL string
sql = frappe.db.get_list('Task', run=False)

# v16 — run=False returns Query Builder object
qb_obj = frappe.db.get_list('Task', run=False)
sql = qb_obj.get_sql()

---

Critical Rules Summary

1. NEVER use string formatting in SQL — ALWAYS use parameterized queries 2. NEVER call frappe.db.commit() inside controller hooks (validate, on_update, etc.) 3. ALWAYS paginate list queries — use page_length parameter 4. ALWAYS specify fields — NEVER use fields=['*'] in production 5. ALWAYS use frappe.db.exists() for existence checks — NEVER try/except with get_doc 6. ALWAYS prefix table names with tab in raw SQL: ` tabSales Invoice 7. **NEVER** use multiple ignore flags without documented justification 8. **ALWAYS** use batch fetching to avoid N+1 queries 9. **ALWAYS** prefer frappe.qb over frappe.db.sql() for new code 10. **NEVER** use frappe.db.truncate()` without understanding it CANNOT be rolled back

---

Query Builder: Dedicated Skill

For complex queries (joins, aggregations, subqueries, cross-DB compatibility), see [frappe-syntax-query-builder](../../syntax/frappe-syntax-query-builder/SKILL.md).

• frappe.db methods (this skill) — Simple CRUD, get_value, get_list, exists checks
• frappe.qb (query-builder skill) — Joins, GROUP BY, HAVING, subqueries, cross-DB functions
• frappe.db.sql — Very complex SQL not expressible in qb (ALWAYS parameterized)

Reference Files

• [methods-reference.md](references/methods-reference.md) — Complete API signatures for all database and document methods
• [query-patterns.md](references/query-patterns.md) — Query Builder patterns, subqueries, ImportMapper, custom functions
• [caching-patterns.md](references/caching-patterns.md) — Redis cache, @redis_cache, hash operations, invalidation
• [examples.md](references/examples.md) — Real-world patterns: CRUD, reports, batch processing, transactions
• [anti-patterns.md](references/anti-patterns.md) — SQL injection, N+1, commit mistakes, and 10 more anti-patterns

Database Anti-Patterns

Common database mistakes in Frappe development and their corrections. Every anti-pattern includes a CORRECT alternative.

---

1. SQL Injection — CRITICAL SECURITY VULNERABILITY

NEVER: String Formatting in SQL

# ❌ ALL of these are SQL injection vulnerabilities
user_input = "admin'; DROP TABLE tabUser; --"

frappe.db.sql(f"SELECT * FROM `tabUser` WHERE name = '{user_input}'")
frappe.db.sql("SELECT * FROM `tabUser` WHERE name = '%s'" % user_input)
frappe.db.sql("SELECT * FROM `tabUser` WHERE name = " + user_input)

ALWAYS: Parameterized Queries

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

# ✅ SAFE — ORM (always parameterized internally)
frappe.get_all('User', filters={'name': user_input})

# ✅ SAFE — Query Builder (always parameterized)
User = frappe.qb.DocType('User')
frappe.qb.from_(User).select('*').where(User.name == user_input).run()

RULE: EVERY frappe.db.sql() call MUST use %(param)s placeholders with a dict. No exceptions.

---

2. N+1 Query Problem

NEVER: Query Inside a Loop

# ❌ N+1 queries — fetches one document per iteration
def get_order_details(order_names):
    results = []
    for name in order_names:
        order = frappe.get_doc('Sales Order', name)        # N queries
        customer = frappe.get_doc('Customer', order.customer)  # N more queries
        results.append({'order': order.name, 'customer': customer.customer_name})
    return results

ALWAYS: Batch Fetch

# ✅ Two queries total, regardless of list size
def get_order_details(order_names):
    orders = frappe.get_all('Sales Order',
        filters={'name': ['in', order_names]},
        fields=['name', 'customer', 'grand_total']
    )
    customer_names = list(set(o.customer for o in orders))
    customers = {c.name: c for c in frappe.get_all('Customer',
        filters={'name': ['in', customer_names]},
        fields=['name', 'customer_name']
    )}
    return [{'order': o.name, 'customer': customers[o.customer].customer_name} for o in orders]

---

3. Commit Inside Controller Hooks

NEVER: Manual Commit in Hooks

# ❌ Breaks framework transaction management
class SalesInvoice(Document):
    def validate(self):
        self.calculate_totals()
        frappe.db.commit()       # NEVER do this

    def on_submit(self):
        self.create_gl_entries()
        frappe.db.commit()       # NEVER do this

ALWAYS: Let the Framework Handle Commits

# ✅ Framework commits after successful request
class SalesInvoice(Document):
    def validate(self):
        self.calculate_totals()
        # No commit — framework handles it

    def on_submit(self):
        self.create_gl_entries()
        # No commit — framework handles it

Exception: frappe.db.commit() is acceptable ONLY in background jobs or standalone scripts where you explicitly manage transactions.

---

4. SELECT * in Production

NEVER: Fetch All Fields

# ❌ Fetches ALL columns including large TEXT/LONGTEXT fields
docs = frappe.get_all('Sales Invoice', fields=['*'])
frappe.db.sql("SELECT * FROM `tabSales Invoice`")

ALWAYS: Specify Needed Fields

# ✅ Fetch only what you need
docs = frappe.get_all('Sales Invoice',
    fields=['name', 'customer', 'grand_total', 'status']
)

---

5. No Pagination

NEVER: Fetch Unlimited Records

# ❌ Can return millions of records — crashes server
all_logs = frappe.get_all('Error Log')
frappe.db.sql("SELECT * FROM `tabError Log`")

ALWAYS: Limit Results

# ✅ Always use page_length
logs = frappe.get_all('Error Log',
    fields=['name', 'error', 'creation'],
    order_by='creation desc',
    page_length=100
)

# ✅ Always use LIMIT in SQL
frappe.db.sql("""
    SELECT name, error FROM `tabError Log`
    ORDER BY creation DESC LIMIT 100
""")

---

6. Overusing ignore Flags

NEVER: Ignore Everything

# ❌ Bypasses all safety checks — data integrity at risk
doc.insert(
    ignore_permissions=True,
    ignore_mandatory=True,
    ignore_links=True
)

ALWAYS: Use Minimum Required Flags

# ✅ Only bypass what you must, with documented reason
doc.flags.ignore_permissions = True  # Reason: system background job
doc.insert()

---

7. Using db_set/set_value for Business Logic

NEVER: Direct DB Update for Stateful Changes

# ❌ Skips validate, on_update, and all other hooks
def update_status(name, status):
    frappe.db.set_value('Task', name, 'status', status)

ALWAYS: Use ORM for Business Logic

# ✅ Triggers validate, on_update, permission checks
def update_status(name, status):
    doc = frappe.get_doc('Task', name)
    doc.status = status
    doc.save()

`db_set`/`set_value` is acceptable ONLY for: hidden fields, counters, timestamps, performance-critical background jobs.

---

8. No Error Handling in Batch Operations

NEVER: Bare Loop Without Error Handling

# ❌ One failure kills the entire batch
def process_data(items):
    for item in items:
        doc = frappe.get_doc('Item', item)
        doc.status = 'Processed'
        doc.save()
    frappe.db.commit()

ALWAYS: Handle Errors Per Item

# ✅ Process what you can, log what fails
def process_data(items):
    processed, errors = [], []

    for item in items:
        try:
            doc = frappe.get_doc('Item', item)
            doc.status = 'Processed'
            doc.save()
            processed.append(item)
        except frappe.DoesNotExistError:
            errors.append({'item': item, 'error': 'Not found'})
        except Exception as e:
            frappe.log_error(frappe.get_traceback(), f'Process Error: {item}')
            errors.append({'item': item, 'error': str(e)})

    return {'processed': len(processed), 'errors': errors}

---

9. Cache Without Invalidation

NEVER: Unbounded Cache

# ❌ Cache never cleared — returns stale data forever
@redis_cache
def get_settings():
    return frappe.get_doc('My Settings')

ALWAYS: Set TTL or Explicit Invalidation

# ✅ TTL ensures refresh
@redis_cache(ttl=3600)
def get_settings():
    return frappe.get_doc('My Settings')

# ✅ Explicit invalidation on change
class MySettings(Document):
    def on_update(self):
        get_settings.clear_cache()

---

10. Blocking Operations in Web Requests

NEVER: Long-Running Code in Request Handler

# ❌ Request times out — user sees error
@frappe.whitelist()
def process_all_invoices():
    invoices = frappe.get_all('Sales Invoice', filters={'status': 'Unpaid'})
    for inv in invoices:
        send_reminder_email(inv.name)  # Could take minutes
    return "Done"

ALWAYS: Use Background Jobs for Heavy Work

# ✅ Returns immediately, processes in background
@frappe.whitelist()
def process_all_invoices():
    frappe.enqueue(
        'myapp.tasks.send_reminders',
        queue='long',
        timeout=3600
    )
    return "Processing started"

def send_reminders():
    invoices = frappe.get_all('Sales Invoice', filters={'status': 'Unpaid'})
    for inv in invoices:
        send_reminder_email(inv.name)
        frappe.db.commit()  # Commit per iteration in background jobs

---

11. Using get_doc for Existence Checks

NEVER: Try/Except with get_doc

# ❌ Loads entire document just to check existence
try:
    doc = frappe.get_doc('User', email)
    exists = True
except:
    exists = False

ALWAYS: Use frappe.db.exists()

# ✅ Lightweight query, returns True/False
exists = frappe.db.exists('User', email)
exists = frappe.db.exists('User', {'email': email})

---

12. Wrong Table Names in Raw SQL

NEVER: Missing tab Prefix

# ❌ Table not found — Frappe prefixes all tables with `tab`
frappe.db.sql("SELECT * FROM Task")
frappe.db.sql("SELECT * FROM sales_invoice")

ALWAYS: Use Backtick-Quoted tab Prefix

# ✅ Correct table naming
frappe.db.sql("SELECT * FROM `tabTask`")
frappe.db.sql("SELECT * FROM `tabSales Invoice`")

Format: ` tab{Exact DocType Name} ` — including spaces, exact capitalization.

---

13. Using truncate Without Understanding Consequences

NEVER: Truncate Without Awareness

# ❌ DDL operation — auto-commits, CANNOT be rolled back
frappe.db.truncate('Important Table')

ALWAYS: Use delete() for Rollback-Safe Deletion

# ✅ DML operation — can be rolled back
frappe.db.delete('Error Log', {'creation': ['<', '2024-01-01']})

Use truncate ONLY for log/temp tables where rollback is not needed and performance matters.

Caching Patterns Reference

Redis cache, @redis_cache decorator, document caching, and invalidation strategies. Verified against Frappe v14-v16 docs.

---

Document Caching

frappe.get_cached_doc(doctype, name)

Returns cached Document object. Falls back to database if not in cache.

# ALWAYS use for read-only access to rarely-changing documents
company = frappe.get_cached_doc('Company', 'My Company')
settings = frappe.get_cached_doc('Selling Settings')

# NEVER use when:
# - You need to modify the document
# - You need guaranteed up-to-date data
# - You just saved changes to this document

frappe.db.get_value with cache=True

Caches single field lookups.

country = frappe.db.get_value('Company', 'My Company', 'country', cache=True)

---

Redis Cache — Basic Operations

Set / Get / Delete

# Simple value
frappe.cache.set_value('my_key', 'my_value')
value = frappe.cache.get_value('my_key')

# Dict or list
frappe.cache.set_value('user_data', {'name': 'Admin', 'role': 'System Manager'})
data = frappe.cache.get_value('user_data')

# With expiry (seconds)
frappe.cache.set_value('temp_key', 'value', expires_in_sec=3600)   # 1 hour
frappe.cache.set_value('short_lived', 'value', expires_in_sec=300) # 5 min

# Delete
frappe.cache.delete_value('my_key')

Site-Specific Keys

Frappe automatically prefixes all cache keys with the site name:

# Site: site1.example.com
frappe.cache.set_value('key', 'value')
# Actual Redis key: site1.example.com|key

This means the same key on different sites stores separate values.

Local Request Cache

Within a single request, repeated get_value calls return from in-memory cache without hitting Redis:

frappe.cache.get_value('key')  # Redis call
frappe.cache.get_value('key')  # In-memory (no Redis call)

---

Hash Operations

Use for complex objects where individual fields need independent updates.

# Set individual fields
frappe.cache.hset('user|admin', 'name', 'Administrator')
frappe.cache.hset('user|admin', 'email', 'admin@example.com')

# Get single field
name = frappe.cache.hget('user|admin', 'name')

# Get all fields
user_data = frappe.cache.hgetall('user|admin')
# {'name': 'Administrator', 'email': 'admin@example.com'}

# Delete field
frappe.cache.hdel('user|admin', 'name')

---

@redis_cache Decorator

Basic Usage

from frappe.utils.caching import redis_cache

@redis_cache
def expensive_calculation(param1, param2):
    # Heavy computation
    return param1 + param2

# First call: executes function
result = expensive_calculation(10, 20)

# Subsequent calls with same args: returns from cache
result = expensive_calculation(10, 20)

With TTL (Time To Live)

@redis_cache(ttl=300)    # 5 minutes
def get_dashboard_data(user):
    return calculate_dashboard(user)

@redis_cache(ttl=3600)   # 1 hour
def get_monthly_report(month, year):
    return generate_report(month, year)

Cache Invalidation

@redis_cache(ttl=300)
def get_user_stats(user):
    return calculate_stats(user)

# Manually clear cache
get_user_stats.clear_cache()

---

TTL Guidelines

RULE: ALWAYS set a TTL unless you have explicit invalidation logic. Unbounded caches cause stale data bugs.

---

Cache Invalidation Patterns

Pattern 1: Invalidate on Document Update

@redis_cache(ttl=3600)
def get_company_settings(company):
    return frappe.get_doc('Company', company)

class Company(Document):
    def on_update(self):
        get_company_settings.clear_cache()

Pattern 2: Key-Based Invalidation

def get_dashboard_data(user):
    cache_key = f"dashboard_{user}"
    data = frappe.cache.get_value(cache_key)
    if data:
        return data

    data = compute_dashboard(user)
    frappe.cache.set_value(cache_key, data, expires_in_sec=300)
    return data

class SalesInvoice(Document):
    def on_submit(self):
        # Invalidate dashboard for the invoice owner
        frappe.cache.delete_value(f"dashboard_{self.owner}")

Pattern 3: Bulk Cache with Hash

def cache_all_companies():
    companies = frappe.get_all('Company', fields=['name', 'country', 'default_currency'])
    for company in companies:
        frappe.cache.hset('companies', company.name, company)

def get_company_cached(name):
    data = frappe.cache.hget('companies', name)
    if not data:
        data = frappe.db.get_value('Company', name, ['name', 'country', 'default_currency'], as_dict=True)
        frappe.cache.hset('companies', name, data)
    return data

Pattern 4: Graceful Degradation

def get_data_with_fallback(key):
    try:
        data = frappe.cache.get_value(key)
        if data:
            return data
    except Exception:
        pass  # Redis down — fall back to database

    return fetch_from_database(key)

---

Anti-Patterns

NEVER: Cache Without Invalidation

# ❌ Cache never cleared — stale data guaranteed
@redis_cache
def get_settings():
    return frappe.get_doc('My Settings')

NEVER: Generic Cache Keys

# ❌ Collisions and confusion
frappe.cache.set_value('data', result)

# ✅ Specific, namespaced keys
frappe.cache.set_value(f"sales_report_{user}_{month}", result)

NEVER: Cache Large Objects Unnecessarily

# ❌ Caching full document with all children
frappe.cache.set_value('invoice', frappe.get_doc('Sales Invoice', 'SINV-001'))

# ✅ Cache only what you need
frappe.cache.set_value('invoice_total', {'name': 'SINV-001', 'total': 50000})

Database Examples

Real-world patterns for common Frappe database operations. All examples use parameterized queries and follow best practices.

---

Example 1: Document CRUD

Create

doc = frappe.get_doc({
    'doctype': 'Task',
    'subject': 'Review Sales Report',
    'status': 'Open',
    'priority': 'High',
    'description': 'Monthly sales review',
    'exp_start_date': frappe.utils.today(),
    'expected_time': 2
})
doc.insert()
# Framework auto-commits after request completes

Read

# Full document
doc = frappe.get_doc('Task', 'TASK-001')

# Specific fields only
subject, status = frappe.db.get_value('Task', 'TASK-001', ['subject', 'status'])

# Cached document (read-only, for stable data)
company = frappe.get_cached_doc('Company', 'My Company')

Update

# Via ORM — triggers validate, on_update hooks
doc = frappe.get_doc('Task', 'TASK-001')
doc.status = 'Working'
doc.save()

# Direct database — skips ALL validations
frappe.db.set_value('Task', 'TASK-001', 'status', 'Completed')

Delete

frappe.delete_doc('Task', 'TASK-001')

---

Example 2: Filtered Lists with Pagination

def get_open_tasks(page=0, page_size=50):
    """Fetch open tasks with pagination. ALWAYS paginate."""
    return frappe.get_all('Task',
        filters={
            'status': 'Open',
            'priority': ['in', ['High', 'Urgent']]
        },
        fields=['name', 'subject', 'assigned_to', 'exp_end_date'],
        order_by='exp_end_date asc',
        start=page * page_size,
        page_length=page_size
    )

With OR Filters

urgent_tasks = frappe.get_all('Task',
    filters={'docstatus': 0},
    or_filters={
        'priority': 'Urgent',
        'exp_end_date': ['<', frappe.utils.today()]
    },
    fields=['name', 'subject', 'priority', 'exp_end_date']
)

Iterate All Records in Batches

def process_all_invoices():
    """Process all submitted invoices in batches."""
    page = 0
    page_size = 100

    while True:
        batch = frappe.get_all('Sales Invoice',
            filters={'docstatus': 1},
            fields=['name', 'customer', 'grand_total'],
            start=page * page_size,
            page_length=page_size,
            order_by='creation asc'
        )
        if not batch:
            break

        for inv in batch:
            process_invoice(inv)

        page += 1

---

Example 3: Aggregation with Query Builder

from frappe.query_builder.functions import Count, Sum

Task = frappe.qb.DocType('Task')

stats = (
    frappe.qb.from_(Task)
    .select(
        Task.status,
        Count(Task.name).as_('count'),
        Sum(Task.expected_time).as_('total_hours')
    )
    .where(Task.docstatus == 0)
    .groupby(Task.status)
    .orderby(Count(Task.name), order='desc')
).run(as_dict=True)

for stat in stats:
    print(f"{stat.status}: {stat.count} tasks, {stat.total_hours}h total")

---

Example 4: JOIN — Sales Report

With Query Builder (preferred)

from frappe.query_builder.functions import Sum, Count

SI = frappe.qb.DocType('Sales Invoice')
Customer = frappe.qb.DocType('Customer')

report = (
    frappe.qb.from_(SI)
    .inner_join(Customer).on(SI.customer == Customer.name)
    .select(
        Customer.customer_name,
        Customer.territory,
        Sum(SI.grand_total).as_('total_sales'),
        Count(SI.name).as_('invoice_count')
    )
    .where(SI.docstatus == 1)
    .where(SI.posting_date >= '2024-01-01')
    .groupby(Customer.name)
    .orderby(Sum(SI.grand_total), order='desc')
    .limit(10)
).run(as_dict=True)

With Raw SQL (parameterized)

results = frappe.db.sql("""
    SELECT
        c.customer_name,
        c.territory,
        SUM(si.grand_total) as total_sales,
        COUNT(si.name) as invoice_count
    FROM `tabSales Invoice` si
    INNER JOIN `tabCustomer` c ON si.customer = c.name
    WHERE si.docstatus = 1
    AND si.posting_date >= %(from_date)s
    GROUP BY c.name
    ORDER BY total_sales DESC
    LIMIT 10
""", {'from_date': '2024-01-01'}, as_dict=True)

---

Example 5: Batch Processing — Avoid N+1

def get_order_details(order_names):
    """Fetch order details with batch queries — NOT one query per order."""

    # One query for all orders
    orders = frappe.get_all('Sales Order',
        filters={'name': ['in', order_names]},
        fields=['name', 'customer', 'grand_total']
    )

    # One query for all customers
    customer_names = list(set(o.customer for o in orders))
    customers = {c.name: c for c in frappe.get_all('Customer',
        filters={'name': ['in', customer_names]},
        fields=['name', 'customer_name', 'territory']
    )}

    return [{
        'order': o.name,
        'total': o.grand_total,
        'customer': customers.get(o.customer, {}).get('customer_name', 'Unknown')
    } for o in orders]

Bulk Update [v15+]

def close_old_tasks():
    """Close all tasks older than 30 days."""
    old_tasks = frappe.get_all('Task',
        filters={
            'status': 'Open',
            'creation': ['<', frappe.utils.add_days(frappe.utils.today(), -30)]
        },
        pluck='name'
    )

    if old_tasks:
        updates = {name: {'status': 'Closed'} for name in old_tasks}
        frappe.db.bulk_update('Task', updates, chunk_size=100)
        frappe.db.commit()

---

Example 6: Transaction with Savepoint

def process_payment(invoice_name, amount):
    """Process payment with rollback on failure."""
    frappe.db.savepoint('before_payment')

    try:
        invoice = frappe.get_doc('Sales Invoice', invoice_name)

        pe = frappe.get_doc({
            'doctype': 'Payment Entry',
            'payment_type': 'Receive',
            'party_type': 'Customer',
            'party': invoice.customer,
            'paid_amount': amount,
            'received_amount': amount,
            'references': [{
                'reference_doctype': 'Sales Invoice',
                'reference_name': invoice_name,
                'allocated_amount': amount
            }]
        })
        pe.insert()
        pe.submit()

        return {'success': True, 'payment': pe.name}

    except Exception as e:
        frappe.db.rollback(save_point='before_payment')
        frappe.log_error(frappe.get_traceback(), f'Payment Error: {invoice_name}')
        return {'success': False, 'error': str(e)}

---

Example 7: Cached Dashboard

from frappe.utils.caching import redis_cache

@redis_cache(ttl=300)
def get_sales_dashboard(user):
    """Cached sales dashboard — refreshes every 5 minutes."""
    today = frappe.utils.today()
    month_start = frappe.utils.get_first_day(today)

    this_month = frappe.db.sql("""
        SELECT COALESCE(SUM(grand_total), 0) as total
        FROM `tabSales Invoice`
        WHERE docstatus = 1 AND posting_date >= %(month_start)s
    """, {'month_start': month_start}, as_dict=True)[0].total

    open_orders = frappe.db.count('Sales Order', {
        'docstatus': 1,
        'status': ['in', ['To Deliver and Bill', 'To Bill', 'To Deliver']]
    })

    return {
        'this_month_sales': this_month,
        'open_orders': open_orders
    }

# Invalidate when new invoice is submitted
class SalesInvoice(Document):
    def on_submit(self):
        get_sales_dashboard.clear_cache()

---

Example 8: Background Job for Heavy Operations

@frappe.whitelist()
def process_all_invoices():
    """Queue heavy processing as background job — NEVER block the request."""
    frappe.enqueue(
        'myapp.tasks.process_invoices_bg',
        queue='long',
        timeout=3600
    )
    return {'message': 'Processing started'}

def process_invoices_bg():
    """Background job: commit per batch to avoid losing progress."""
    invoices = frappe.get_all('Sales Invoice',
        filters={'status': 'Unpaid', 'docstatus': 1},
        fields=['name', 'customer'],
        page_length=0  # All records (OK in background job)
    )

    for i, inv in enumerate(invoices):
        try:
            send_reminder_email(inv.name)
        except Exception:
            frappe.log_error(frappe.get_traceback(), f'Reminder Error: {inv.name}')

        # Commit every 100 records in background jobs
        if (i + 1) % 100 == 0:
            frappe.db.commit()

    frappe.db.commit()

---

Example 9: Existence Check and Conditional Create

def ensure_task_exists(subject, project):
    """Create task only if it does not exist. ALWAYS use exists(), not try/except."""
    if not frappe.db.exists('Task', {'subject': subject, 'project': project}):
        doc = frappe.get_doc({
            'doctype': 'Task',
            'subject': subject,
            'project': project,
            'status': 'Open'
        })
        doc.insert(ignore_permissions=True)
        return doc.name

    return frappe.db.get_value('Task', {'subject': subject, 'project': project}, 'name')

---

Example 10: Permission-Aware Data Access

def get_user_invoices(user=None):
    """Fetch invoices respecting user permissions."""
    user = user or frappe.session.user

    if not frappe.has_permission('Sales Invoice', 'read', user=user):
        frappe.throw('No read permission for Sales Invoice', frappe.PermissionError)

    # get_list automatically applies User Permissions
    return frappe.db.get_list('Sales Invoice',
        fields=['name', 'customer', 'grand_total', 'status'],
        order_by='posting_date desc',
        page_length=100
    )

Database & Document Methods Reference

Complete API reference for frappe.db.* and Document methods. Verified against Frappe v14-v16 official docs.

---

Document API (Global Functions)

frappe.get_doc(doctype, name)

Returns a Document object. Raises DoesNotExistError if not found.

doc = frappe.get_doc('Sales Invoice', 'SINV-00001')

# Single DocType — no name needed
settings = frappe.get_doc('System Settings')

# Create from dict (in-memory, not saved)
doc = frappe.get_doc({'doctype': 'Task', 'subject': 'New task'})

frappe.get_cached_doc(doctype, name)

Same as get_doc but checks Redis cache first. ALWAYS use for read-only access to rarely-changing documents.

company = frappe.get_cached_doc('Company', 'My Company')

frappe.new_doc(doctype)

Creates a new Document with defaults applied.

doc = frappe.new_doc('Task')
doc.subject = 'Review report'
doc.insert()

frappe.get_last_doc(doctype, filters=None, order_by='creation desc')

Returns the most recently created document matching filters.

last_task = frappe.get_last_doc('Task')
last_open = frappe.get_last_doc('Task', filters={'status': 'Open'})

frappe.delete_doc(doctype, name)

Deletes document and its children, linked Communications, Comments, etc.

frappe.delete_doc('Task', 'TASK-001')

frappe.rename_doc(doctype, old_name, new_name, merge=False)

Renames document primary key. Requires "Allow Rename" on DocType.

frappe.rename_doc('Task', 'OLD-NAME', 'NEW-NAME')
frappe.rename_doc('Customer', 'Old Co', 'New Co', merge=True)  # Merge if exists

frappe.get_meta(doctype)

Returns DocType metadata with custom fields and property setters applied.

meta = frappe.get_meta('Sales Invoice')
fields = meta.fields

---

Document Instance Methods

doc.insert()

Inserts new document into database. Runs validate hooks.

doc.insert()
doc.insert(ignore_permissions=True)
doc.insert(ignore_links=True)
doc.insert(ignore_if_duplicate=True)
doc.insert(ignore_mandatory=True)

doc.save()

Saves changes to existing document. Runs validate and on_update hooks.

doc.save()
doc.save(ignore_permissions=True)
doc.save(ignore_version=True)  # No version record

doc.delete()

Deletes the document. Alias to frappe.delete_doc.

doc.submit()

Submits document (sets docstatus=1). Only for submittable DocTypes.

doc.cancel()

Cancels submitted document (sets docstatus=2).

doc.amend()

Creates amendment copy of cancelled document.

doc.db_set(field, value, **kwargs)

Sets field value directly in database. SKIPS all validations and hooks.

doc.db_set('status', 'Closed')
doc.db_set({'status': 'Closed', 'priority': 'High'})
doc.db_set('status', 'Closed', update_modified=False)
doc.db_set('status', 'Closed', commit=True)
doc.db_set('status', 'Closed', notify=True)  # Triggers realtime update

doc.reload()

Refreshes document with latest database values.

doc.reload()

doc.get_doc_before_save()

Returns document state before current save operation. Use in validate/on_update.

old_doc = doc.get_doc_before_save()

doc.has_value_changed(fieldname)

Returns True if field value changed during current save.

if doc.has_value_changed('status'):
    notify_status_change(doc)

doc.check_permission(permtype)

Throws frappe.PermissionError if user lacks specified permission.

doc.check_permission('write')

doc.append(child_table_field, values)

Appends row to child table.

doc.append('items', {'item_code': 'ITEM-001', 'qty': 10})

doc.get_url()

Returns Desk URL for the document.

url = doc.get_url()  # e.g., '/app/task/TASK-001'

doc.add_comment(comment_type, text)

Adds comment visible in document timeline.

doc.add_tag(tag)

Adds tag for filtering/grouping.

doc.get_tags()

Returns list of document tags.

doc.run_method(method_name)

Executes controller method with hooks.

doc.queue_action(method, **kwargs)

Runs controller method in background via job queue.

doc.db_insert() / doc.db_update()

Low-level insert/update bypassing all validations. NEVER use in application code — use doc.insert() and doc.save() instead.

---

Database API (frappe.db.*)

frappe.db.get_list(doctype, **kwargs)

Returns list of records with user permissions applied.

tasks = frappe.db.get_list('Task',
    filters={'status': 'Open'},
    or_filters={'priority': 'Urgent'},
    fields=['name', 'subject'],
    order_by='creation desc',
    group_by='status',
    start=0,
    page_length=20
)

frappe.get_all(doctype, **kwargs)

Same as get_list but with ignore_permissions=True. Use for system/admin context.

all_tasks = frappe.get_all('Task', filters={'status': 'Open'}, pluck='name')

frappe.db.get_value(doctype, name_or_filters, fieldname, **kwargs)

Returns field value(s) from a single record.

# Single field → scalar
status = frappe.db.get_value('Task', 'TASK-001', 'status')

# Multiple fields → tuple
subject, status = frappe.db.get_value('Task', 'TASK-001', ['subject', 'status'])

# As dict
data = frappe.db.get_value('Task', 'TASK-001', ['subject', 'status'], as_dict=True)

# With filters
status = frappe.db.get_value('Task', {'project': 'PROJ-001'}, 'status')

# Cached
country = frappe.db.get_value('Company', 'X', 'country', cache=True)

frappe.db.get_single_value(doctype, fieldname)

Returns field from Single DocType.

tz = frappe.db.get_single_value('System Settings', 'time_zone')

frappe.db.set_value(doctype, name, fieldname, value=None, **kwargs)

Direct database update. SKIPS ORM validations and hooks.

frappe.db.set_value('Task', 'TASK-001', 'status', 'Closed')
frappe.db.set_value('Task', 'TASK-001', {'status': 'Closed', 'priority': 'Low'})
frappe.db.set_value('Task', 'TASK-001', 'status', 'Closed', update_modified=False)

frappe.db.exists(doctype, name_or_filters, cache=False)

Boolean existence check. ALWAYS use instead of try/except with get_doc.

exists = frappe.db.exists('User', 'admin@example.com')
exists = frappe.db.exists('User', {'email': 'admin@example.com'})
exists = frappe.db.exists('User', 'admin@example.com', cache=True)

frappe.db.count(doctype, filters=None)

Returns integer count of matching records.

total = frappe.db.count('Task')
open_count = frappe.db.count('Task', {'status': 'Open'})

frappe.db.delete(doctype, filters)

DML DELETE — can be rolled back.

frappe.db.delete('Error Log', {'creation': ['<', '2024-01-01']})

frappe.db.truncate(doctype)

DDL TRUNCATE — auto-commits, CANNOT be rolled back.

frappe.db.truncate('Error Log')

frappe.db.sql(query, values=None, **kwargs)

Executes raw SQL. ALWAYS use parameterized queries.

results = frappe.db.sql("""
    SELECT name, subject FROM `tabTask`
    WHERE status = %(status)s
""", {'status': 'Open'}, as_dict=True)

Return types:

• Default: tuple of tuples
• as_dict=True: list of dicts
• as_list=True: list of lists
• debug=True: prints SQL to console

frappe.db.multisql(queries)

Executes database-engine-specific SQL.

frappe.db.multisql({
    'mariadb': "SELECT IFNULL(field, 0) FROM `tabDoc`",
    'postgres': "SELECT COALESCE(field, 0) FROM `tabDoc`"
})

frappe.db.bulk_update(doctype, doc_updates, **kwargs) [v15+]

Batch updates using CASE expressions. Direct DB — skips ORM.

frappe.db.bulk_update('Task', {
    'TASK-001': {'status': 'Closed'},
    'TASK-002': {'status': 'Closed'}
}, chunk_size=100, update_modified=True)

---

Transaction Control

frappe.db.commit()

Manual commit. NEVER call inside controller hooks.

frappe.db.rollback(save_point=None)

Rollback to savepoint or entire transaction.

frappe.db.savepoint(save_point)

Creates named savepoint.

frappe.db.savepoint('my_savepoint')
try:
    # operations
except Exception:
    frappe.db.rollback(save_point='my_savepoint')

Transaction Hooks [v15+]

frappe.db.before_commit.add(func)
frappe.db.after_commit.add(func)
frappe.db.before_rollback.add(func)
frappe.db.after_rollback.add(func)

---

Schema Methods

frappe.db.add_index(doctype, fields, index_name=None)

frappe.db.add_index('Task', ['status', 'priority'])
frappe.db.add_index('Task', ['description(500)'])  # TEXT fields need length

frappe.db.add_unique(doctype, fields, constraint_name=None)

frappe.db.add_unique('User', ['email'])

frappe.db.describe(doctype)

Returns table schema description as tuple.

frappe.db.rename_table(old_name, new_name)

Renames database table. Use frappe.rename_doc for DocType renames instead.

frappe.db.change_column_type(doctype, column, new_type)

Alters column data type.

Query Patterns Reference

Query Builder (frappe.qb), filter operators, raw SQL patterns. Verified against Frappe v14-v16 docs.

Note: For comprehensive Query Builder coverage (advanced joins, HAVING, window functions, subquery patterns, cross-DB compatibility), see the dedicated [frappe-syntax-query-builder](../../../syntax/frappe-syntax-query-builder/SKILL.md) skill.

---

Filter Operators

Comparison

{'status': 'Open'}                    # = (equality)
{'status': ['!=', 'Cancelled']}       # !=
{'amount': ['>', 1000]}               # >
{'amount': ['>=', 1000]}              # >=
{'amount': ['<', 5000]}               # <
{'amount': ['<=', 5000]}              # <=

List Operators

{'status': ['in', ['Open', 'Working', 'Pending']]}       # IN
{'status': ['not in', ['Cancelled', 'Closed']]}           # NOT IN

Pattern Matching

{'subject': ['like', '%urgent%']}       # LIKE
{'email': ['like', '%@example.com']}    # LIKE

Range

{'date': ['between', ['2024-01-01', '2024-12-31']]}  # BETWEEN

NULL Checks

{'description': ['is', 'set']}         # IS NOT NULL
{'description': ['is', 'not set']}     # IS NULL

Combining Filters

# AND — dict (all conditions ANDed)
filters = {'status': 'Open', 'priority': 'High'}

# AND — list format (allows duplicate field names)
filters = [
    ['status', '=', 'Open'],
    ['priority', '=', 'High']
]

# OR — separate parameter
or_filters = {'priority': 'Urgent', 'status': 'Overdue'}

# Combined AND + OR
frappe.get_all('Task',
    filters={'docstatus': 0},
    or_filters={'priority': 'Urgent', 'exp_end_date': ['<', today()]}
)

---

Query Builder (frappe.qb)

The Query Builder wraps PyPika and generates parameterized SQL automatically. ALWAYS prefer over raw SQL.

Basic Select

Task = frappe.qb.DocType('Task')

results = (
    frappe.qb.from_(Task)
    .select(Task.name, Task.subject, Task.status)
    .where(Task.status == 'Open')
).run(as_dict=True)

DocType vs Table

# DocType — adds `tab` prefix automatically
Task = frappe.qb.DocType('Task')        # → `tabTask`

# Table — NO prefix (for internal tables)
Auth = frappe.qb.Table('__Auth')        # → `__Auth`

# Field — standalone column reference
field = frappe.qb.Field('name')

WHERE Conditions

Task = frappe.qb.DocType('Task')

# AND — chain .where()
query = (
    frappe.qb.from_(Task)
    .select('*')
    .where(Task.status == 'Open')
    .where(Task.priority == 'High')
)

# OR — use pipe operator
query = (
    frappe.qb.from_(Task)
    .select('*')
    .where(
        (Task.status == 'Open') | (Task.status == 'Working')
    )
)

# Combined AND + OR
query = (
    frappe.qb.from_(Task)
    .select('*')
    .where(
        (Task.priority == 'High') | (Task.priority == 'Urgent')
    )
    .where(Task.docstatus == 0)  # AND
)

# LIKE
query.where(Task.subject.like('%urgent%'))

# IN
query.where(Task.status.isin(['Open', 'Working']))

# IS NULL / IS NOT NULL
query.where(Task.description.isnull())
query.where(Task.description.isnotnull())

# BETWEEN
query.where(Task.creation.between('2024-01-01', '2024-12-31'))

INNER JOIN

SI = frappe.qb.DocType('Sales Invoice')
Customer = frappe.qb.DocType('Customer')

results = (
    frappe.qb.from_(SI)
    .inner_join(Customer).on(SI.customer == Customer.name)
    .select(SI.name, SI.grand_total, Customer.customer_name)
    .where(SI.docstatus == 1)
).run(as_dict=True)

LEFT JOIN

results = (
    frappe.qb.from_(SI)
    .left_join(Customer).on(SI.customer == Customer.name)
    .select(SI.name, Customer.customer_name)
).run(as_dict=True)

Aggregate Functions

from frappe.query_builder.functions import Count, Sum, Avg, Max, Min

Task = frappe.qb.DocType('Task')

stats = (
    frappe.qb.from_(Task)
    .select(
        Task.status,
        Count(Task.name).as_('count'),
        Sum(Task.expected_time).as_('total_time'),
        Avg(Task.expected_time).as_('avg_time'),
        Max(Task.expected_time).as_('max_time'),
        Min(Task.expected_time).as_('min_time')
    )
    .groupby(Task.status)
).run(as_dict=True)

Order, Limit, Offset

results = (
    frappe.qb.from_(Task)
    .select(Task.name, Task.subject)
    .orderby(Task.creation, order='desc')
    .limit(10)
    .offset(20)
).run(as_dict=True)

Subquery

User = frappe.qb.DocType('User')
Task = frappe.qb.DocType('Task')

subquery = (
    frappe.qb.from_(Task)
    .select(Task.assigned_to)
    .where(Task.status == 'Open')
)

results = (
    frappe.qb.from_(User)
    .select(User.name, User.full_name)
    .where(User.name.isin(subquery))
).run(as_dict=True)

Inspect Generated SQL

query = frappe.qb.from_(Task).select('*').where(Task.name == 'X')

# Parameterized form
sql, params = query.walk()
# ('SELECT * FROM `tabTask` WHERE `name`=%(param1)s', {'param1': 'X'})

# SQL string
sql_str = query.get_sql()
# Also: str(query)

run() Options

query.run()                    # Tuple of tuples (default)
query.run(as_dict=True)        # List of dicts
query.run(as_list=True)        # List of lists
query.run(debug=True)          # Print SQL to console

---

Cross-Database Compatibility

ImportMapper

Maps functions to correct database dialect automatically.

from frappe.query_builder.utils import ImportMapper, db_type_is
from frappe.query_builder.custom import GROUP_CONCAT, STRING_AGG

GroupConcat = ImportMapper({
    db_type_is.MARIADB: GROUP_CONCAT,
    db_type_is.POSTGRES: STRING_AGG
})

# Use GroupConcat in queries — resolves to correct function per DB

ConstantColumn

Adds a constant value as a pseudo-column.

from frappe.query_builder.custom import ConstantColumn

results = (
    frappe.qb.from_('DocType')
    .select('name', ConstantColumn('admin').as_('created_by'))
).run(as_dict=True)

Custom Functions

Extend PyPika for database-specific functions.

from pypika import CustomFunction

DateDiff = CustomFunction('DATEDIFF', ['date1', 'date2'])

Task = frappe.qb.DocType('Task')
results = (
    frappe.qb.from_(Task)
    .select(Task.name, DateDiff(Task.exp_end_date, Task.exp_start_date).as_('duration'))
).run(as_dict=True)

multisql for Raw SQL

When you MUST write database-specific raw SQL:

results = frappe.db.multisql({
    'mariadb': """
        SELECT name, IFNULL(description, '') as description
        FROM `tabTask` WHERE status = %(status)s
    """,
    'postgres': """
        SELECT name, COALESCE(description, '') as description
        FROM "tabTask" WHERE status = %(status)s
    """
}, {'status': 'Open'}, as_dict=True)

---

Raw SQL Patterns

ALWAYS Use Parameterized Queries

# ✅ CORRECT
results = frappe.db.sql("""
    SELECT name, subject FROM `tabTask`
    WHERE status = %(status)s AND owner = %(owner)s
""", {'status': 'Open', 'owner': frappe.session.user}, as_dict=True)

# ❌ NEVER — SQL injection vulnerability
frappe.db.sql(f"SELECT * FROM `tabTask` WHERE status = '{status}'")

JOIN Pattern

results = frappe.db.sql("""
    SELECT si.name, si.grand_total, c.customer_name
    FROM `tabSales Invoice` si
    INNER JOIN `tabCustomer` c ON si.customer = c.name
    WHERE si.docstatus = 1
    AND si.posting_date >= %(from_date)s
    ORDER BY si.grand_total DESC
    LIMIT %(limit)s
""", {'from_date': '2024-01-01', 'limit': 100}, as_dict=True)

Aggregate Pattern

results = frappe.db.sql("""
    SELECT status, COUNT(*) as count, SUM(expected_time) as total_time
    FROM `tabTask`
    GROUP BY status
    ORDER BY count DESC
""", as_dict=True)

Table Name Rules

• ALWAYS use backtick-quoted tab prefix: ` tabSales Invoice `
• Exact DocType name including spaces: ` tabPayment Entry `
• Child tables: ` tabSales Invoice Item `

---

v16 Breaking Changes

Aggregate Fields in get_list/get_all

# v14/v15 — string-based aggregates
frappe.db.get_list('Task',
    fields=['count(name) as count', 'status'],
    group_by='status'
)

# v16 — dict-based aggregates
frappe.db.get_list('Task',
    fields=[{'COUNT': 'name', 'as': 'count'}, 'status'],
    group_by='status'
)

run=False Behavior

# v14/v15 — returns SQL string
sql = frappe.db.get_list('Task', run=False)

# v16 — returns Query Builder object
qb_obj = frappe.db.get_list('Task', run=False)
sql = qb_obj.get_sql()   # Get SQL string
qb_obj.run()              # Execute