[AWS] python에서 aws기능을 사용하고 싶다면 boto3

1. boto3란?

- AWS CLI의 Python 라이브러리 버전

- AWS의 모든 서비스를 Python 코드로 제어할 수 있음

 

2. boto3 설치

pip install boto3

 

3. 서비스별 boto3 사용법

1) Clinet : API와 1:1 매핑 (AWS CLI와 거의 동일한 방식, 딕셔너리 형태로 응답) [저수준 API]

import boto3

s3 = boto3.client('s3')

# 기본 작업들
s3.create_bucket(Bucket='my-bucket')
s3.upload_file('local.txt', 'my-bucket', 'remote.txt')
s3.download_file('my-bucket', 'remote.txt', 'downloaded.txt')
s3.delete_object(Bucket='my-bucket', Key='remote.txt')

# 목록 조회
response = s3.list_buckets()
response = s3.list_objects_v2(Bucket='my-bucket')

ecr = boto3.client('ecr')

# 저장소 관리
response = ecr.describe_repositories()
ecr.create_repository(repositoryName='my-repo')

# 이미지 관리
response = ecr.list_images(repositoryName='my-repo')
response = ecr.describe_images(repositoryName='my-repo')

# 로그인 토큰
response = ecr.get_login_password()

 

2) Resource : 객체지향적 접근 [고수준 API], 더 직관적이고 간편하나, 일부 서비스만 지원

s3 = boto3.resource('s3')

# 주요 객체들
s3.buckets.all()                    # 모든 버킷
s3.Bucket('bucket-name')           # 특정 버킷
s3.Object('bucket', 'key')         # 특정 객체
s3.ObjectSummary('bucket', 'key')  # 객체 요약 정보
s3.MultipartUpload('bucket', 'object-key', 'upload-id')

ec2 = boto3.resource('ec2')

# 주요 객체들
ec2.instances.all()              # 모든 인스턴스
ec2.Instance('instance-id')      # 특정 인스턴스
ec2.Vpc('vpc-id')               # VPC
ec2.Subnet('subnet-id')         # 서브넷
ec2.SecurityGroup('sg-id')      # 보안 그룹
ec2.Volume('volume-id')         # EBS 볼륨
ec2.Image('ami-id')             # AMI 이미지
ec2.KeyPair('key-name')         # 키 페어
ec2.NetworkInterface('eni-id')   # 네트워크 인터페이스

dynamodb = boto3.resource('dynamodb')

# 주요 객체들
dynamodb.tables.all()           # 모든 테이블
dynamodb.Table('table-name')    # 특정 테이블

sqs = boto3.resource('sqs')

# 주요 객체들
sqs.queues.all()               # 모든 큐
sqs.Queue('queue-url')         # 특정 큐
sqs.Message(queue, receipt_handle)  # 메시지

sns = boto3.resource('sns')

# 주요 객체들
sns.topics.all()                    # 모든 토픽
sns.Topic('topic-arn')             # 특정 토픽
sns.PlatformApplication('app-arn')  # 플랫폼 애플리케이션
sns.PlatformEndpoint('endpoint-arn') # 엔드포인트
sns.Subscription('subscription-arn') # 구독

iam = boto3.resource('iam')

# 주요 객체들
iam.users.all()                 # 모든 사용자
iam.User('username')           # 특정 사용자
iam.Group('group-name')        # 그룹
iam.Role('role-name')          # 역할
iam.Policy('policy-arn')       # 정책
iam.AccessKey('access-key-id') # 액세스 키
iam.CurrentUser()              # 현재 사용자

 

4. 공통 패턴

1) 응답 구조

# 대부분의 Client 응답은 딕셔너리 형태
response = {
    'ResponseMetadata': {
        'RequestId': '...',
        'HTTPStatusCode': 200,
        'HTTPHeaders': {...}
    },
    # 실제 데이터는 서비스별로 다름
    'Buckets': [...],  # S3
    'Instances': [...], # EC2
    'Functions': [...] # Lambda
}

 

2) 페이지네이션 처리

# 많은 결과가 있을 때 페이지네이션 처리
def list_all_s3_objects(bucket_name):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    
    for page in paginator.paginate(Bucket=bucket_name):
        for obj in page.get('Contents', []):
            print(obj['Key'])

 

3) 에러처리

from botocore.exceptions import ClientError

try:
    s3 = boto3.client('s3')
    s3.head_object(Bucket='my-bucket', Key='nonexistent.txt')
except ClientError as e:
    error_code = e.response['Error']['Code']
    if error_code == 'NoSuchKey':
        print("File not found")
    else:
        print(f"Error: {error_code}")