오늘 학습 내용 ✅

qna 전용 이미지 처리

• 원래 개별적으로 s3_client.py를 내가 이용하는 app에서 사용하고 있었는데
• 원래 사용하던 명령어 몇개를 빼고 기본적인 명령어만 남겨놓고 core로 통합되어 다시 구성 시작

---

apps/core/utils/s3_client.py

class S3Client:
    def __init__(self) -> None:
        aws_access_key_id = getattr(settings, "AWS_S3_ACCESS_KEY_ID", None)
        aws_secret_access_key = getattr(settings, "AWS_S3_SECRET_ACCESS_KEY", None)
        aws_region = getattr(settings, "AWS_S3_REGION", "ap-northeast-2")
        self.bucket_name = getattr(settings, "AWS_S3_BUCKET_NAME", "my-bucket")

        self.s3: BotoS3Client = boto3.client(
            "s3",
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
            region_name=aws_region,
        )

• __init__(self): S3 연결 설정 및 초기화
  • 이 클래스가 생성될 때 가장 먼저 실행되는 부분
  • Django의 settings.py에서 AWS 관련 설정값을 가져와 연결함
• getattr(settings, "키", "기본값")
  • Django 설정 파일(settings.py)에서 값을 안전하게 가져옴
  • 만약 설정 파일에 해당 키가 없으면 "기본값"을 사용하거나 None을 반환하여 에러를 방지함
• boto3.client("s3", ...)
  • AWS S3와 통신할 수 있는 클라이언트 객체를 생성함
  • 여기서 생성된 self.s3 객체를 통해 이후의 모든 작업을 수행함

---

    def upload(self, file: Any, path_prefix: str = "", extra_args: Optional[Dict[str, Any]] = None) -> str:
        original_name = getattr(file, "name", "unknown_file")
        ext = original_name.split(".")[-1] if "." in original_name else "bin"

        file_name = f"{uuid.uuid4()}.{ext}"

        clean_prefix = path_prefix.rstrip("/")
        key = f"{clean_prefix}/{file_name}" if clean_prefix else file_name
        key = key.lstrip("/")

        upload_params: Dict[str, Any] = extra_args.copy() if extra_args else {}

        if "ContentType" not in upload_params:
            content_type = getattr(file, "content_type", None)
            if content_type:
                upload_params["ContentType"] = content_type

        try:
            self.s3.upload_fileobj(file, self.bucket_name, key, ExtraArgs=upload_params)
            return key
        except ClientError as e:
            logger.error(f"S3 Upload Error: {e}", exc_info=True)
            raise e

• upload(self, file, ...) - 서버에서 직접 파일 업로드

  • 백엔드 서버가 파일 객체(file)를 가지고 있을 때, 이를 S3로 전송하는 메서드
• uuid.uuid4()
  • 파일 이름 중복을 막기 위해 사용
  • 사용자가 올린 파일명(예: a.jpg)을 그대로 쓰면 덮어씌워질 위험이 있으므로
    • 랜덤한 고유 문자열(UUID)로 파일명을 변경
• getattr(file, "content_type", None)
  • 파일의 타입을 알아냄(image/jpeg, application/pdf)
  • 이를 S3에 같이 저장해야 브라우저에서 이미지를 열었을 때 다운로드가 되지 않고 바로 보임
• self.s3.upload_fileobj(...)
  • 파일을 메모리에 통째로 올리지 않고, 스트림(stream) 방식으로 효율적으로 S3 버킷에 업로드
  • ExtraArgs: 파일의 메타데이터(Content-Type 등)를 함께 설정할 때 사용

---

    def delete(self, key: str) -> None:
        if not key:
            return
        try:
            self.s3.delete_object(Bucket=self.bucket_name, Key=key)
        except ClientError as e:
            logger.warning(f"S3 Delete Failed (Key: {key}): {e}", exc_info=True)

• delete(self, key) - 파일 삭제
• self.s3.delete_object(Bucket=..., Key=key)
  • 지정한 버킷에서 해당 Key(파일 경로)를 가진 객체를 삭제
• try...except ClientError
  • 삭제 과정에서 AWS 통신 에러가 발생해도 서버가 멈추지 않도록 예외 처리
    • 삭제 실패는 로그만 남기고 넘어감

---

    def build_url(self, key: str) -> str:
        if not key:
            return ""

        custom_domain = getattr(settings, "AWS_S3_CUSTOM_DOMAIN", None)

        if custom_domain:
            domain = custom_domain
        else:
            region = getattr(settings, "AWS_S3_REGION", "ap-northeast-2")
            domain = f"{self.bucket_name}.s3.{region}.amazonaws.com"

        return f"https://{domain.rstrip('/')}/{key.lstrip('/')}"

