0

Boto3 is the Amazon Web Services (AWS) Software Development Kit (SDK) for Python, which allows Python developers to write software that makes use of services like Amazon S3 and Amazon EC2. You can find the latest, most up to date, documentation at our doc site, including a list of services that are supported.

First, install the library and set a default region:

$ pip install boto3

Next, set up credentials (in e.g. ~/.aws/credentials):

[default]
aws_access_key_id = YOUR_KEY
aws_secret_access_key = YOUR_SECRET

In your settings.py

AWS_S3_BUCKET_NAME = 'your_bucket_name'

and then, here the script to upload file and images into amazon s3, for example in your yourapp/utils/uploader.py:

import os
import boto3
import urllib3
import datetime

from django.conf import settings
from django.core.files.base import ContentFile
from botocore.exceptions import ClientError


class AmazonS3(object):
    path_format = datetime.datetime.now().strftime('%Y/%m/%d')
    bucket_name = getattr(settings, 'AWS_S3_BUCKET_NAME')
    client = boto3.client('s3')

    def get_object_name(self, file_url):
        """
        function to get the file object name
        or file path url by `file_url`.

        :param `file_url` is string uploaded url.
                eg: 'https://s3-ap-southeast-1.amazonaws.com/assets.dev.doain/files/2019/08/13/no-image.svg'

        :return string object_name.
                eg: 'files/2019/07/22/logo.png'
        """
        return urllib3.util.parse_url(file_url).path[1:]

    def get_uploaded_url(self, object_name):
        """
        function to get the uploaded url.

        :param `object_name` is string object name.
                eg: 'files/2019/07/22/logo.png'

        :return string of uploaded file/images url.
                eg: 'https://s3-ap-southeast-1.amazonaws.com/assets.dev.doain/files/2019/08/13/no-image.svg'
        """
        location = self.client.get_bucket_location(Bucket=self.bucket_name)['LocationConstraint']
        return 'https://s3-%s.amazonaws.com/%s/%s' % (location, self.bucket_name, object_name)

    def upload_file(self, file_name, folder='files'):
        """
        function to upload the file or image into aws s3.

        :param `file_name` is file name original path source.
        :param `folder` is target folder in aws s3.

        :return `url` eg: 'https://s3-ap-southeast-1.amazonaws.com/assets.dev.doain/files/2019/08/13/no-image.svg'
        """
        object_name = os.path.join(folder, self.path_format, file_name.split('/')[-1])
        res = self.client.upload_file(file_name, self.bucket_name, object_name,
                                      ExtraArgs={'ACL': 'public-read'})
        return self.get_uploaded_url(object_name)

    def upload_fileobj(self, file_obj, folder='files'):
        """
        function to upload the file object or images into aws s3.

        :param `field_obj` is file object, eg: request.FILES['image']
        :param `folder` is target folder in aws s3.

        :return `url` eg: 'https://s3-ap-southeast-1.amazonaws.com/assets.dev.doain/files/2019/08/13/no-image.svg'
        """
        object_name = os.path.join(folder, self.path_format, file_obj.name.split('/')[-1])
        file_data = ContentFile(file_obj.read())
        self.client.upload_fileobj(file_data, self.bucket_name, object_name,
                                   ExtraArgs={'ACL': 'public-read'})
        return self.get_uploaded_url(object_name)

For example, in this case I using django rest framework.
In yourapp/views.py:

from django.conf import settings
from django.utils.translation import ugettext_lazy as _

from rest_framework import status
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import AllowAny

from yourapp.utils.uploader import AmazonS3


class UploadImageView(APIView):
    permission_classes = [AllowAny,]
    image_types = ['image/png', 'image/jpg', 'image/jpeg',
                   'image/pjpeg', 'image/gif']
    max_size = getattr(settings, 'MAX_IMAGE_UPLOAD_SIZE', 5242880)
    s3 = AmazonS3()

    def post(self, request, *args, **kwargs):
        """
        curl --location --request POST "http://127.0.0.1:8000/api/uploader/image/" \
             --header "Accept-Language: en" \
             --form "[email protected]" \
             --form "folder=images"
        """
        folder = request.POST.get('folder', 'images')

        if 'image' in request.FILES:
            image = request.FILES['image']
            if image.content_type not in self.image_types:
                response = {'status': status.HTTP_400_BAD_REQUEST,
                            'detail': _('Bad image format!')}
                return Response(response, status=response.get('status'))

            if image.size > self.max_size:
                max_size = self.max_size / (1024 * 1024)
                response = {'status': status.HTTP_400_BAD_REQUEST,
                            'detail': _('Maximum image file is %(max_size)s MB.') % {'max_size': max_size}}
                return Response(response, status=response.get('status'))

            image_url = self.s3.upload_fileobj(image, folder=folder)
            if not image_url:
                response = {'status': status.HTTP_400_BAD_REQUEST,
                            'detail': _('Failed to upload the image.')}
                return Response(response, status=response.get('status'))

            response = {'status': status.HTTP_200_OK,
                        'image_url': image_url,
                        'image_name': image.name,
                        'detail': _('Image successfully uploaded.')}
            return Response(response, status=response.get('status'))

        response = {'status': status.HTTP_400_BAD_REQUEST,
                    'detail': _('Invalid "image" key.')}
        return Response(response, status=response.get('status'))


class UploadFileView(APIView):
    permission_classes = [AllowAny,]
    max_size = getattr(settings, 'MAX_FILE_UPLOAD_SIZE', 10485760)
    s3 = AmazonS3()

    def post(self, request, *args, **kwargs):
        """
        curl --location --request POST "http://127.0.0.1:8000/api/uploader/file/" \
             --header "Accept-Language: en" \
             --form "[email protected]" \
             --form "folder=files"
        """
        folder = request.POST.get('folder', 'files')

        if 'file' in request.FILES:
            file = request.FILES['file']
            if file.size > self.max_size:
                max_size = self.max_size / (1024 * 1024)
                response = {'status': status.HTTP_400_BAD_REQUEST,
                            'detail': _('Maximum file is %(max_size)s MB.') % {'max_size': max_size}}
                return Response(response, status=response.get('status'))

            file_url = self.s3.upload_fileobj(file, folder=folder)
            if not file_url:
                response = {'status': status.HTTP_400_BAD_REQUEST,
                            'detail': _('Failed to upload the file.')}
                return Response(response, status=response.get('status'))

            response = {'status': status.HTTP_200_OK,
                        'file_url': file_url,
                        'file_name': file.name,
                        'detail': _('File successfully uploaded.')}
            return Response(response, status=response.get('status'))

        response = {'status': status.HTTP_400_BAD_REQUEST,
                    'detail': _('Invalid "file" key.')}
        return Response(response, status=response.get('status'))
aws solution module api django

Your Answer

blog comments powered by Disqus