mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-05-05 08:31:00 -04:00
Use Spreadsheet class to generate example CSVs
This means not repeating a bunch of `with StringIO` blocks all over the place.
This commit is contained in:
@@ -112,7 +112,7 @@ def send_messages(service_id, template_id):
|
||||
try:
|
||||
upload_id = s3upload(
|
||||
service_id,
|
||||
Spreadsheet.from_file(form.file.data.filename, form.file.data).as_dict,
|
||||
Spreadsheet.from_file(form.file.data, filename=form.file.data.filename).as_dict,
|
||||
current_app.config['AWS_REGION']
|
||||
)
|
||||
session['upload_data'] = {
|
||||
@@ -142,16 +142,13 @@ def send_messages(service_id, template_id):
|
||||
@user_has_permissions('send_texts', 'send_emails', 'send_letters', 'manage_templates', any_=True)
|
||||
def get_example_csv(service_id, template_id):
|
||||
template = Template(service_api_client.get_service_template(service_id, template_id)['data'])
|
||||
with io.StringIO() as output:
|
||||
writer = csv.writer(output)
|
||||
writer.writerows([
|
||||
[first_column_heading[template.template_type]] + list(template.placeholders),
|
||||
get_example_csv_rows(template)
|
||||
])
|
||||
return output.getvalue(), 200, {
|
||||
'Content-Type': 'text/csv; charset=utf-8',
|
||||
'Content-Disposition': 'inline; filename="{}.csv"'.format(template.name)
|
||||
}
|
||||
return Spreadsheet.from_rows([
|
||||
[first_column_heading[template.template_type]] + list(template.placeholders),
|
||||
get_example_csv_rows(template)
|
||||
]).as_csv_data, 200, {
|
||||
'Content-Type': 'text/csv; charset=utf-8',
|
||||
'Content-Disposition': 'inline; filename="{}.csv"'.format(template.name)
|
||||
}
|
||||
|
||||
|
||||
@main.route("/services/<service_id>/send/<template_id>/test", methods=['GET', 'POST'])
|
||||
@@ -167,31 +164,28 @@ def send_test(service_id, template_id):
|
||||
)
|
||||
|
||||
if len(template.placeholders) == 0 or request.method == 'POST':
|
||||
with io.StringIO() as output:
|
||||
writer = csv.writer(output)
|
||||
writer.writerows([
|
||||
[first_column_heading[template.template_type]] + list(template.placeholders),
|
||||
get_example_csv_rows(template, use_example_as_example=False, submitted_fields=request.form)
|
||||
])
|
||||
upload_id = s3upload(
|
||||
service_id,
|
||||
{
|
||||
'file_name': file_name,
|
||||
'data': output.getvalue()
|
||||
},
|
||||
current_app.config['AWS_REGION']
|
||||
)
|
||||
session['upload_data'] = {
|
||||
"template_id": template_id,
|
||||
"original_file_name": file_name
|
||||
}
|
||||
return redirect(url_for(
|
||||
'.check_messages',
|
||||
upload_id=upload_id,
|
||||
service_id=service_id,
|
||||
template_type=template.template_type,
|
||||
from_test=True
|
||||
))
|
||||
upload_id = s3upload(
|
||||
service_id,
|
||||
{
|
||||
'file_name': file_name,
|
||||
'data': Spreadsheet.from_rows([
|
||||
[first_column_heading[template.template_type]] + list(template.placeholders),
|
||||
get_example_csv_rows(template, use_example_as_example=False, submitted_fields=request.form)
|
||||
]).as_csv_data
|
||||
},
|
||||
current_app.config['AWS_REGION']
|
||||
)
|
||||
session['upload_data'] = {
|
||||
"template_id": template_id,
|
||||
"original_file_name": file_name
|
||||
}
|
||||
return redirect(url_for(
|
||||
'.check_messages',
|
||||
upload_id=upload_id,
|
||||
service_id=service_id,
|
||||
template_type=template.template_type,
|
||||
from_test=True
|
||||
))
|
||||
|
||||
return render_template(
|
||||
'views/send-test.html',
|
||||
|
||||
32
app/utils.py
32
app/utils.py
@@ -143,7 +143,7 @@ class Spreadsheet():
|
||||
|
||||
allowed_file_extensions = ['csv', 'xlsx', 'xls', 'ods', 'xlsm', 'tsv']
|
||||
|
||||
def __init__(self, filename, csv_data):
|
||||
def __init__(self, csv_data, filename=''):
|
||||
self.filename = filename
|
||||
self.as_csv_data = csv_data
|
||||
self.as_dict = {
|
||||
@@ -164,24 +164,28 @@ class Spreadsheet():
|
||||
return '\r\n'.join(file_content.getvalue().decode('utf-8').splitlines())
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, filename, file_content):
|
||||
def from_rows(cls, rows, filename=''):
|
||||
|
||||
with StringIO() as converted:
|
||||
output = csv.writer(converted)
|
||||
|
||||
for row in rows:
|
||||
output.writerow(row)
|
||||
|
||||
return cls(converted.getvalue(), filename)
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, file_content, filename=''):
|
||||
|
||||
extension = cls.get_extension(filename)
|
||||
|
||||
if extension == 'csv':
|
||||
return cls(filename, Spreadsheet.normalise_newlines(file_content))
|
||||
return cls(Spreadsheet.normalise_newlines(file_content), filename)
|
||||
|
||||
if extension == 'tsv':
|
||||
file_content = StringIO(Spreadsheet.normalise_newlines(file_content))
|
||||
|
||||
with StringIO() as converted:
|
||||
|
||||
output = csv.writer(converted)
|
||||
|
||||
for row in pyexcel.get_sheet(
|
||||
file_type=extension,
|
||||
file_content=file_content.getvalue()
|
||||
).to_array():
|
||||
output.writerow(row)
|
||||
|
||||
return cls(filename, converted.getvalue())
|
||||
return cls.from_rows(pyexcel.get_sheet(
|
||||
file_type=extension,
|
||||
file_content=file_content.getvalue()
|
||||
).to_array(), filename)
|
||||
|
||||
Reference in New Issue
Block a user