• build_url(self, key) - 파일 접근 URL 생성
  • 저장된 파일의 Key를 가지고 실제 웹에서 접근 가능한 https://... 형태의 주소를 만듬
• AWS_S3_CUSTOM_DOMAIN 확인
  • 기본 AWS S3 도메인(bucket.s3.region.amazonaws.com)을 사용
• f"https://{domain}/{key}"
  • 최종적으로 도메인과 파일 경로를 합쳐서 전체 URL 문자열을 반환

---

    def generate_presigned_url(self, key: str, expires_in: int = 3600) -> str:
        try:
            url = self.s3.generate_presigned_url(
                ClientMethod="put_object", Params={"Bucket": self.bucket_name, "Key": key}, ExpiresIn=expires_in
            )
            return url
        except ClientError as e:
            logger.error(f"Failed to generate presigned URL (Key: {key}): {e}", exc_info=True)
            raise e

• generate_presigned_url(self, key, ...) - 프론트엔드 업로드용 임시 URL 발급
  • 서버를 거치지 않고
  • 프론트엔드(클라이언트)가 S3로 파일을 직접 업로드할 수 있는 권한이 담긴 임시 URL을 만듬
• self.s3.generate_presigned_url(...)
  • ClientMethod="put_object"
    • 이 URL은 파일을 '업로드(PUT)' 하는 용도라고 명시
  • ExpiresIn
    • 이 URL의 유효 시간을 설정(기본 1시간). 시간이 지나면 이 URL로는 업로드할 수 없음

• 역할

  • 이 메서드가 만든 URL을 프론트엔드에 주면, 프론트엔드는 그 주소로 이미지를 바로 쏘아 올림
  • 덕분에 백엔드 서버의 부하가 크게 줄어듬

---

qna/services/common/image_service.py

• core의 S3Client를 사용하여 이미지 동기화(삭제/추가) 로직을 수정
• URL에서 Key를 추출하는 로직을 추가하여 core의 delete(key) 메서드와 호환을 목표로 함

from django.db import transaction
from apps.core.utils.s3_client import S3Client
from apps.qna.models import Question, QuestionImage

from apps.qna.utils.content_image_parser import extract_image_urls_from_content
from apps.qna.utils.s3_utils import extract_key_from_url, is_valid_s3_url

def sync_question_images(question: Question, content: str) -> None:
    """
    본문(Content)에 포함된 이미지 URL을 추출하여 DB 및 S3와 동기화(삭제 및 추가)를 수행합니다.
    """
    s3_client = S3Client()

    # 1. 본문 파싱
    current_urls_in_content = set(extract_image_urls_from_content(content))

    # 2. DB 상태 확인
    existing_images_qs = QuestionImage.objects.filter(question=question)
    existing_images_map = {img.img_url: img for img in existing_images_qs}
    existing_urls = set(existing_images_map.keys())

    # 3. 비교 (Diff)
    urls_to_delete = existing_urls - current_urls_in_content
    urls_to_add = current_urls_in_content - existing_urls

    # 4. 삭제 처리
    if urls_to_delete:
        existing_images_qs.filter(img_url__in=urls_to_delete).delete()

        def delete_s3_files():
            for url in urls_to_delete:
                key = extract_key_from_url(url)
                if key:
                    s3_client.delete(key)

        transaction.on_commit(delete_s3_files)

    # 5. 추가 처리
    new_images = []
    for url in urls_to_add:
        if is_valid_s3_url(url):
            new_images.append(QuestionImage(question=question, img_url=url))

    if new_images:
        QuestionImage.objects.bulk_create(new_images)

---

qna/views/common/presigned_url_view.py

import uuid
from drf_spectacular.utils import extend_schema
from rest_framework import status
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView

from apps.core.constants import ANSWER_IMAGE_UPLOAD_PATH, QUESTION_IMAGE_UPLOAD_PATH
from apps.core.utils.s3_client import S3Client  #
from apps.qna.serializers.common.presigned_url_serializer import (
    PresignedUploadSerializer,
)


