Example wrapping S3 to upload a file like object and generate pre-signed URL for this

s3.py

import boto3
from botocore.exceptions import ClientError
from django.conf import settings


class S3ConfigException(Exception):
    '''
    Raised when an issue is found during S3 client creation
    '''


class S3GeneratePreSignedUrlException(Exception):
    '''
    Raised when an issue is found generating a pre-signed URL for an S3 object
    '''


class S3UploadFileObjException(Exception):
    '''
    Raised when an issue is found during S3 file object upload
    '''


def create_presigned_url(bucket_name, object_key, expiration=60):
    """
    Generate a presigned URL to share an S3 object

    :param bucket_name: string
    :param object_key: string
    :param expiration: Time in seconds for the presigned URL to remain valid
    :return: Presigned URL as string, or raises exception when issue found
    """
    s3_client = _get_client()
    try:
        return s3_client.generate_presigned_url(
            'get_object',
            Params={'Bucket': bucket_name, 'Key': object_key},
            ExpiresIn=expiration
        )
    except (ClientError, S3ConfigException) as e:
        raise S3GeneratePreSignedUrlException("Error generating pre-signed URL for object") from e


def upload_file_object_to_s3(file_object, bucket_name, object_key):
    """
    Uploads a file like object to a specific key in named bucket

    :param file_object: a file like object (buffer or stream)
    :param bucket_name: string
    :param object_key: string
    :return: None, raises exception when issue found
    """
    try:
        s3_client = _get_client()
        s3_client.upload_fileobj(file_object, bucket_name, object_key)
    except (ClientError, S3ConfigException) as e:
        raise S3UploadFileObjException("Failed to upload '{}/{}'".format(bucket_name, object_key)) from e


def _get_client():
    """
    Returns boto3 client using credentials from settings
    """
    try:
        return boto3.client(
            's3',
            aws_access_key_id=settings.S3_ACCESS_KEY_ID,
            aws_secret_access_key=settings.S3_SECRET_ACCESS_KEY,
        )
    except ClientError as e:
        raise S3ConfigException("Failure to configure client with specified credentials") from e

test_s3.py

from unittest.mock import patch

from botocore.client import BaseClient
from botocore.exceptions import ClientError
from django.test import TestCase, override_settings

from common.s3 import (
    _get_client,
    create_presigned_url,
    S3ConfigException,
    S3GeneratePreSignedUrlException,
    S3UploadFileObjException,
    upload_file_object_to_s3,
)


class FakeS3Client:
    def __init__(self):
        self.upload_fileobj_called = False
        self.file_object = None
        self.bucket_name = None
        self.object_key = None

        self.generate_presigned_url_called = False
        self.ClientMethod = None
        self.Params = None
        self.ExpiresIn = None

    def upload_fileobj(self, file_object, bucket_name, object_key):
        self.upload_fileobj_called = True
        self.file_object = file_object
        self.bucket_name = bucket_name
        self.object_key = object_key

    def generate_presigned_url(self, ClientMethod, Params, ExpiresIn):
        self.generate_presigned_url_called = True
        self.ClientMethod = ClientMethod
        self.Params = Params
        self.ExpiresIn = ExpiresIn


class FakeS3ClientRaisesClientError:
    def __init__(self):
        self.upload_fileobj_called = False

    def upload_fileobj(self, file_object, bucket_name, object_key):
        raise ClientError({'Error': {'Code': 'code'}}, 'operation_name')

    def generate_presigned_url(self, ClientMethod, Params, ExpiresIn):
        raise ClientError({'Error': {'Code': 'code'}}, 'operation_name')


class FakeS3ClientRaisesS3ConfigException:
    def __init__(self):
        self.upload_fileobj_called = False

    def upload_fileobj(self, file_object, bucket_name, object_key):
        raise S3ConfigException('Uh oh')

    def generate_presigned_url(self, ClientMethod, Params, ExpiresIn):
        raise S3ConfigException('Uh oh')


