fix: (Linter) Write queries using QB/ORM and other minor lines for semgrep to skip

This commit is contained in:
marination 2022-02-02 12:58:31 +05:30
parent 780e29b42e
commit 4b62d2d7fe
13 changed files with 89 additions and 84 deletions

View File

@ -435,7 +435,7 @@ def get_existing_payment_request_amount(ref_dt, ref_dn):
""", (ref_dt, ref_dn))
return flt(existing_payment_request_amount[0][0]) if existing_payment_request_amount else 0
def get_gateway_details(args):
def get_gateway_details(args): # nosemgrep
"""return gateway and payment account of default payment gateway"""
if args.get("payment_gateway_account"):
return get_payment_gateway_account(args.get("payment_gateway_account"))

View File

@ -41,7 +41,7 @@ class TestECommerceSettings(unittest.TestCase):
def test_tax_rule_validation(self):
frappe.db.sql("update `tabTax Rule` set use_for_shopping_cart = 0")
frappe.db.commit()
frappe.db.commit() # nosemgrep
cart_settings = self.get_cart_settings()
cart_settings.enabled = 1

View File

@ -57,16 +57,19 @@ class WebsiteItem(WebsiteGenerator):
self.publish_unpublish_desk_item(publish=True)
if not self.get("__islocal"):
self.old_website_item_groups = frappe.db.sql_list("""
select
item_group
from
`tabWebsite Item Group`
where
parentfield='website_item_groups'
and parenttype='Website Item'
and parent=%s
""", self.name)
wig = frappe.qb.DocType("Website Item Group")
query = (
frappe.qb.from_(wig)
.select(wig.item_group)
.where(
(wig.parentfield == "website_item_groups")
& (wig.parenttype == "Website Item")
& (wig.parent == self.name)
)
)
result = query.run(as_list=True)
self.old_website_item_groups = [x[0] for x in result]
def on_update(self):
invalidate_cache_for_web_item(self)
@ -330,18 +333,22 @@ class WebsiteItem(WebsiteGenerator):
return tab_values
def get_recommended_items(self, settings):
items = frappe.db.sql(f"""
select
ri.website_item_thumbnail, ri.website_item_name,
ri.route, ri.item_code
from
`tabRecommended Items` ri, `tabWebsite Item` wi
where
ri.item_code = wi.item_code
and ri.parent = '{self.name}'
and wi.published = 1
order by ri.idx
""", as_dict=1)
ri = frappe.qb.DocType("Recommended Items")
wi = frappe.qb.DocType("Website Item")
query = (
frappe.qb.from_(ri)
.join(wi).on(ri.item_code == wi.item_code)
.select(
ri.item_code, ri.route,
ri.website_item_name,
ri.website_item_thumbnail
).where(
(ri.parent == self.name)
& (wi.published == 1)
).orderby(ri.idx)
)
items = query.run(as_dict=True)
if settings.show_price:
is_guest = frappe.session.user == "Guest"

View File

@ -57,7 +57,7 @@ def remove_from_wishlist(item_code):
"parent": frappe.session.user
}
)
frappe.db.commit()
frappe.db.commit() # nosemgrep
wishlist_items = frappe.db.get_values(
"Wishlist Item",

View File

@ -99,18 +99,14 @@ class ProductFiltersBuilder:
if not attributes:
return []
result = frappe.db.sql(
"""
select
distinct attribute, attribute_value
from
`tabItem Variant Attribute`
where
attribute in %(attributes)s
and attribute_value is not null
""",
{"attributes": attributes},
as_dict=1,
result = frappe.get_all(
"Item Variant Attribute",
filters={
"attribute": ["in", attributes],
"attribute_value": ["is", "set"]
},
fields=["attribute", "attribute_value"],
distinct=True
)
attribute_value_map = {}

View File

@ -585,10 +585,20 @@ def get_shipping_rules(quotation=None, cart_settings=None):
if quotation.shipping_address_name:
country = frappe.db.get_value("Address", quotation.shipping_address_name, "country")
if country:
shipping_rules = frappe.db.sql_list("""select distinct sr.name
from `tabShipping Rule Country` src, `tabShipping Rule` sr
where src.country = %s and
sr.disabled != 1 and sr.name = src.parent""", country)
sr_country = frappe.qb.DocType("Shipping Rule Country")
sr = frappe.qb.DocType("Shipping Rule")
query = (
frappe.qb.from_(sr_country)
.join(sr).on(sr.name == sr_country.parent)
.select(sr.name)
.distinct()
.where(
(sr_country.country == country)
& (sr.disabled != 1)
)
)
result = query.run(as_list=True)
shipping_rules = [x[0] for x in result]
return shipping_rules

View File

@ -60,7 +60,7 @@ def get_item_codes_by_attributes(attribute_filters, template_item_code=None):
NULL
'''.format(attribute_query=attribute_query, variant_of_query=variant_of_query)
item_codes = set([r[0] for r in frappe.db.sql(query, query_values)])
item_codes = set([r[0] for r in frappe.db.sql(query, query_values)]) # nosemgrep
items.append(item_codes)
res = list(set.intersection(*items))

View File

@ -17,7 +17,7 @@ def execute():
"website_warehouse", "web_long_description", "website_content", "thumbnail"]
# get all valid columns (fields) from Item master DB schema
item_table_fields = frappe.db.sql("desc `tabItem`", as_dict=1)
item_table_fields = frappe.db.sql("desc `tabItem`", as_dict=1) # nosemgrep
item_table_fields = [d.get('Field') for d in item_table_fields]
# prepare fields to query from Item, check if the web field exists in Item master

View File

@ -24,17 +24,17 @@ def execute():
settings = frappe.get_doc("E Commerce Settings")
def map_into_e_commerce_settings(doctype, fields):
data = frappe.db.sql("""
Select
field, value
from `tabSingles`
where
doctype='{doctype}'
and field in ({fields})
""".format(
doctype=doctype,
fields=(",").join(['%s'] * len(fields))
), tuple(fields), as_dict=1)
singles = frappe.qb.DocType("Singles")
query = (
frappe.qb.from_(singles)
.select(
singles.field, singles.value
).where(
(singles.doctype == doctype)
& (singles.field in fields)
)
)
data = query.run(as_dict=True)
# {'enable_attribute_filters': '1', ...}
mapper = {row.field: row.value for row in data}
@ -51,10 +51,12 @@ def execute():
# move filters and attributes tables to E Commerce Settings from Products Settings
for doctype in ("Website Filter Field", "Website Attribute"):
frappe.db.sql("""Update `tab{doctype}`
set
parenttype = 'E Commerce Settings',
parent = 'E Commerce Settings'
where
parent = 'Products Settings'
""".format(doctype=doctype))
frappe.db.set_value(
doctype,
{"parent": "Products Settings"},
{
"parenttype": "E Commerce Settings",
"parent": "E Commerce Settings"
},
update_modified=False
)

View File

@ -29,7 +29,7 @@ def create_fiscal_year_and_company(args):
'domain': args.get('domains')[0]
}).insert()
def enable_shopping_cart(args):
def enable_shopping_cart(args): # nosemgrep
# Needs price_lists
frappe.get_doc({
"doctype": "E Commerce Settings",

View File

@ -535,7 +535,7 @@ def create_bank_account(args):
# bank account same as a CoA entry
pass
def update_shopping_cart_settings(args):
def update_shopping_cart_settings(args): # nosemgrep
shopping_cart = frappe.get_doc("E Commerce Settings")
shopping_cart.update({
"enabled": 1,

View File

@ -53,9 +53,7 @@ def get_product_data(search=None, start=0, limit=12):
# order by
query += """ ORDER BY ranking desc, modified desc limit %s, %s""" % (cint(start), cint(limit))
return frappe.db.sql(query, {
"search": search
}, as_dict=1)
return frappe.db.sql(query, {"search": search}, as_dict=1) # nosemgrep
@frappe.whitelist(allow_guest=True)
def search(query):

View File

@ -56,30 +56,22 @@ def get_category_records(categories):
categorical_data = {}
for category in categories:
if category == "item_group":
categorical_data["item_group"] = frappe.db.sql("""
Select
name, parent_item_group, is_group, image, route
from
`tabItem Group`
where
parent_item_group = 'All Item Groups'
and show_in_website = 1
""",
as_dict=1)
categorical_data["item_group"] = frappe.db.get_all(
"Item Group",
filters={
"parent_item_group": "All Item Groups",
"show_in_website": 1
},
fields=["name", "parent_item_group", "is_group", "image", "route"],
as_dict=True
)
else:
doctype = frappe.unscrub(category)
fields = ["name"]
if frappe.get_meta(doctype, cached=True).get_field("image"):
fields += ["image"]
categorical_data[category] = frappe.db.sql(
f"""
Select
{",".join(fields)}
from
`tab{doctype}`
""",
as_dict=1)
categorical_data[category] = frappe.db.get_all(doctype, fields=fields, as_dict=True)
return categorical_data