refactor: extract common methods

frappe/desk/query_report.py

@@ -5,7 +5,6 @@
 import json
 import os
 from datetime import timedelta
-from io import StringIO
 
 import frappe
 import frappe.desk.reportview
@@ -320,7 +319,10 @@ def get_report_data(doc, data):
 @frappe.whitelist()
 def export_query():
 	"""export from query reports"""
+	from frappe.desk.utils import get_csv_bytes, pop_csv_params, provide_binary_file
+
 	form_params = frappe._dict(frappe.local.form_dict)
+	csv_params = pop_csv_params(form_params)
 	clean_params(form_params)
 	parse_json(form_params)
 
@@ -351,29 +353,15 @@ def export_query():
 	xlsx_data, column_widths = build_xlsx_data(data, visible_idx, include_indentation)
 
 	if file_format_type == "CSV":
-		import csv
-
-		file = StringIO()
-		writer = csv.writer(
-			file,
-			quoting=csv.QUOTE_NONNUMERIC
-			if form_params.csv_quoting is None
-			else cint(form_params.csv_quoting),
-			delimiter=cstr(form_params.csv_delimiter or ","),
-		)
-		writer.writerows(xlsx_data)
-		content = file.getvalue().encode("utf-8")
+		content = get_csv_bytes(xlsx_data, csv_params)
 		file_extension = "csv"
 	elif file_format_type == "Excel":
 		from frappe.utils.xlsxutils import make_xlsx
 
-		file = make_xlsx(xlsx_data, "Query Report", column_widths=column_widths)
 		file_extension = "xlsx"
-		content = file.getvalue()
+		content = make_xlsx(xlsx_data, "Query Report", column_widths=column_widths).getvalue()
 
-	frappe.response["filename"] = f"{report_name}.{file_extension}"
-	frappe.response["filecontent"] = content
-	frappe.response["type"] = "binary"
+	provide_binary_file(report_name, file_extension, content)
 
 
 def format_duration_fields(data: frappe._dict) -> None:

frappe/desk/reportview.py

@@ -4,7 +4,6 @@
 """build query for doclistview and return results"""
 
 import json
-from io import StringIO
 
 import frappe
 import frappe.permissions
@@ -14,7 +13,7 @@
 from frappe.model.base_document import get_controller
 from frappe.model.db_query import DatabaseQuery
 from frappe.model.utils import is_virtual_doctype
-from frappe.utils import add_user_info, cint, cstr, format_duration
+from frappe.utils import add_user_info, cint, format_duration
 from frappe.utils.data import sbool
 
 
@@ -347,6 +346,8 @@ def delete_report(name):
 @frappe.read_only()
 def export_query():
 	"""export from report builder"""
+	from frappe.desk.utils import get_csv_bytes, pop_csv_params, provide_binary_file
+
 	title = frappe.form_dict.title
 	frappe.form_dict.pop("title", None)
 
@@ -357,11 +358,7 @@ def export_query():
 	add_totals_row = None
 	file_format_type = form_params["file_format_type"]
 	title = title or doctype
-	if file_format_type == "CSV":
-		csv_delimiter = cstr(form_params.get("csv_delimiter", ","))
-		csv_quoting = cint(form_params.get("csv_quoting", 2))
-		del form_params["csv_delimiter"]
-		del form_params["csv_quoting"]
+	csv_params = pop_csv_params(form_params)
 
 	del form_params["doctype"]
 	del form_params["file_format_type"]
@@ -390,37 +387,26 @@ def export_query():
 	if add_totals_row:
 		ret = append_totals_row(ret)
 
-	data = [[_("Sr"), *get_labels(db_query.fields, doctype)]]
-	for i, row in enumerate(ret):
-		data.append([i + 1, *list(row)])
-
+	data = [[_("Sr")] + get_labels(db_query.fields, doctype)].extend(
+		[i + 1] + list(row) for i, row in enumerate(ret)
+	)
 	data = handle_duration_fieldtype_values(doctype, data, db_query.fields)
 
 	if file_format_type == "CSV":
-		# convert to csv
-		import csv
-
 		from frappe.utils.xlsxutils import handle_html
 
-		f = StringIO()
-		writer = csv.writer(f, quoting=csv_quoting, delimiter=csv_delimiter)
-		for r in data:
-			# encode only unicode type strings and not int, floats etc.
-			writer.writerow([handle_html(frappe.as_unicode(v)) if isinstance(v, str) else v for v in r])
-
-		f.seek(0)
-		frappe.response["result"] = cstr(f.read())
-		frappe.response["type"] = "csv"
-		frappe.response["doctype"] = title
-
+		file_extension = "csv"
+		content = get_csv_bytes(
+			[[handle_html(frappe.as_unicode(v)) if isinstance(v, str) else v for v in r] for r in data],
+			csv_params,
+		)
 	elif file_format_type == "Excel":
 		from frappe.utils.xlsxutils import make_xlsx
 
-		xlsx_file = make_xlsx(data, doctype)
+		file_extension = "xlsx"
+		content = make_xlsx(data, doctype).getvalue()
 
-		frappe.response["filename"] = _(title) + ".xlsx"
-		frappe.response["filecontent"] = xlsx_file.getvalue()
-		frappe.response["type"] = "binary"
+	provide_binary_file(title, file_extension, content)
 
 
 def append_totals_row(data):

frappe/desk/utils.py

@@ -25,3 +25,34 @@ def validate_route_conflict(doctype, name):
 
 def slug(name):
 	return name.lower().replace(" ", "-")
+
+
+def pop_csv_params(form_dict):
+	"""Pop csv params from form_dict and return them as a dict."""
+	from csv import QUOTE_NONNUMERIC
+
+	from frappe.utils.data import cint, cstr
+
+	return {
+		"delimiter": cstr(form_dict.pop("csv_delimiter", ","))[0],
+		"quoting": cint(form_dict.pop("csv_quoting", QUOTE_NONNUMERIC)),
+	}
+
+
+def get_csv_bytes(data: list[list], csv_params: dict) -> bytes:
+	"""Convert data to csv bytes."""
+	from csv import writer
+	from io import StringIO
+
+	file = StringIO()
+	csv_writer = writer(file, **csv_params)
+	csv_writer.writerows(data)
+
+	return file.getvalue().encode("utf-8")
+
+
+def provide_binary_file(filename: str, extension: str, content: bytes) -> None:
+	"""Provide a binary file to the client."""
+	frappe.response["type"] = "binary"
+	frappe.response["filecontent"] = content
+	frappe.response["filename"] = f"{filename}.{extension}"