mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-03-10 05:12:52 -04:00
74c1648c8d
* Fix for lazy loading while checking if file exists * Fixed linting errors * Fixed for static scan
84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
import os
|
|
import uuid
|
|
|
|
from flask import current_app
|
|
|
|
from app.s3_client import (
|
|
check_s3_file_exists,
|
|
get_s3_contents,
|
|
get_s3_metadata,
|
|
get_s3_object,
|
|
set_s3_metadata,
|
|
)
|
|
from notifications_utils.s3 import s3upload as utils_s3upload
|
|
|
|
NEW_FILE_LOCATION_STRUCTURE = "{}-service-notify/{}.csv"
|
|
|
|
|
|
def get_csv_location(service_id, upload_id):
|
|
bucket = current_app.config["CSV_UPLOAD_BUCKET"]["bucket"]
|
|
key = NEW_FILE_LOCATION_STRUCTURE.format(service_id, upload_id)
|
|
region = current_app.config["CSV_UPLOAD_BUCKET"]["region"]
|
|
|
|
return (
|
|
bucket,
|
|
key,
|
|
current_app.config["CSV_UPLOAD_BUCKET"]["access_key_id"],
|
|
current_app.config["CSV_UPLOAD_BUCKET"]["secret_access_key"],
|
|
region,
|
|
)
|
|
|
|
|
|
def get_csv_upload(service_id, upload_id):
|
|
return get_s3_object(*get_csv_location(service_id, upload_id))
|
|
|
|
|
|
def remove_blank_lines(filedata):
|
|
# sometimes people upload files with hundreds of blank lines at the end
|
|
data = filedata["data"]
|
|
cleaned_data = "\r\n".join(line for line in data.splitlines() if line.strip())
|
|
filedata["data"] = cleaned_data
|
|
return filedata
|
|
|
|
|
|
def s3upload(service_id, filedata):
|
|
|
|
filedata = remove_blank_lines(filedata)
|
|
upload_id = str(uuid.uuid4())
|
|
bucket_name, file_location, access_key, secret_key, region = get_csv_location(
|
|
service_id, upload_id
|
|
)
|
|
if bucket_name == "":
|
|
exp_bucket = current_app.config["CSV_UPLOAD_BUCKET"]["bucket"]
|
|
exp_region = current_app.config["CSV_UPLOAD_BUCKET"]["region"]
|
|
tier = os.getenv("NOTIFY_ENVIRONMENT")
|
|
raise Exception(
|
|
f"NO BUCKET NAME SHOULD BE: {exp_bucket} WITH REGION {exp_region} TIER {tier}"
|
|
)
|
|
|
|
utils_s3upload(
|
|
filedata=filedata["data"],
|
|
region=region,
|
|
bucket_name=bucket_name,
|
|
file_location=file_location,
|
|
access_key=access_key,
|
|
secret_key=secret_key,
|
|
)
|
|
return upload_id
|
|
|
|
|
|
def s3download(service_id, upload_id):
|
|
return get_s3_contents(get_csv_upload(service_id, upload_id))
|
|
|
|
|
|
def set_metadata_on_csv_upload(service_id, upload_id, **kwargs):
|
|
return set_s3_metadata(get_csv_upload(service_id, upload_id), **kwargs)
|
|
|
|
|
|
def get_csv_metadata(service_id, upload_id):
|
|
return get_s3_metadata(get_csv_upload(service_id, upload_id))
|
|
|
|
|
|
def check_s3_report_exists(service_id, upload_id):
|
|
return check_s3_file_exists(get_csv_upload(service_id, upload_id))
|