vassili/seitime-frappe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix(postgres): misc query building fixes + CI (#34831)

Browse source 
* fix(query): check standard field definitions

Signed-off-by: Akhil Narang <me@akhilnarang.dev>

* fix(postgres): fix order_by problem in pg

* fix(postgres): fix order_by in get_all for _test_connection_query

* fix: add check to a proper numeric fallback in _get_ifnull_fallback

* test(postgres): fix pg query used in assertion in test_permission_query

* fix(postgres): fix order_by in get_all for possible_link

* fix(postgres): fix order_by in get_all for set_modules

* fix(postgres): fix pg query count *

* fix(postgres): fix order_by in get_all for ask_pass_update

* fix(postgres): fix order_by statement in search_widget

* fix(postgres): fix order_by in get_list for get_stats

* test(postgres): normalize_sql for pg queries in test_arithmetic_operators_in_fields

* test(postgres): normalize_sql for pg queries in test_field_alias_in_group_by

* test(postgres): normalize_sql for pg queries in test_field_alias_permission_check

* test(postgres): fix order_by statement in get_all for test_db_keywords_as_fields

* test(postgres): fix order_by statement in get_all for test_prepare_select_args

* fix(treeview): use 0 instead of false to check since check field is an integer

* fix(postgres): fix order_by in get_all for sync_communication

* fix(postgres): fix order_by in get_all for get_references_across_doctypes_by_dynamic_link_field

* test(postgres): fix order_by in get_all for test_list_summary

* fix(postgres): fix order_by in get_all for email queries

* test(postgres): use order_by none and update assertion for postgres

* fix(postgres): use ILIKE to support case insensitive search in postgres

* test(test_query): update pg specific query assert to use ILIKE

* test(test_query): update test_nested_filters to use ilike instead for PG

* test(postgres): update pg query in assert to test updated qb query

* fix(search): update query to be db-agnostic

* test(postgres): normalize query for pg in test_build_match_conditions

* fix(postgres): suppress ORDER BY when SELECT DISTINCT in query for postgres specific behavior

* fix(postgres): suppress ORDER BY when GROUP BY is explicitly asked for pg specific behavior

* test(postgres): fix test behavior for pg ORDER BY drop when used with GROUP BY

* refactor: reducing noise in code by formatting code

* fix(query): use Star() to handle SQL wildcard character * correctly

* fix(postgres): display warning for ORDER BY fields that will be dropped

---------

Signed-off-by: Akhil Narang <me@akhilnarang.dev>
Co-authored-by: Akhil Narang <me@akhilnarang.dev>
This commit is contained in:
Aarol D'Souza 2025-12-05 10:30:49 +05:30 • committed by GitHub
parent 4e464cfb8e
commit cf69e4bed1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 143 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
frappe/core/doctype/communication/communication.py
View file

@ -505,7 +505,10 @@ def get_permission_query_conditions_for_communication(user):
return None
else:
accounts = frappe.get_all(
"User Email", filters={"parent": user}, fields=["email_account"], distinct=True, order_by="idx"
"User Email",
filters={"parent": user},
fields=["email_account"],
distinct=True,
)
if not accounts:

5
frappe/core/doctype/domain_settings/domain_settings.py
View file

@ -78,7 +78,10 @@ def get_active_domains():
def _get_active_domains():
domains = frappe.get_all(
"Has Domain", filters={"parent": "Domain Settings"}, fields=["domain"], distinct=True
"Has Domain",
filters={"parent": "Domain Settings"},
fields=["domain"],
distinct=True,
)
active_domains = [row.get("domain") for row in domains]

5
frappe/core/doctype/server_script/test_server_script.py
View file

@ -158,10 +158,7 @@ class TestServerScript(IntegrationTestCase):
def test_permission_query(self):
sql = frappe.db.get_list("ToDo", run=False)
if frappe.conf.db_type != "postgres":
self.assertTrue("where (1 = 1)" in sql.lower())
else:
self.assertTrue("where (1 = '1')" in sql.lower())
self.assertTrue("where (1 = 1)" in sql.lower())
self.assertTrue(isinstance(frappe.db.get_list("ToDo"), list))
def test_attribute_error(self):

5
frappe/core/doctype/user/user.py
View file

@ -1018,7 +1018,10 @@ def ask_pass_update():
from frappe.utils import set_default
password_list = frappe.get_all(
"User Email", filters={"awaiting_password": 1, "used_oauth": 0}, pluck="parent", distinct=True
"User Email",
filters={"awaiting_password": 1, "used_oauth": 0},
pluck="parent",
distinct=True,
)
set_default("email_user_password", ",".join(password_list))

12
frappe/database/operator_map.py
View file

@ -24,6 +24,17 @@ def like(key: Field, value: str) -> frappe.qb:
return key.like(value)
def ilike(key: Field, value: str) -> frappe.qb:
"""Wrapper method for `ILIKE`
Args:
key (str): field
value (str): criterion
Return:
frappe.qb: `frappe.qb` object with `ILIKE`
"""
return key.ilike(value)
def func_in(key: Field, value: list | tuple) -> frappe.qb:
"""Wrapper method for `IN`.
@ -136,6 +147,7 @@ OPERATOR_MAP: dict[str, Callable] = {
"in": func_in,
"not in": func_not_in,
"like": like,
"ilike": ilike,
"not like": not_like,
"regex": func_regex,
"between": func_between,

26
frappe/database/query.py
View file

@ -1,5 +1,6 @@
import datetime
import re
import warnings
from functools import lru_cache
from typing import TYPE_CHECKING, Any
@ -227,6 +228,7 @@ class Engine:
if self.apply_permissions:
self.check_read_permission()
is_select = False
if update:
self.query = qb.update(self.table, immutable=False)
elif into:
@ -236,6 +238,7 @@ class Engine:
else:
self.query = qb.from_(self.table, immutable=False)
self.apply_fields(fields)
is_select = True
self.apply_filters(filters)
self.apply_or_filters(or_filters)
@ -260,7 +263,19 @@ class Engine:
self.apply_group_by(group_by)
if order_by:
self.apply_order_by(order_by)
if not (
self.is_postgres and is_select and (distinct or group_by)
): # ignore in Postgres since order by fields need to appear in select distinct
self.apply_order_by(order_by)
else:
warnings.warn(
(
"ORDER BY fields have been ignored because PostgreSQL requires them to "
"appear in the SELECT list when using DISTINCT or GROUP BY."
),
UserWarning,
stacklevel=2,
)
if self.apply_permissions:
self.add_permission_conditions()
@ -512,7 +527,12 @@ class Engine:
)
return operator_fn(_field, nodes or ("",))
operator_fn = OPERATOR_MAP[_operator.casefold()]
if (
self.is_postgres and _operator.casefold() == "like"
): # use `ILIKE` to support case insensitive search in postgres
operator_fn = OPERATOR_MAP["ilike"]
else:
operator_fn = OPERATOR_MAP[_operator.casefold()]
if _value is None and isinstance(_field, Field):
if operator_fn == builtin_operator.ne:
filter_field_name = (
@ -1425,7 +1445,7 @@ class Engine:
if fieldtype == "Time":
return "'00:00:00'"
if fieldtype in ("Float", "Int", "Currency", "Percent"):
if fieldtype in ("Float", "Int", "Currency", "Percent", "Check"):
return "0"
try:

7
frappe/desk/form/linked_with.py
View file

@ -305,7 +305,12 @@ def get_references_across_doctypes_by_dynamic_link_field(
for doctype, fieldname, doctype_fieldname in links:
try:
filters = [[doctype_fieldname, "in", to_doctypes]] if to_doctypes else []
for linked_to in frappe.get_all(doctype, pluck=doctype_fieldname, filters=filters, distinct=1):
for linked_to in frappe.get_all(
doctype,
pluck=doctype_fieldname,
filters=filters,
distinct=1,
):
if linked_to:
links_by_doctype[linked_to].append(
{"doctype": doctype, "fieldname": fieldname, "doctype_fieldname": doctype_fieldname}

12
frappe/desk/search.py
View file

@ -195,15 +195,9 @@ def search_widget(
_relevance_expr = {"DIV": [1, {"NULLIF": [{"LOCATE": [_txt, "name"]}, 0]}]}
# For MariaDB, wrap in IFNULL for sorting to push nulls to end
if frappe.db.db_type in ("mariadb", "sqlite"):
_relevance = {"IFNULL": [_relevance_expr, -9999], "as": "_relevance"}
formatted_fields.append(_relevance)
order_by = f"_relevance desc, {order_by}"
elif frappe.db.db_type == "postgres":
_relevance = {**_relevance_expr, "as": "_relevance"}
formatted_fields.append(_relevance)
# Since we are sorting by alias postgres needs to know number of column we are sorting
order_by = f"{len(formatted_fields)} desc nulls last, {order_by}"
_relevance = {"IFNULL": [_relevance_expr, -9999], "as": "_relevance"}
formatted_fields.append(_relevance)
order_by = f"_relevance desc, {order_by}"
values = frappe.get_list(
doctype,

4
frappe/desk/treeview.py
View file

@ -57,8 +57,8 @@ def _get_children(doctype, parent="", ignore_permissions=False, include_disabled
)
if frappe.db.has_column(doctype, "disabled") and not include_disabled:
qb = qb.where(Field("disabled").eq(False))
# used 0 instead of `false` since type of check in postgres is smallint
qb = qb.where(Field("disabled").eq(0))
# Order by name and execute
return qb.orderby("name").run(as_dict=True)

3
frappe/email/test_smtp.py
View file

@ -18,7 +18,8 @@ class TestSMTP(IntegrationTestCase):
def test_get_email_account(self):
existing_email_accounts = frappe.get_all(
"Email Account", fields=["name", "enable_outgoing", "default_outgoing", "append_to", "use_imap"]
"Email Account",
fields=["name", "enable_outgoing", "default_outgoing", "append_to", "use_imap"],
)
unset_details = {"enable_outgoing": 0, "default_outgoing": 0, "append_to": None, "use_imap": 0}
for email_account in existing_email_accounts:

4
frappe/tests/test_db.py
View file

@ -403,8 +403,8 @@ class TestDB(IntegrationTestCase):
random_field,
)
self.assertEqual(
next(iter(frappe.get_all("ToDo", fields=[{"COUNT": random_field}], limit=1)[0])),
"COUNT" if frappe.conf.db_type == "postgres" else f"COUNT(`{random_field}`)",
next(iter(frappe.get_all("ToDo", fields=[{"COUNT": random_field}], limit=1, order_by=None)[0])),
"count" if frappe.conf.db_type == "postgres" else f"COUNT(`{random_field}`)",
)
# Testing update

6
frappe/tests/test_db_query.py
View file

@ -848,11 +848,7 @@ class TestDBQuery(IntegrationTestCase):
limit=1,
as_list=True,
)
if frappe.conf.db_type == "mariadb":
self.assertTrue(len(doctypes[0]) == 2)
else:
self.assertTrue(len(doctypes[0]) == 3)
self.assertTrue(isinstance(doctypes[0][2], datetime.datetime))
self.assertTrue(len(doctypes[0]) == 2) # same for pg as well since we order_by None
def test_field_comparison(self):
"""Test DatabaseQuery.execute to test field comparison"""

160
frappe/tests/test_query.py
View file

@ -68,6 +68,8 @@ class TestQuery(IntegrationTestCase):
setup_for_tests()
def test_multiple_tables_in_filters(self):
query = "SELECT `tabDocType`.* FROM `tabDocType` LEFT JOIN `tabDocField` ON `tabDocField`.`parent`=`tabDocType`.`name` AND `tabDocField`.`parenttype`='DocType' AND `tabDocField`.`parentfield`='fields' WHERE `tabDocField`.`name` LIKE 'f%' AND `tabDocType`.`parent`='something'"
query = query.replace("LIKE", "ILIKE" if frappe.db.db_type == "postgres" else "LIKE")
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
@ -77,7 +79,7 @@ class TestQuery(IntegrationTestCase):
["DocType", "parent", "=", "something"],
],
).get_sql(),
"SELECT `tabDocType`.* FROM `tabDocType` LEFT JOIN `tabDocField` ON `tabDocField`.`parent`=`tabDocType`.`name` AND `tabDocField`.`parenttype`='DocType' AND `tabDocField`.`parentfield`='fields' WHERE `tabDocField`.`name` LIKE 'f%' AND `tabDocType`.`parent`='something'",
query,
)
def test_string_fields(self):
@ -360,13 +362,15 @@ class TestQuery(IntegrationTestCase):
"SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module` WHERE `tabModule Def`.`app_name`='frappe'",
)
query = "SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module` WHERE `tabModule Def`.`app_name` LIKE 'frap%'"
query = query.replace("LIKE", "ILIKE" if frappe.db.db_type == "postgres" else "LIKE")
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
filters={"module.app_name": ("like", "frap%")},
).get_sql(),
"SELECT `tabDocType`.`name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module` WHERE `tabModule Def`.`app_name` LIKE 'frap%'",
query,
)
self.assertQueryEqual(
@ -422,141 +426,125 @@ class TestQuery(IntegrationTestCase):
def test_or_filters(self):
"""Test OR filter conditions."""
# Test 1: Basic dict or_filters
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
or_filters={"name": "User", "module": "Core"},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name`='User' OR `module`='Core'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name`='User' OR `module`='Core'",
)
# Test 2: List format or_filters
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
or_filters=[["name", "=", "User"], ["module", "=", "Core"]],
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name`='User' OR `module`='Core'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name`='User' OR `module`='Core'",
)
# Test 3: OR filters with operators
self.assertEqual(
query = "SELECT `name` FROM `tabDocType` WHERE `name` LIKE 'User%' OR `module` IN ('Core','Custom')"
query = query = query.replace("LIKE", "ILIKE" if frappe.db.db_type == "postgres" else "LIKE")
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
or_filters={"name": ("like", "User%"), "module": ("in", ["Core", "Custom"])},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name` LIKE 'User%' OR `module` IN ('Core','Custom')".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
query,
)
# Test 4: Combining filters (AND) with or_filters (OR)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
filters={"issingle": 0},
or_filters={"name": "User", "module": "Core"},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `issingle`=0 AND (`name`='User' OR `module`='Core')".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `issingle`=0 AND (`name`='User' OR `module`='Core')",
)
# Test 5: Multiple AND filters with OR filters
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
filters={"issingle": 0, "custom": 0},
or_filters={"name": "User", "module": "Core"},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `issingle`=0 AND `custom`=0 AND (`name`='User' OR `module`='Core')".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `issingle`=0 AND `custom`=0 AND (`name`='User' OR `module`='Core')",
)
# Test 6: OR filters with simple list (name IN)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
or_filters=["User", "Role", "Note"],
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name` IN ('User','Role','Note')".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name` IN ('User','Role','Note')",
)
# Test 7: OR filters with greater than and less than
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
or_filters={"idx": (">", 5), "issingle": ("=", 1)},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `idx`>5 OR `issingle`=1".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `idx`>5 OR `issingle`=1",
)
# Test 8: OR filters with list including doctype
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
or_filters=[["DocType", "name", "=", "User"], ["DocType", "name", "=", "Role"]],
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name`='User' OR `name`='Role'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name`='User' OR `name`='Role'",
)
# Test 9: OR filters with != operator
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
or_filters={"name": ("!=", "User"), "module": ("!=", "Core")},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name`<>'User' OR `module`<>'Core'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name`<>'User' OR `module`<>'Core'",
)
# Test 10: Empty or_filters should return query without OR conditions
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
filters={"custom": 0},
or_filters={},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `custom`=0".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `custom`=0",
)
# Test 11: OR filters with not in operator
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name"],
or_filters={"name": ("not in", ["User", "Role"]), "module": ("=", "Core")},
).get_sql(),
"SELECT `name` FROM `tabDocType` WHERE `name` NOT IN ('User','Role') OR `module`='Core'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `name` FROM `tabDocType` WHERE `name` NOT IN ('User','Role') OR `module`='Core'",
)
# Test 12: OR filters with mixed field types
self.assertEqual(
query = (
"SELECT `name`,`module` FROM `tabDocType` WHERE `name` LIKE 'User%' OR `issingle`=1 OR `custom`=0"
)
query = query = query.replace("LIKE", "ILIKE" if frappe.db.db_type == "postgres" else "LIKE")
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name", "module"],
@ -566,9 +554,7 @@ class TestQuery(IntegrationTestCase):
["custom", "=", 0],
],
).get_sql(),
"SELECT `name`,`module` FROM `tabDocType` WHERE `name` LIKE 'User%' OR `issingle`=1 OR `custom`=0".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
query,
)
def test_nested_filters(self):
@ -665,7 +651,11 @@ class TestQuery(IntegrationTestCase):
.where(
(User.creation > "2023-01-01")
& (
(User.email.like("%@example.com"))
(
User.email.ilike("%@example.com")
if frappe.db.db_type == "postgres"
else User.email.like("%@example.com")
)
| ((User.first_name.isin(["Admin", "Guest"])) & (User.enabled != 1))
)
)
@ -710,49 +700,41 @@ class TestQuery(IntegrationTestCase):
def test_implicit_join_query(self):
self.maxDiff = None
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"Note",
filters={"name": "Test Note Title"},
fields=["name", "`tabNote Seen By`.`user` as seen_by"],
).get_sql(),
"SELECT `tabNote`.`name`,`tabNote Seen By`.`user` `seen_by` FROM `tabNote` LEFT JOIN `tabNote Seen By` ON `tabNote Seen By`.`parent`=`tabNote`.`name` AND `tabNote Seen By`.`parenttype`='Note' WHERE `tabNote`.`name`='Test Note Title'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `tabNote`.`name`,`tabNote Seen By`.`user` `seen_by` FROM `tabNote` LEFT JOIN `tabNote Seen By` ON `tabNote Seen By`.`parent`=`tabNote`.`name` AND `tabNote Seen By`.`parenttype`='Note' WHERE `tabNote`.`name`='Test Note Title'",
)
# output doesn't contain parentfield condition because it can't be inferred
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"Note",
filters={"name": "Test Note Title"},
fields=["name", "`tabNote Seen By`.`user` as seen_by", "`tabNote Seen By`.`idx` as idx"],
).get_sql(),
"SELECT `tabNote`.`name`,`tabNote Seen By`.`user` `seen_by`,`tabNote Seen By`.`idx` `idx` FROM `tabNote` LEFT JOIN `tabNote Seen By` ON `tabNote Seen By`.`parent`=`tabNote`.`name` AND `tabNote Seen By`.`parenttype`='Note' WHERE `tabNote`.`name`='Test Note Title'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `tabNote`.`name`,`tabNote Seen By`.`user` `seen_by`,`tabNote Seen By`.`idx` `idx` FROM `tabNote` LEFT JOIN `tabNote Seen By` ON `tabNote Seen By`.`parent`=`tabNote`.`name` AND `tabNote Seen By`.`parenttype`='Note' WHERE `tabNote`.`name`='Test Note Title'",
)
# output contains parentfield condition because it can be inferred by "seen_by.user"
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"Note",
filters={"name": "Test Note Title"},
fields=["name", "seen_by.user as seen_by", "`tabNote Seen By`.`idx` as idx"],
).get_sql(),
"SELECT `tabNote`.`name`,`tabNote Seen By`.`user` `seen_by`,`tabNote Seen By`.`idx` `idx` FROM `tabNote` LEFT JOIN `tabNote Seen By` ON `tabNote Seen By`.`parent`=`tabNote`.`name` AND `tabNote Seen By`.`parenttype`='Note' AND `tabNote Seen By`.`parentfield`='seen_by' WHERE `tabNote`.`name`='Test Note Title'".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `tabNote`.`name`,`tabNote Seen By`.`user` `seen_by`,`tabNote Seen By`.`idx` `idx` FROM `tabNote` LEFT JOIN `tabNote Seen By` ON `tabNote Seen By`.`parent`=`tabNote`.`name` AND `tabNote Seen By`.`parenttype`='Note' AND `tabNote Seen By`.`parentfield`='seen_by' WHERE `tabNote`.`name`='Test Note Title'",
)
self.assertEqual(
self.assertQueryEqual(
frappe.qb.get_query(
"DocType",
fields=["name", "module.app_name as app_name"],
).get_sql(),
"SELECT `tabDocType`.`name`,`tabModule Def`.`app_name` `app_name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module`".replace(
"`", '"' if frappe.db.db_type == "postgres" else "`"
),
"SELECT `tabDocType`.`name`,`tabModule Def`.`app_name` `app_name` FROM `tabDocType` LEFT JOIN `tabModule Def` ON `tabModule Def`.`name`=`tabDocType`.`module`",
)
# fields now has strict validation, so this test is not valid anymore
@ -878,7 +860,9 @@ class TestQuery(IntegrationTestCase):
if frappe.db.db_type == "mariadb":
self.assertIn("IFNULL(`name`,'')='' OR `name` IN ('_Test Blog Post 1','_Test Blog Post')", query)
elif frappe.db.db_type == "postgres":
self.assertIn("\"name\" IS NULL OR \"name\" IN ('_Test Blog Post 1','_Test Blog Post')", query)
self.assertIn(
"IFNULL(\"name\",'')='' OR \"name\" IN ('_Test Blog Post 1','_Test Blog Post')", query
) # works in pg due to `coalesce` sub during sql execution
frappe.set_user("Administrator")
clear_user_permissions_for_doctype("Test Blog Post", "test2@example.com")
@ -1762,39 +1746,39 @@ class TestQuery(IntegrationTestCase):
# Test simple addition
query = frappe.qb.get_query("User", fields=[{"ADD": [1, 2], "as": "sum_result"}])
sql = query.get_sql()
self.assertIn("1+2 `sum_result`", sql)
self.assertIn(self.normalize_sql("1+2 `sum_result`"), sql)
# Test simple subtraction
query = frappe.qb.get_query("User", fields=[{"SUB": [10, 5], "as": "diff_result"}])
sql = query.get_sql()
self.assertIn("10-5 `diff_result`", sql)
self.assertIn(self.normalize_sql("10-5 `diff_result`"), sql)
# Test simple multiplication
query = frappe.qb.get_query("User", fields=[{"MUL": [3, 4], "as": "prod_result"}])
sql = query.get_sql()
self.assertIn("3*4 `prod_result`", sql)
self.assertIn(self.normalize_sql("3*4 `prod_result`"), sql)
# Test simple division
query = frappe.qb.get_query("User", fields=[{"DIV": [10, 2], "as": "div_result"}])
sql = query.get_sql()
self.assertIn("10/2 `div_result`", sql)
self.assertIn(self.normalize_sql("10/2 `div_result`"), sql)
# Test operator with field names
query = frappe.qb.get_query("User", fields=[{"ADD": ["enabled", "login_after"], "as": "field_sum"}])
sql = query.get_sql()
self.assertIn("`enabled`+`login_after` `field_sum`", sql)
self.assertIn(self.normalize_sql("`enabled`+`login_after` `field_sum`"), sql)
# Test nested operators
query = frappe.qb.get_query("User", fields=[{"ADD": [{"MUL": [2, 3]}, 4], "as": "nested_result"}])
sql = query.get_sql()
self.assertIn("2*3+4 `nested_result`", sql)
self.assertIn(self.normalize_sql("2*3+4 `nested_result`"), sql)
# Test operator with function - NULLIF
query = frappe.qb.get_query(
"User", fields=[{"DIV": [1, {"NULLIF": ["enabled", 0]}], "as": "safe_div"}]
)
sql = query.get_sql()
self.assertIn("1/NULLIF(`enabled`,0) `safe_div`", sql)
self.assertIn(self.normalize_sql("1/NULLIF(`enabled`,0) `safe_div`"), self.normalize_sql(sql))
# Test complex nested expression: (1 / NULLIF(value, 0))
query = frappe.qb.get_query(
@ -1805,8 +1789,8 @@ class TestQuery(IntegrationTestCase):
],
)
sql = query.get_sql()
self.assertIn("`name`", sql)
self.assertIn("1/NULLIF(`enabled`,0) `inverse`", sql)
self.assertIn(self.normalize_sql("`name`"), sql)
self.assertIn(self.normalize_sql("1/NULLIF(`enabled`,0) `inverse`"), self.normalize_sql(sql))
# Test operator with LOCATE function (search relevance pattern)
query = frappe.qb.get_query(
@ -1817,7 +1801,10 @@ class TestQuery(IntegrationTestCase):
],
)
sql = query.get_sql()
self.assertIn("1/NULLIF(LOCATE('test',`name`),0) `relevance`", sql)
self.assertIn(
self.normalize_sql("1/NULLIF(LOCATE('test',`name`),0) `relevance`"),
self.normalize_sql(sql),
)
# Test multiple operators in fields
query = frappe.qb.get_query(
@ -1829,9 +1816,9 @@ class TestQuery(IntegrationTestCase):
],
)
sql = query.get_sql()
self.assertIn("`name`", sql)
self.assertIn("`enabled`+1 `enabled_plus_one`", sql)
self.assertIn("`enabled`*2 `enabled_times_two`", sql)
self.assertIn(self.normalize_sql("`name`"), sql)
self.assertIn(self.normalize_sql("`enabled`+1 `enabled_plus_one`"), sql)
self.assertIn(self.normalize_sql("`enabled`*2 `enabled_times_two`"), sql)
# Test operator without alias
query = frappe.qb.get_query("User", fields=[{"ADD": [1, 1]}])
@ -1898,9 +1885,12 @@ class TestQuery(IntegrationTestCase):
)
sql = query.get_sql()
self.assertIn("GROUP BY `created_date`", sql)
self.assertIn("ORDER BY `created_date`", sql)
self.assertIn("`creation` `created_date`", sql)
self.assertIn(self.normalize_sql("GROUP BY `created_date`"), self.normalize_sql(sql))
if (
frappe.db.db_type != "postgres"
): # since Postgres requires fields in Order by to be grouped or aggregated, order by is dropped
self.assertIn(self.normalize_sql("ORDER BY `created_date`"), self.normalize_sql(sql))
self.assertIn(self.normalize_sql("`creation` `created_date`"), self.normalize_sql(sql))
def test_field_alias_permission_check(self):
query = frappe.qb.get_query(
@ -1910,7 +1900,7 @@ class TestQuery(IntegrationTestCase):
)
sql = query.get_sql()
# If we get here without PermissionError, the test passes
self.assertIn("GROUP BY `created_date`", sql)
self.assertIn(self.normalize_sql("GROUP BY `created_date`"), self.normalize_sql(sql))
# This function is used as a permission query condition hook

5
frappe/utils/user.py
View file

@ -176,7 +176,10 @@ class UserPermissions:
self.can_read += self.can_write
self.shared = frappe.get_all(
"DocShare", {"user": self.name, "read": 1}, distinct=True, pluck="share_doctype"
"DocShare",
{"user": self.name, "read": 1},
distinct=True,
pluck="share_doctype",
)
self.can_read = list(set(self.can_read + self.shared))
self.all_read += self.can_read