LocalStack S3를 python boto3로 접근하기

boto3 docs

---

어떻게 접근해야할까?

boto3 클래스의 클라이언트 메서드는 액세스 키를 이용해서 접근한다. 그런데 로컬스택은 엔드 포인트를 지정해주어야 한다. 다행히 boto3 를라이언트는 엔드 포인트를 지정하는 파라미터가 존재한다. 

버킷 리스트 확인

import boto3

# Retrieve the list of existing buckets
s3 = boto3.client('s3', endpoint_url="http://localhost:4566")
response = s3.list_buckets()

# Output the bucket names
print('Existing buckets:')
for bucket in response['Buckets']:
    print(f'  {bucket["Name"]}')

버킷 생성

import logging
import boto3
from botocore.exceptions import ClientError


def create_bucket(bucket_name, region=None):
    """Create an S3 bucket in a specified region

    If a region is not specified, the bucket is created in the S3 default
    region (us-east-1).

    :param bucket_name: Bucket to create
    :param region: String region to create bucket in, e.g., 'us-west-2'
    :return: True if bucket created, else False
    """

    # Create bucket
    try:
        if region is None:
            s3_client = boto3.client('s3', endpoint_url="http://localhost:4566")
            s3_client.create_bucket(Bucket=bucket_name)
        else:
            s3_client = boto3.client('s3', region_name=region, endpoint_url="http://localhost:4566")
            location = {'LocationConstraint': region}
            s3_client.create_bucket(Bucket=bucket_name,
                                    CreateBucketConfiguration=location)
    except ClientError as e:
        logging.error(e)
        return False
    return True


response = create_bucket("test-bucket")

if response:
    print("Success")
else:
    print("Fail")

버킷 제거

import boto3

# Retrieve the list of existing buckets
s3 = boto3.resource('s3', endpoint_url="http://localhost:4566")
bucket = s3.Bucket("test-bucket")
bucket.delete()

 

버킷에 파일 업로드

import boto3

s3 = boto3.client('s3', endpoint_url="http://localhost:4566")
s3.upload_file(Bucket="test-bucket", 
               Filename="./dummy.txt",  # 업로드 할 파일 
               Key="dummy2.txt")        # 저장 시 사용할 파일 이름

 

버킷에 업로드한 파일 목록 확인

import boto3

s3 = boto3.client('s3', endpoint_url="http://localhost:4566")
res = s3.list_objects(Bucket="test-bucket")

for obj in res["Contents"]:
    print(obj["Key"])

 

 

---