mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-06-06 06:20:53 -04:00
4fef2861c6
This repeats the pattern we already have for previewing a letter,
where we assume the error is because the notification has already
been sent and redirect the user to see it.
I've improved the original pattern a bit:
- I've DRYed-up the low-level boto code and moved the error handler
there so it can be reused.
- I've introduced a custom exception, which the calling code can
choose to log.
- I've introduced the moto library, which we use elsewhere, to make
it easier to test S3 code.
I've used an error level log when sending a notification - now that
we have a more descriptive log, we can verify the assumption is true
and then make an informed decision to downgrade the log.
In future we may want to merge this handler with the similar code
in utils [1], but we'll need to be careful as the utils handler is
superficial - it doesn't check the reason for the error.
[1]: bce0f4e596/notifications_utils/s3.py (L52)
98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
import json
|
|
import urllib
|
|
|
|
import botocore
|
|
from boto3 import resource
|
|
from flask import current_app
|
|
from notifications_utils.s3 import s3upload as utils_s3upload
|
|
|
|
|
|
class LetterNotFoundError(Exception):
|
|
pass
|
|
|
|
|
|
def get_transient_letter_file_location(service_id, upload_id):
|
|
return 'service-{}/{}.pdf'.format(service_id, upload_id)
|
|
|
|
|
|
def backup_original_letter_to_s3(
|
|
data,
|
|
upload_id,
|
|
):
|
|
utils_s3upload(
|
|
filedata=data,
|
|
region=current_app.config['AWS_REGION'],
|
|
bucket_name=current_app.config['PRECOMPILED_ORIGINALS_BACKUP_LETTERS'],
|
|
file_location=f'{upload_id}.pdf',
|
|
)
|
|
|
|
|
|
def upload_letter_to_s3(
|
|
data,
|
|
*,
|
|
file_location,
|
|
status,
|
|
page_count,
|
|
filename,
|
|
message=None,
|
|
invalid_pages=None,
|
|
recipient=None
|
|
):
|
|
# Use of urllib.parse.quote encodes metadata into ascii, which is required by s3.
|
|
# Making sure data for displaying to users is decoded is taken care of by LetterMetadata
|
|
metadata = {
|
|
'status': status,
|
|
'page_count': str(page_count),
|
|
'filename': urllib.parse.quote(filename),
|
|
}
|
|
if message:
|
|
metadata['message'] = message
|
|
if invalid_pages:
|
|
metadata['invalid_pages'] = json.dumps(invalid_pages)
|
|
if recipient:
|
|
metadata['recipient'] = urllib.parse.quote(recipient)
|
|
|
|
utils_s3upload(
|
|
filedata=data,
|
|
region=current_app.config['AWS_REGION'],
|
|
bucket_name=current_app.config['TRANSIENT_UPLOADED_LETTERS'],
|
|
file_location=file_location,
|
|
metadata=metadata,
|
|
)
|
|
|
|
|
|
class LetterMetadata:
|
|
KEYS_TO_DECODE = ["filename", "recipient"]
|
|
|
|
def __init__(self, metadata):
|
|
self._metadata = metadata
|
|
|
|
def get(self, key, default=None):
|
|
value = self._metadata.get(key, default)
|
|
if value and key in self.KEYS_TO_DECODE:
|
|
value = urllib.parse.unquote(value)
|
|
return value
|
|
|
|
|
|
def get_letter_s3_object(service_id, file_id):
|
|
try:
|
|
file_location = get_transient_letter_file_location(service_id, file_id)
|
|
s3 = resource('s3')
|
|
return s3.Object(current_app.config['TRANSIENT_UPLOADED_LETTERS'], file_location).get()
|
|
except botocore.exceptions.ClientError as e:
|
|
if e.response['Error']['Code'] == 'NoSuchKey':
|
|
raise LetterNotFoundError(f'Letter not found for service {service_id} and file {file_id}')
|
|
|
|
raise
|
|
|
|
|
|
def get_letter_pdf_and_metadata(service_id, file_id):
|
|
s3_object = get_letter_s3_object(service_id, file_id)
|
|
pdf = s3_object['Body'].read()
|
|
return pdf, LetterMetadata(s3_object['Metadata'])
|
|
|
|
|
|
def get_letter_metadata(service_id, file_id):
|
|
s3_object = get_letter_s3_object(service_id, file_id)
|
|
return LetterMetadata(s3_object['Metadata'])
|