Cloud/AWS Cloud Computing
[AWS] Boto3 사용하기 1. S3 다루기
스파이디웹
2023. 4. 12. 13:49
728x90
1. 사전 준비
Client
- low-level 인터페이스
- service description에 의해 만들어짐
- botocore 수준의 client를 공개(botocore는 AWS CLI와 boto3의 기초가 되는 라이브러리)
- AWS API와 1:1 매핑됨
- 메소드가 스네이크 케이스로 정의되어 있음
Resource
- high-level, 객체지향적 인터페이스
- resource description에 의해 만들어짐
- 식별자(identifier)와 속성(attribute)을 사용
- 자원에 대한 조작 위주
2. 버킷 생성
import logging
import boto3
from botocore.exceptions import ClientError
class AWSBoto3():
def __init__(self):
self.access_key = 'accesskey'
self.secret_access_key = 'secret_key'
def create_bucket(self, 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', aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
s3_client.create_bucket(Bucket=bucket_name)
else:
s3_client = boto3.client('s3', region_name=region, aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
#location = {'LocationConstraint': region}# when the region is configured to us-east-1, default bucket is created in us-east-1 region
s3_client.create_bucket(Bucket=bucket_name)#,CreateBucketConfiguration=location) 한국 리전은 추가되어야 작동함
except ClientError as e:
logging.error(e)
return False
return True
if __name__=="__main__":
AWSboto3 = AWSBoto3()
AWSboto3.create_bucket(bucket_name='bucket-createdby-boto3',region='us-east-1')
여러개의 버킷을 루프(반복문)으로 만들고 싶다면 다음과 같이 사용
if __name__=="__main__":
AWSboto3 = AWSBoto3()
for i in range(3):
AWSboto3.create_bucket(bucket_name=f'bucket{i}-createdby-boto3',region='us-east-1')
3. 버킷 내 폴더 생성
import logging
import boto3
from botocore.exceptions import ClientError
class AWSBoto3():
def __init__(self):
self.access_key = 'access_key'
self.secret_access_key = 'secret_key'
def create_folder(self, bucket_name, folder_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', aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
s3_client.put_object(Bucket=bucket_name, Key=(folder_name+'/'))
else:
s3_client = boto3.client('s3', region_name=region, aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
#location = {'LocationConstraint': region}# when the region is configured to us-east-1, default bucket is created in us-east-1 region
s3_client.put_object(Bucket=bucket_name, Key=(folder_name+'/'))#,CreateBucketConfiguration=location)
except ClientError as e:
logging.error(e)
return False
return True
if __name__=="__main__":
AWSboto3 = AWSBoto3()
AWSboto3.create_folder('bucket-createdby-boto3','folder_lv1/folder_lv2/folder_lv3','us-east-1')
4. 존재하는 버킷 출력(List)
import logging
import boto3
from botocore.exceptions import ClientError
class AWSBoto3():
def __init__(self):
self.access_key = 'access_key'
self.secret_access_key = 'secret_key'
def list_bucket(self, region):
# List bucket
try:
if region is None:
s3_client = boto3.client('s3', aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
response = s3_client.list_buckets()
print('Existing buckets:')
print('----------------------')
for bucket in response['Buckets']:
print(f'{bucket["Name"]}')
print('----------------------')
else:
s3_client = boto3.client('s3', region_name=region, aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
#location = {'LocationConstraint': region}# when the region is configured to us-east-1, default bucket is created in us-east-1 region
response = s3_client.list_buckets()#,CreateBucketConfiguration=location)
print('Existing buckets:')
print('----------------------')
for bucket in response['Buckets']:
print(f'{bucket["Name"]}')
print('----------------------')
except ClientError as e:
logging.error(e)
return False
return True
if __name__=="__main__":
AWSboto3 = AWSBoto3()
AWSboto3.list_bucket('us-east-1')
5. 파일 업로드
import logging
import boto3
from botocore.exceptions import ClientError
import os
class AWSBoto3():
def __init__(self):
self.access_key = 'access_key'
self.secret_access_key = 'secret_key'
def upload_file_to_s3(self, file_name, bucket, object_name=None):
"""Upload a file to an S3 bucket
:param file_name: File to upload
:param bucket: Bucket to upload to
:param object_name: S3 object name. If not specified then file_name is used
:return: True if file was uploaded, else False
"""
# If S3 object_name was not specified, use file_name
if object_name is None:
object_name = os.path.basename(file_name)
# Upload the file
s3_client = boto3.client('s3')
try:
response = s3_client.upload_file(file_name, bucket, object_name)
except ClientError as e:
logging.error(e)
return False
return True
if __name__=="__main__":
AWSboto3 = AWSBoto3()
AWSboto3.upload_file_to_s3('파일경로/motor_ny.csv','bucket-createdby-boto3-korea','folder_lv1/folder_lv2/folder_lv3/motor_ny.csv')
6. 파일 다운로드
import logging
import boto3
from botocore.exceptions import ClientError
import os
class AWSBoto3():
def __init__(self):
self.access_key = 'access_key'
self.secret_access_key = 'secret_key'
def download_file_from_s3(self, bucket, object_name, file_name):
s3_client = boto3.client('s3')
try:
s3_client.download_file(bucket, object_name, file_name)
except ClientError as e:
logging.error(e)
return False
return True
if __name__=="__main__":
AWSboto3 = AWSBoto3()
AWSboto3.download_file_from_s3('bucket-createdby-boto3-korea','folder_lv1/folder_lv2/folder_lv3/motor_ny.csv','C:/Users/JIHO PARK/Downloads/download_from_s3.csv')
7. 파일 이동 설정값
1) Multipart transfer
import boto3
from boto3.s3.transfer import TransferConfig
# Set the desired multipart threshold value (5GB)
GB = 1024 ** 3
config = TransferConfig(multipart_threshold=5*GB)
# Perform the transfer
s3 = boto3.client('s3')
s3.upload_file('FILE_NAME', 'BUCKET_NAME', 'OBJECT_NAME', Config=config)
Multipart
웹 클라이언트가 요청을 보낼 때, http 프로토콜의 바디 부분에 데이터를 여러 부분으로 나눠서 보내는 것
2) Concurrent transfewr operations
- 최대 병렬처리 개수 지정
- 기본값은 10으로 지정
# To consume less downstream bandwidth, decrease the maximum concurrency
config = TransferConfig(max_concurrency=5)
# Download an S3 object
s3 = boto3.client('s3')
s3.download_file('BUCKET_NAME', 'OBJECT_NAME', 'FILE_NAME', Config=config)3) Threads
- 병렬성을 실행시키기 위해 threads를 사용함
- use_threads = False로 설정하면 threads 사용 안하는 것이고, 병렬수행은 이뤄지지 않음(max_concurrency 속성은 무시 됨)
# Disable thread use/transfer concurrency
config = TransferConfig(use_threads=False)
s3 = boto3.client('s3')
s3.download_file('BUCKET_NAME', 'OBJECT_NAME', 'FILE_NAME', Config=config)사용된 최종 코드
import logging
import boto3
from botocore.exceptions import ClientError
import os
class AWSBoto3():
def __init__(self):
self.access_key = 'access_key'
self.secret_access_key = 'secret_key'
def create_bucket(self, 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', aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
s3_client.create_bucket(Bucket=bucket_name)
else:
s3_client = boto3.client('s3', region_name=region, aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
location = {'LocationConstraint': region}# when the region is configured to us-east-1, default bucket is created in us-east-1 region
s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration=location)#,CreateBucketConfiguration=location)
except ClientError as e:
logging.error(e)
return False
return True
def create_folder(self, bucket_name, folder_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 folder
try:
if region is None:
s3_client = boto3.client('s3', aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
s3_client.put_object(Bucket=bucket_name, Key=(folder_name+'/'))
else:
s3_client = boto3.client('s3', region_name=region, aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
#location = {'LocationConstraint': region}# when the region is configured to us-east-1, default bucket is created in us-east-1 region
s3_client.put_object(Bucket=bucket_name, Key=(folder_name+'/'))#,CreateBucketConfiguration=location)
except ClientError as e:
logging.error(e)
return False
return True
def list_bucket(self, region):
# List bucket
try:
if region is None:
s3_client = boto3.client('s3', aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
response = s3_client.list_buckets()
print('Existing buckets:')
print('----------------------')
for bucket in response['Buckets']:
print(f'{bucket["Name"]}')
print('----------------------')
else:
s3_client = boto3.client('s3', region_name=region, aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_access_key)
#location = {'LocationConstraint': region}# when the region is configured to us-east-1, default bucket is created in us-east-1 region
response = s3_client.list_buckets()#,CreateBucketConfiguration=location)
print('Existing buckets:')
print('----------------------')
for bucket in response['Buckets']:
print(f'{bucket["Name"]}')
print('----------------------')
except ClientError as e:
logging.error(e)
return False
return True
def upload_file_to_s3(self, file_name, bucket, object_name=None):
"""Upload a file to an S3 bucket
:param file_name: File to upload
:param bucket: Bucket to upload to
:param object_name: S3 object name. If not specified then file_name is used
:return: True if file was uploaded, else False
"""
# If S3 object_name was not specified, use file_name
if object_name is None:
object_name = os.path.basename(file_name)
# Upload the file
s3_client = boto3.client('s3')
try:
response = s3_client.upload_file(file_name, bucket, object_name)
except ClientError as e:
logging.error(e)
return False
return True
def download_file_from_s3(self, bucket, object_name, file_name):
"""Upload a file to an S3 bucket
:param file_name: file name and a path to download
:param bucket: Bucket name which contains objects to download
:param object_name: S3 object name. If not specified then file_name is used
:return: True if file was uploaded, else False
"""
s3_client = boto3.client('s3')
try:
s3_client.download_file(bucket, object_name, file_name)
except ClientError as e:
logging.error(e)
return False
return True
if __name__=="__main__":
AWSboto3 = AWSBoto3()
참조:
https://planbs.tistory.com/entry/boto3resource%EC%99%80-boto3client%EC%9D%98-%EC%B0%A8%EC%9D%B4
728x90