class PresignedUploadAPIView(APIView):
    permission_classes = [IsAuthenticated]
    ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}

    @extend_schema(request=PresignedUploadSerializer)
    def post(self, request):
        # 1. 시리얼라이저 검증
        serializer = PresignedUploadSerializer(data=request.data)
        if not serializer.is_valid():
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

        original_name = serializer.validated_data["file_name"]
        upload_type = serializer.validated_data["upload_type"]

        # 2. 확장자 검증
        ext = original_name.split(".")[-1].lower() if "." in original_name else ""
        if ext not in self.ALLOWED_EXTENSIONS:
            return Response(
                {"error": f"지원하지 않는 파일 형식입니다. ({', '.join(self.ALLOWED_EXTENSIONS)} 만 가능)"},
                status=status.HTTP_400_BAD_REQUEST,
            )

        new_filename = f"{uuid.uuid4()}.{ext}"

        # 3. 경로 결정 (core/constants.py에 정의된 상수 사용)
        if upload_type == "answer":
            path_prefix = ANSWER_IMAGE_UPLOAD_PATH
        else:
            path_prefix = QUESTION_IMAGE_UPLOAD_PATH

        key = f"{path_prefix}{new_filename}"

        try:
            s3_client = S3Client()

            # Presigned URL 생성 (업로드용)
            presigned_url = s3_client.generate_presigned_url(key=key)

            # 최종 이미지 URL 생성 (DB 저장용)
            full_url = s3_client.build_url(key=key)

            return Response(
                {
                    "presigned_url": presigned_url,
                    "img_url": full_url,
                    "key": key
                },
                status=status.HTTP_200_OK
            )

        except Exception as e:
            return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

---

DB에 key값만 저장하도록 수정

• 작성 중 (Frontend)
  • PresignedUploadView에서 받은 img_url로 미리보기를 보여줌
    • 여기선 Serializer 필요 없음
• 저장 (Backend)
  • image_service가 URL에서 도메인을 떼고 Key만 DB에 저장
• 조회 (Frontend)
  • 나중에 사용자가 글을 클릭해서 들어오면,
    • Serializer가 DB의 Key에 현재 설정된 도메인을 붙여서 img_url을 내려줌

---

전체 분석

업로드 (PresignedUploadAPIView)

• 클라이언트에게 img_url(Full URL)과 key를 모두 내려줌
  • 프론트엔드는 즉시 img_url을 사용해 미리보기를 띄울 수 있고,
  • 백엔드는 나중에 이 URL이 들어오면 key로 변환하여 저장할 준비 완료
  • 확장자 검증과 uuid 파일명 생성 로직도 있음

동기화 및 저장 (image_service.py)

• key = extract_key_from_url(url)을 통해 DB에 Key만 저장하도록 변경
• transaction.on_commit을 사용하여
  • DB 트랜잭션이 확실히 성공했을 때만 S3 파일을 삭제하도록 함
• utils로 파싱 로직과 S3 검증 로직을 분리하여 코드 가독성을 상승시킴

조회 (QuestionImageSerializer)

• DB에 있는 Key(folder/img.jpg)를 꺼낼 때 S3Client().build_url()을
  • 태워서 Full URL로 내보냄

---

presigned url 요청

• 클라이언트가 요청을 보낼때는 파일 이름(file_name)만 받음
• 확장자 확인용 (.jpg, .png)
  • file_name 뒤에 붙은 확장자를 보고 "이게 이미지 파일이 맞나?" 검사
• Key 생성용
  • 서버는 이 확장자를 이용해서 question_images/난수이름.jpg 같은 고유한 경로(Key)를 미리 만들어 둠

---

---

stash

• 현재 브랜치에서 수정 중인 파일이나 최신 커밋 상태의 파일들을 임시 저장소(Stash)에 넣음
  • git stash만 사용하면 "커밋되지 않은 변경사항"만 저장

# --- [ 1단계: 현재 작업 내용 저장하기 (Stash) ] ---
git add .         # 모든 변경 사항을 추적 대상으로 등록
git stash         # 임시 저장 (현재 워킹 디렉토리는 깨끗해짐)

# --- [ 2단계: 기준이 되는 브랜치로 이동 및 새 브랜치 생성 ] ---
git checkout develop      # 기준 브랜치로 이동
git pull origin develop   # (선택) 최신 상태 유지
git checkout -b 새로운브랜치이름  # 새로 시작할 브랜치 생성 및 이동

# --- [ 3단계: 저장했던 작업 내용 불러오기 ] ---
git stash pop

---

테스트 코드 현황

• 이미지 추출 및 저장
  • 본문에 <img> 태그가 있을 때 DB(QuestionImage)에 레코드가 잘 생성되는지 확인.
• Key 기반 저장
  • URL 전체가 아닌 S3 Key(question_images/...)만 DB에 저장되는지 확인.
• 이미지 삭제 동기화
  • 본문에서 이미지를 지웠을 때 DB에서도 지워지고 S3 삭제 함수가 호출되는지 확인.
• 상세 조회 응답
  • DB의 Key가 Serializer를 통해 다시 Full URL로 변환되어 응답되는지 확인.

---

이력서

• 1일차부터 스토리를 만들어 봐라
• 내가 구현해야하는 기능에 대한 전반적인 큰 시야로 기능을 구현해라

새롭게 알게된 내용 ✅

오늘 발생한 문제(발생 했다면) ✅