notifications-admin/app/s3_client/s3_letter_upload_client.py

import json

from boto3 import resource
from flask import current_app
from notifications_utils.s3 import s3upload as utils_s3upload
from notifications_utils.sanitise_text import SanitiseASCII


def get_transient_letter_file_location(service_id, upload_id):
    return 'service-{}/{}.pdf'.format(service_id, upload_id)


def upload_letter_to_s3(
    data,
    *,
    file_location,
    status,
    page_count,
    filename,
    message=None,
    invalid_pages=None,
    recipient=None
):
    metadata = {
        'status': status,
        'page_count': str(page_count),
        'filename': filename,
    }
    if message:
        metadata['message'] = message
    if invalid_pages:
        metadata['invalid_pages'] = json.dumps(invalid_pages)
    if recipient:
        metadata['recipient'] = format_recipient(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,
    )


def get_letter_pdf_and_metadata(service_id, file_id):
    file_location = get_transient_letter_file_location(service_id, file_id)
    s3 = resource('s3')
    s3_object = s3.Object(current_app.config['TRANSIENT_UPLOADED_LETTERS'], file_location).get()

    pdf = s3_object['Body'].read()
    metadata = s3_object['Metadata']

    return pdf, metadata


def get_letter_metadata(service_id, file_id):
    file_location = get_transient_letter_file_location(service_id, file_id)
    s3 = resource('s3')
    s3_object = s3.Object(current_app.config['TRANSIENT_UPLOADED_LETTERS'], file_location).get()

    return s3_object['Metadata']


def format_recipient(address):
    '''
    To format the recipient we need to:
    - remove new line characters
    - remove whitespace around the lines
    - join the address lines, separated by a comma
    - convert the string to ASCII (S3 metadata must be stored as ASCII)
    '''
    stripped_address_lines_no_trailing_commas = [
        line.lstrip().rstrip(' ,')
        for line in address.splitlines() if line
    ]
    one_line_address = ', '.join(stripped_address_lines_no_trailing_commas)

    return SanitiseASCII.encode(one_line_address)