Query Builder & PyPika Extensions

Frappe's Query Builder is built on PyPika. While the standard API covers basic queries, advanced use cases require direct PyPika manipulation and custom function extensions.

Basic Query Builder Usage

import frappe
from frappe.query_builder import DocType
from frappe.query_builder.functions import Coalesce, Sum
from pypika.functions import NullIf

Item = DocType("Item")
Bin = DocType("Bin")

query = (
    frappe.qb.from_(Item)
    .inner_join(Bin).on(Bin.item_code == Item.name)
    .select(
        Item.name,
        Coalesce(Sum(Bin.actual_qty), 0).as_("total_qty"),
    )
    .where(Item.disabled == 0)
    .groupby(Item.name)
    .orderby(Item.name)
)
result = query.run(as_dict=True)

Table Aliasing

Use .as_() on DocType references for self-joins or clarity:

IG = DocType("Item Group")

query = (
    frappe.qb.from_(IG.as_("ig"))
    .select(
        IG.as_("ig").name.as_("ig_name"),
        IG.as_("ig").item_group_name,
    )
    .orderby(IG.as_("ig").lft)
)

Subquery JOINs (Derived Tables)

Build subqueries and join them as derived tables:

# Build subquery
stock_used_sq = (
    frappe.qb.from_(StockLedgerEntry)
    .inner_join(StockEntry)
    .on(
        (StockEntry.name == StockLedgerEntry.voucher_no)
        & (StockEntry.stock_entry_type == "Manufacture")
    )
    .select(
        StockLedgerEntry.item_code,
        Sum(StockLedgerEntry.actual_qty * -1).as_("stock_used"),
    )
    .where(StockLedgerEntry.actual_qty < 0)
    .groupby(StockLedgerEntry.item_code)
)

# Alias and join as derived table
StockUsed = stock_used_sq.as_("stock_used_sq")

query = (
    frappe.qb.from_(Items)
    .left_join(StockUsed).on(StockUsed.item_code == Items.name)
    .select(
        Items.name,
        Coalesce(StockUsed.stock_used, 0).as_("stock_used"),
    )
)

Literal Values with ValueWrapper

Insert literal values into SELECT for client-side-only columns:

from pypika.terms import ValueWrapper

query = (
    frappe.qb.from_(Items)
    .select(
        Items.name,
        ValueWrapper(0).as_("reorder_qty"),       # Editable on client only
        ValueWrapper(0).as_("projected_stock"),    # Computed on client only
    )
)

Computed Expressions in SELECT

Inline arithmetic in column selection:

query = (
    frappe.qb.from_(Bins)
    .select(
        (
            Coalesce(Sum(Bins.actual_qty), 0)
            - Coalesce(Sum(Bins.reserved_qty), 0)
            + Coalesce(OrderedStock.ordered_stock, 0)
        ).as_("available_stock"),
    )
)

Custom PyPika Functions Module

Create a utility module for custom SQL functions not available in PyPika:

# myapp/utils/json_functions.py

from typing import Any
from frappe.query_builder.functions import AggregateFunction
from frappe.query_builder.terms import ValueWrapper
from pypika.terms import Function, Term


class _Fn:
    """Callable wrapper that creates a Function term with any number of arguments.
    
    PyPika's CustomFunction drops ALL positional arguments when `params`
    is not provided. This wrapper always passes them through.
    """
    def __init__(self, name: str):
        self.name = name

    def __call__(self, *args: Any, **kwargs: Any) -> Function:
        return Function(self.name, *args, alias=kwargs.get("alias"))


# JSON creation functions (MariaDB/MySQL)
JSON_OBJECT = _Fn("JSON_OBJECT")
JSON_ARRAY = _Fn("JSON_ARRAY")
JSON_MERGE_PATCH = _Fn("JSON_MERGE_PATCH")
JSON_EXTRACT = _Fn("JSON_EXTRACT")
JSON_UNQUOTE = _Fn("JSON_UNQUOTE")
JSON_KEYS = _Fn("JSON_KEYS")

# SQL utility functions
IFNULL = _Fn("IFNULL")
CONCAT = _Fn("CONCAT")

JSON Aggregate Functions

class JSON_OBJECTAGG(AggregateFunction):
    """Aggregates key-value pairs into a JSON object."""
    def __init__(self, key, value, alias=None):
        super().__init__("JSON_OBJECTAGG", key, value, alias=alias)


