refactor: migrate a few queries to use query builder

Signed-off-by: Akhil Narang <me@akhilnarang.dev>
This commit is contained in:
Akhil Narang 2025-11-10 17:18:49 +05:30
parent 977aee5ab3
commit 5d992a5eb1
No known key found for this signature in database
GPG key ID: 9DCC61E211BF645F
5 changed files with 85 additions and 62 deletions

View file

@ -153,24 +153,25 @@ class TestDocType(IntegrationTestCase):
def test_all_depends_on_fields_conditions(self):
import re
docfields = frappe.get_all(
"DocField",
or_filters={
"ifnull(depends_on, '')": ("!=", ""),
"ifnull(collapsible_depends_on, '')": ("!=", ""),
"ifnull(mandatory_depends_on, '')": ("!=", ""),
"ifnull(read_only_depends_on, '')": ("!=", ""),
},
fields=[
"parent",
"depends_on",
"collapsible_depends_on",
"mandatory_depends_on",
"read_only_depends_on",
"fieldname",
"fieldtype",
],
DocField = frappe.qb.DocType("DocField")
docfields_query = (
frappe.qb.from_(DocField)
.select(
DocField.parent,
DocField.depends_on,
DocField.collapsible_depends_on,
DocField.mandatory_depends_on,
DocField.read_only_depends_on,
DocField.fieldname,
)
.where(
(DocField.depends_on != "")
| (DocField.collapsible_depends_on != "")
| (DocField.mandatory_depends_on != "")
| (DocField.read_only_depends_on != "")
)
)
docfields = docfields_query.run(as_dict=True)
pattern = r'[\w\.:_]+\s*={1}\s*[\w\.@\'"]+'
for field in docfields:

View file

@ -876,14 +876,20 @@ def get_all_roles():
"""return all roles"""
active_domains = frappe.get_active_domains()
roles = frappe.get_all(
"Role",
filters={
"name": ("not in", frappe.permissions.AUTOMATIC_ROLES),
"disabled": 0,
},
or_filters={"ifnull(restrict_to_domain, '')": "", "restrict_to_domain": ("in", active_domains)},
order_by="name",
Role = frappe.qb.DocType("Role")
domain_condition = (Role.restrict_to_domain.isnull()) | (Role.restrict_to_domain == "")
if active_domains:
domain_condition = domain_condition | Role.restrict_to_domain.isin(active_domains)
roles = (
frappe.qb.from_(Role)
.select(Role.name)
.where(
(Role.name.notin(frappe.permissions.AUTOMATIC_ROLES)) & (Role.disabled == 0) & domain_condition
)
.orderby(Role.name)
.run(as_dict=True)
)
return sorted([role.get("name") for role in roles])

View file

@ -32,14 +32,20 @@ def get_roles_and_doctypes():
active_domains = frappe.get_active_domains()
doctypes = frappe.get_all(
"DocType",
filters={
"istable": 0,
"name": ("not in", ",".join(not_allowed_in_permission_manager)),
},
or_filters={"ifnull(restrict_to_domain, '')": "", "restrict_to_domain": ("in", active_domains)},
fields=["name"],
DocType = frappe.qb.DocType("DocType")
doctype_domain_condition = (DocType.restrict_to_domain.isnull()) | (DocType.restrict_to_domain == "")
if active_domains:
doctype_domain_condition = doctype_domain_condition | DocType.restrict_to_domain.isin(active_domains)
doctypes = (
frappe.qb.from_(DocType)
.select(DocType.name)
.where(
(DocType.istable == 0)
& (DocType.name.notin(not_allowed_in_permission_manager))
& doctype_domain_condition
)
.run(as_dict=True)
)
restricted_roles = ["Administrator"]
@ -48,14 +54,16 @@ def get_roles_and_doctypes():
restricted_roles.extend(row.role for row in custom_user_type_roles)
restricted_roles.extend(AUTOMATIC_ROLES)
roles = frappe.get_all(
"Role",
filters={
"name": ("not in", restricted_roles),
"disabled": 0,
},
or_filters={"ifnull(restrict_to_domain, '')": "", "restrict_to_domain": ("in", active_domains)},
fields=["name"],
Role = frappe.qb.DocType("Role")
role_domain_condition = (Role.restrict_to_domain.isnull()) | (Role.restrict_to_domain == "")
if active_domains:
role_domain_condition = role_domain_condition | Role.restrict_to_domain.isin(active_domains)
roles = (
frappe.qb.from_(Role)
.select(Role.name)
.where((Role.name.notin(restricted_roles)) & (Role.disabled == 0) & role_domain_condition)
.run(as_dict=True)
)
doctypes_list = [{"label": _(d.get("name")), "value": d.get("name")} for d in doctypes]

View file

@ -152,13 +152,19 @@ def get_desktop_icons(user=None, bootinfo=None):
active_domains = frappe.get_active_domains()
blocked_doctypes = frappe.get_all(
"DocType",
filters={"ifnull(restrict_to_domain, '')": ("not in", ",".join(active_domains))},
fields=["name"],
)
blocked_doctypes = [d.get("name") for d in blocked_doctypes]
DocType = frappe.qb.DocType("DocType")
if active_domains:
blocked_condition = (
(DocType.restrict_to_domain.isnull())
| (DocType.restrict_to_domain == "")
| (DocType.restrict_to_domain.notin(active_domains))
)
else:
blocked_condition = (DocType.restrict_to_domain.isnull()) | (DocType.restrict_to_domain == "")
blocked_doctypes = [
d.get("name")
for d in frappe.qb.from_(DocType).select(DocType.name).where(blocked_condition).run(as_dict=True)
]
standard_icons = frappe.get_all("Desktop Icon", fields=fields, filters={"standard": 1})

View file

@ -3,6 +3,7 @@
import frappe
from frappe import _
from frappe.query_builder import Field, functions
@frappe.whitelist()
@ -42,24 +43,25 @@ def get_children(doctype, parent="", include_disabled=False, **filters):
def _get_children(doctype, parent="", ignore_permissions=False, include_disabled=False):
parent_field = "parent_" + frappe.scrub(doctype)
filters = [[f"ifnull(`{parent_field}`,'')", "=", parent], ["docstatus", "<", 2]]
if frappe.db.has_column(doctype, "disabled") and not include_disabled:
filters.append(["disabled", "=", False])
meta = frappe.get_meta(doctype)
return frappe.get_list(
doctype,
fields=[
"name as value",
"{} as title".format(meta.get("title_field") or "name"),
"is_group as expandable",
],
filters=filters,
order_by="name",
ignore_permissions=ignore_permissions,
qb = (
frappe.qb.from_(doctype)
.select(
Field("name").as_("value"),
Field(meta.get("title_field") or "name").as_("title"),
Field("is_group").as_("expandable"),
)
.where(functions.IfNull(Field(parent_field), "").eq(parent))
.where(Field("docstatus") < 2)
)
if frappe.db.has_column(doctype, "disabled") and not include_disabled:
qb = qb.where(Field("disabled").eq(False))
# Order by name and execute
return qb.orderby("name").run(as_dict=True)
@frappe.whitelist()
def add_node():