[S3] boto3 SDK 활용 Key 리스트 확인, 오브젝트 Copy, Move

오브젝트를 생성한 후, 동일 버킷(Bucket) 내에서 다른 Key로 오브젝트를 이동해야하는 일이 생겼다.


boto3의 s3 관련 함수들 중에는 Move라는 함수가 존재하지 않았다.
그래서, Copy와 remove를 활용하여 Move를 만들었다.


사용한 코드는 아래와 같다

아래 코드를 보기에 앞서,
아래 코드내용 중에 어떤 동작 함수는 boto3.client를 쓰고, 어떤 동작 함수는 boto3.resource를 쓸것이다.
client와 resource의 차이는 다음과 같다고 한다.

client는 low-level 인터페이스로, AWS API와 1:1 매핑하여 사용할 때,
resource는 high-level 인터페이스로, 자원에 대한 조작을 위주로 사용할 떄,

참고 :https://planbs.tistory.com/?page=3

 

s3 client 정의

s3 = boto3.client(
    's3',
    aws_access_key_id = '<ACCESS KEY>',
    aws_secret_access_key='<SECRET ACCESS KEY>'
)

 

Move 함수

def test_move_key():

    #
    # key listing
    # list_objects는 1000개 까지 가져올수 있다.
    # 만약 옮겨야할 데이터가 1000개 이상이라면,
    # list_object Pagenation을 검색하거나, 아래와 같이 Prefix를 지정하여 한다.
    # obj_list = s3.list_objects(Bucket="<내 버킷 명>", Prefix="2021/12")
    #
    obj_list = s3.list_objects(Bucket="<내 버킷 명>")

    contents = obj_list['Contents']

    for c in contents:
        key = c['Key']
        #
        # 이 아래는 자신의 키에 맞게 수정한다.
        # src_info는 Copy 할 파일 명
        # dst_info는 Copy 된 후, 파일 명
        #
        # ==============================================================
        key_path = key.split('.')[0]
        file_name = '.'.join(key.split('/')[4:])
		
        src_info = key
        dst_info = f"{key_path}/{file_name}"
	# ==============================================================
        
        test_copy_key(src_info, dst_info)

 

Copy 함수

def test_copy_key(src_info, dst_info):
    #
    # copy 부분
    # resource와 client 차이는
    # resource는 자원조작을 위주로 할때,
    # client는 AWS API를 1:1로 사용할때,
    #

    s3_resource = boto3.resource(
        's3',
        aws_access_key_id='<ACCESS KEY>',
        aws_secret_access_key='<SECRET_ACCESS_KEY>'
    )
    bucket = s3_resource.Bucket('<내 버킷 명>')

    for obj in bucket.objects.filter(Prefix=src_info):
        old_source = {'Bucket': '<내 버킷 명>', 'Key': obj.key}
        new_key = obj.key.replace(src_info, dst_info, 1)
        print(f"Copy {obj.key} -> {new_key}")
        new_obj = bucket.Object(new_key)
        new_obj.copy(old_source)
        #
        # 복사가 완료되면 이전 파일은 삭제한다.
        #
        test_remove_key(src_info)

 

Remove 함수

def test_remove_key(src_info):
    print(f"delete key = {src_info}")
    s3.delete_object(Bucket='<내 버킷 명>', Key=src_info)