S3 Pregsigned Url

1. S3 이미지 업로드 Flow

서버로 presigned url 요청

서버는 s3로부터 presigned url 생성

생성된 presigned url을 클라이언트에게 전달

클라이언트는 받은 presigned url로 직접 s3에 이미지 업로드

업로드 후 이미지 URL 주소를 서버에 전달

전달받은 url 주소 (string) 그 자체를 모델의 Image field에 저장

2. Django Server Code

presigned url 생성 및 전달 코드

import logging
import uuid

import boto3
from botocore.exceptions import ClientError
from rest_framework import serializers

logger = logging.getLogger(__name__)


class PresignedURLSerializer(serializers.Serializer):
    # name = serializers.CharField(write_only=True)
    # content_type = serializers.CharField(write_only=True)
    url = serializers.ListSerializer(read_only=True, child=serializers.URLField(read_only=True))
    image_list = serializers.ListSerializer(write_only=True, child=serializers.CharField(write_only=True))

    def validate(self, attrs):
        url_list = []

        for image in attrs.get("image_list"):
            url = self.create_presigned_url(image=image)
            url_list.append(url)

        attrs["url"] = url_list
        return attrs

    @staticmethod
    def create_presigned_url(
        image, content_type="images", aws_access_key_id=None, aws_secret_access_key=None, expiration=3600
    ):
        s3_client = boto3.client(
            "s3",
            region_name="ap-northeast-2",
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
        )
        try:

			# 프론트로부터 받은 "이미지 이름.이미지 확장자" 
            # formData가 아닌 string
            name = image.split(".")
            object_key = "/".join([f"media/{content_type}", f"{str(uuid.uuid4())}.{name[-1]}"])
            
            url = s3_client.generate_presigned_url(
                "put_object",
                Params={
                    # "Bucket": settings.AWS_STORAGE_BUCKET_NAME,
                    "Bucket": "my-s3-bucket",
                    "Key": object_key,
                },
                ExpiresIn=expiration,
            )
            logger.info("Got presigned URL: %s", url)

        except ClientError as e:
            logging.error(e)
            return None

        return url

    def create(self, validated_data):
        logger.info("S3 Presigned Url")
        logger.info(validated_data)
        return validated_data

이미지 저장 코드

...
...
    @transaction.atomic
    def create(self, validated_data):
        today = datetime.datetime.now()

        images = validated_data.pop("images_list", None)
        logged_in_user = validated_data.get("user")
        print(f"images : {images}")
        try:
            with transaction.atomic():

                if images:
                    UserProfile.objects.bulk_create(
                        [UserProfile(user=user, ord=idx, image_url=image) for idx, image in enumerate(images)]
                    )