SDK

"""
AWS SDK (boto3) 사용 예제
S3 버킷 생성, 파일 업로드, 다운로드를 포함한 실무 패턴
"""

import boto3
from botocore.exceptions import ClientError
import logging
import os

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class S3Manager:
    """AWS S3 작업을 관리하는 클래스"""

    def __init__(self, region_name: str = "ap-northeast-2"):
        """
        S3 클라이언트 초기화

        Args:
            region_name: AWS 리전 (기본값: 서울)
        """
        self.s3_client = boto3.client(
            "s3",
            region_name=region_name,
            # 자격 증명은 환경 변수 또는 AWS CLI 설정에서 자동 로드
            # AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
        )
        self.region = region_name

    def create_bucket(self, bucket_name: str) -> bool:
        """
        S3 버킷 생성

        Args:
            bucket_name: 생성할 버킷 이름 (전역적으로 고유해야 함)

        Returns:
            성공 여부
        """
        try:
            # 서울 리전은 LocationConstraint 필수
            self.s3_client.create_bucket(
                Bucket=bucket_name,
                CreateBucketConfiguration={
                    "LocationConstraint": self.region
                }
            )
            logger.info(f"버킷 생성 완료: {bucket_name}")
            return True
        except ClientError as e:
            error_code = e.response["Error"]["Code"]
            if error_code == "BucketAlreadyOwnedByYou":
                logger.info(f"버킷이 이미 존재합니다: {bucket_name}")
                return True
            logger.error(f"버킷 생성 실패: {e}")
            return False

    def upload_file(
        self,
        file_path: str,
        bucket_name: str,
        object_key: str = None
    ) -> dict:
        """
        파일을 S3에 업로드

        Args:
            file_path: 로컬 파일 경로
            bucket_name: 대상 버킷 이름
            object_key: S3 객체 키 (기본값: 파일명)

        Returns:
            업로드 결과 딕셔너리
        """
        if object_key is None:
            object_key = os.path.basename(file_path)

        try:
            # 파일 크기 확인
            file_size = os.path.getsize(file_path)

            # 5GB 이상이면 멀티파트 업로드 사용
            if file_size > 5 * 1024 * 1024 * 1024:
                logger.info("대용량 파일, 멀티파트 업로드 사용")

            self.s3_client.upload_file(
                file_path,
                bucket_name,
                object_key,
                ExtraArgs={"ContentType": "application/octet-stream"}
            )

            # 업로드된 객체 URL 생성
            url = f"https://{bucket_name}.s3.{self.region}.amazonaws.com/{object_key}"

            logger.info(f"업로드 완료: {object_key}")
            return {
                "success": True,
                "bucket": bucket_name,
                "key": object_key,
                "size_bytes": file_size,
                "url": url
            }
        except ClientError as e:
            logger.error(f"업로드 실패: {e}")
            return {"success": False, "error": str(e)}

    def download_file(
        self,
        bucket_name: str,
        object_key: str,
        download_path: str
    ) -> dict:
        """
        S3에서 파일 다운로드

        Args:
            bucket_name: 소스 버킷 이름
            object_key: S3 객체 키
            download_path: 다운로드 경로

        Returns:
            다운로드 결과 딕셔너리
        """
        try:
            self.s3_client.download_file(
                bucket_name,
                object_key,
                download_path
            )
            logger.info(f"다운로드 완료: {download_path}")
            return {
                "success": True,
                "path": download_path,
                "size_bytes": os.path.getsize(download_path)
            }
        except ClientError as e:
            logger.error(f"다운로드 실패: {e}")
            return {"success": False, "error": str(e)}


# 사용 예시
if __name__ == "__main__":
    # S3 매니저 초기화
    s3 = S3Manager(region_name="ap-northeast-2")

    # 버킷 생성
    bucket = "my-app-data-bucket-2024"
    s3.create_bucket(bucket)

    # 파일 업로드
    result = s3.upload_file(
        file_path="./data/model.pkl",
        bucket_name=bucket,
        object_key="models/v1/model.pkl"
    )

    if result["success"]:
        print(f"업로드 URL: {result['url']}")
        print(f"파일 크기: {result['size_bytes']:,} bytes")