class JSON_ARRAYAGG(AggregateFunction):
    """JSON_ARRAYAGG with optional DISTINCT and ORDER BY support.
    
    MariaDB syntax: JSON_ARRAYAGG([DISTINCT] expr [ORDER BY ...])
    """
    def __init__(self, term, distinct=False, orderby=None, alias=None):
        super().__init__("JSON_ARRAYAGG", alias=alias)
        self._term = term if isinstance(term, Term) else ValueWrapper(term)
        self._distinct = distinct
        self._orderby = orderby

    def get_function_sql(self, **kwargs: Any) -> str:
        term_sql = self._term.get_sql(with_alias=False, **kwargs)
        distinct_sql = "DISTINCT " if self._distinct else ""
        order_sql = ""
        if self._orderby is not None:
            order_terms = (
                self._orderby if isinstance(self._orderby, list) else [self._orderby]
            )
            order_parts = [
                t.get_sql(with_alias=False, **kwargs) if isinstance(t, Term) else str(t)
                for t in order_terms
            ]
            order_sql = f" ORDER BY {', '.join(order_parts)}"
        return f"JSON_ARRAYAGG({distinct_sql}{term_sql}{order_sql})"

Usage:

from myapp.utils.json_functions import JSON_OBJECTAGG, JSON_ARRAYAGG

result = (
    frappe.qb.from_(Item)
    .left_join(IVA).on(Item.name == IVA.parent)
    .select(
        Item.item_code,
        JSON_OBJECTAGG(IVA.attribute, IVA.attribute_value).as_("attributes"),
    )
    .where(Item.variant_of == template_item)
    .groupby(Item.item_code)
).run(as_dict=True)

Correlated Subquery as a Custom Term

For complex SQL that PyPika can't express natively, subclass Term and emit raw SQL:

class NESTED_SET_DEPTH(Term):
    """Calculate depth of a node in a nested set model by counting ancestors.
    
    Generates: (SELECT COUNT(*) FROM `tabX` a 
                WHERE a.lft < `alias`.lft AND a.rgt > `alias`.rgt
                [AND a.lft >= root_lft AND a.rgt <= root_rgt])
    """
    def __init__(self, table, outer_alias, root_lft=None, root_rgt=None):
        super().__init__()
        self._table = table
        self._outer_alias = outer_alias
        self._root_lft = root_lft
        self._root_rgt = root_rgt

    def get_sql(self, with_alias=False, **kwargs):
        table_name = f"`tab{self._table}`"
        outer_ref = f"`{self._outer_alias}`"
        sql = (
            f"(SELECT COUNT(*) FROM {table_name} a "
            f"WHERE a.lft < {outer_ref}.lft AND a.rgt > {outer_ref}.rgt"
        )
        if self._root_lft is not None and self._root_rgt is not None:
            sql += f" AND a.lft >= {self._root_lft} AND a.rgt <= {self._root_rgt}"
        sql += ")"
        if with_alias and self.alias:
            sql += f" `{self.alias}`"
        return sql

Usage in a query:

from myapp.utils.json_functions import NESTED_SET_DEPTH

IG = DocType("Item Group")
indent = NESTED_SET_DEPTH("Item Group", "ig")

query = (
    frappe.qb.from_(IG.as_("ig"))
    .select(
        IG.as_("ig").name,
        indent.as_("ig_indent"),  # Computed depth column
    )
    .orderby(IG.as_("ig").lft)
)

For subtree queries (depth relative to a subtree root):

root = frappe.qb.from_(IG).select(IG.lft, IG.rgt).where(IG.name == "Timber").run(as_dict=True)[0]
indent = NESTED_SET_DEPTH("Item Group", "ig", root_lft=root["lft"], root_rgt=root["rgt"])

External SQL Files

For complex queries, store SQL in separate files:

from pathlib import Path

query = (Path(__file__).parent / "my_report.sql").read_text()
result = frappe.db.sql(query, {"param": value}, as_dict=True)

Parameterized Optional Filters in SQL

Use this pattern for filters that may or may not be provided:

WHERE
    (p.name = %(project)s OR %(project)s IS NULL OR %(project)s = '')
    AND (p.status = %(status)s OR %(status)s IS NULL OR %(status)s = '')
    AND (
        %(date_start)s IS NULL
        OR %(date_end)s IS NULL
        OR p.actual_end_date BETWEEN %(date_start)s AND %(date_end)s
    )

Pass None for unused filters in the parameter dictionary.

JSON-Returning SQL

MariaDB can return structured JSON in a single result cell:

SELECT JSON_OBJECT(
    'attributes', (SELECT JSON_ARRAYAGG(JSON_OBJECT('name', a.name, 'values', a.values))
                   FROM attributes a),
    'variants', (SELECT JSON_ARRAYAGG(JSON_OBJECT('code', v.code, 'attrs', v.attrs))
                 FROM variants v)
) AS result

Parse in Python:

import json
raw = frappe.db.sql(query, params, pluck="result")
result = json.loads(raw[0] if raw[0] else [])