class CreatePreSignedUrlTestCase(TestCase):

    @patch('common.s3._get_client')
    def test_raises_custom_error_when_get_client_errors(self, patched__get_client):
        fake_s3_client = FakeS3ClientRaisesS3ConfigException()
        patched__get_client.return_value = fake_s3_client

        with self.assertRaises(S3GeneratePreSignedUrlException):
            create_presigned_url('my-bucket-name', 'my-object-key')

    @patch('common.s3._get_client')
    def test_raises_custom_error_when_generate_presigned_url_errors(self, patched__get_client):
        fake_s3_client = FakeS3ClientRaisesClientError()
        patched__get_client.return_value = fake_s3_client

        with self.assertRaises(S3GeneratePreSignedUrlException):
            create_presigned_url('my-bucket-name', 'my-object-key')

    @patch('common.s3._get_client')
    def test_calls_generate_presigned_url(self, patched__get_client):
        fake_s3_client = FakeS3Client()
        patched__get_client.return_value = fake_s3_client
        self.assertFalse(fake_s3_client.generate_presigned_url_called)
        self.assertEqual(
            (fake_s3_client.ClientMethod, fake_s3_client.Params, fake_s3_client.ClientMethod),
            (None, None, None)
        )

        create_presigned_url('my-bucket-name', 'my-object-key')

        self.assertTrue(fake_s3_client.generate_presigned_url_called)
        self.assertEqual(
            (fake_s3_client.ClientMethod, fake_s3_client.Params, fake_s3_client.ExpiresIn),
            ('get_object', {'Bucket': 'my-bucket-name', 'Key': 'my-object-key'}, 60)
        )


class UploadFileObjectToS3TestCase(TestCase):

    @patch('common.s3._get_client')
    def test_raises_custom_error_when_get_client_errors(self, patched__get_client):
        fake_s3_client = FakeS3ClientRaisesS3ConfigException()
        patched__get_client.return_value = fake_s3_client

        with self.assertRaises(S3UploadFileObjException):
            upload_file_object_to_s3('file-object', 'my-bucket-name', 'my-object-key')

    @patch('common.s3._get_client')
    def test_raises_custom_error_when_upload_fileobj_errors(self, patched__get_client):
        fake_s3_client = FakeS3ClientRaisesClientError()
        patched__get_client.return_value = fake_s3_client

        with self.assertRaises(S3UploadFileObjException):
            upload_file_object_to_s3('file-object', 'my-bucket-name', 'my-object-key')

    @patch('common.s3._get_client')
    def test_calls_upload_fileobj(self, patched__get_client):
        fake_s3_client = FakeS3Client()
        patched__get_client.return_value = fake_s3_client
        self.assertFalse(fake_s3_client.upload_fileobj_called)
        self.assertEqual(
            (fake_s3_client.file_object, fake_s3_client.bucket_name, fake_s3_client.object_key),
            (None, None, None)
        )

        upload_file_object_to_s3('file-object', 'my-bucket-name', 'my-object-key')

        self.assertTrue(fake_s3_client.upload_fileobj_called)
        self.assertEqual(
            (fake_s3_client.file_object, fake_s3_client.bucket_name, fake_s3_client.object_key),
            ('file-object', 'my-bucket-name', 'my-object-key')
        )


class GetClientTestCase(TestCase):

    @patch('common.s3.boto3.client')
    def test_raises_custom_error_when_client_errors(self, patched_boto3_client):
        patched_boto3_client.side_effect = ClientError({'Error': {'Code': 'code'}}, 'operation_name')

        with self.assertRaises(S3ConfigException):
            _ = _get_client()

    @override_settings(S3_ACCESS_KEY_ID='my-s3-access-key', S3_SECRET_ACCESS_KEY='my-s3-secret-key')
    def test_returns_client_as_expected(self):
        try:
            client = _get_client()
            self.assertIsInstance(client, BaseClient)
        except Exception as ex:
            self.fail("_get_client() raised Exception unexpectedly: {}".format(ex))