Python boto3 チートシート

response = s3_client.list_buckets()
buckets = [b['Name'] for b in response['Buckets']]
print(buckets) 

buckets = [b.name for b in s3_resource.buckets.all()]
print(buckets) 

# リージョン指定が必要な場合がある (us-east-1以外)
s3_client.create_bucket( Bucket='my-unique-bucket-name-123', CreateBucketConfiguration={ 'LocationConstraint': 'ap-northeast-1' # us-east-1の場合は不要 }
) 

s3_resource.create_bucket( Bucket='my-unique-bucket-name-456', CreateBucketConfiguration={ 'LocationConstraint': 'ap-northeast-1' }
) 

response = s3_client.list_objects_v2(Bucket='my-bucket-name')
if 'Contents' in response: objects = [obj['Key'] for obj in response['Contents']] print(objects)
# Prefix指定
response_prefix = s3_client.list_objects_v2(Bucket='my-bucket-name', Prefix='folder/')
# Paginator利用 (大量オブジェクトの場合)
paginator = s3_client.get_paginator('list_objects_v2')
pages = paginator.paginate(Bucket='my-bucket-name')
for page in pages: if 'Contents' in page: for obj in page['Contents']: print(obj['Key']) 

bucket = s3_resource.Bucket('my-bucket-name')
objects = [obj.key for obj in bucket.objects.all()]
print(objects)
# Prefix指定
objects_prefix = [obj.key for obj in bucket.objects.filter(Prefix='folder/')]
print(objects_prefix) 

# ファイルパスからアップロード
s3_client.upload_file( Filename='/path/to/local/file.txt', # ローカルファイルパス Bucket='my-bucket-name', Key='destination/object/key.txt' # S3上のキー
)
# ファイルオブジェクトからアップロード
with open('/path/to/local/image.jpg', 'rb') as f: s3_client.upload_fileobj( Fileobj=f, Bucket='my-bucket-name', Key='images/image.jpg', ExtraArgs={'ContentType': 'image/jpeg'} # ContentTypeなどを指定 ) 

bucket = s3_resource.Bucket('my-bucket-name')
# ファイルパスからアップロード
bucket.upload_file( Filename='/path/to/local/file.txt', Key='destination/object/key.txt'
)
# ファイルオブジェクトからアップロード
with open('/path/to/local/image.jpg', 'rb') as f: bucket.upload_fileobj( Fileobj=f, Key='images/image.jpg', ExtraArgs={'ContentType': 'image/jpeg'} )
# put_object (バイナリデータや文字列を直接)
s3_resource.Object('my-bucket-name', 'data/text.txt').put(Body=b'Hello S3 from resource') 

# バイト列をアップロード
binary_data = b'\x01\x02\x03\x04'
s3_client.put_object( Bucket='my-bucket-name', Key='binary/data.bin', Body=binary_data, ContentType='application/octet-stream'
)
# 文字列をアップロード (UTF-8エンコード推奨)
text_data = 'こんにちは、S3！'
s3_client.put_object( Bucket='my-bucket-name', Key='text/greeting.txt', Body=text_data.encode('utf-8'), # バイト列に変換 ContentType='text/plain; charset=utf-8'
) 

s3_object = s3_resource.Object('my-bucket-name', 'binary/data_res.bin')
s3_object.put( Body=b'\x05\x06\x07\x08', ContentType='application/octet-stream'
)
s3_object_text = s3_resource.Object('my-bucket-name', 'text/greeting_res.txt')
s3_object_text.put( Body='Hello from resource API!'.encode('utf-8'), ContentType='text/plain; charset=utf-8'
) 

# ファイルにダウンロード
s3_client.download_file( Bucket='my-bucket-name', Key='destination/object/key.txt', Filename='/path/to/save/local/file.txt' # 保存先ファイルパス
)
# ファイルオブジェクトにダウンロード
with open('/path/to/save/local/image.jpg', 'wb') as f: s3_client.download_fileobj( Bucket='my-bucket-name', Key='images/image.jpg', Fileobj=f ) 

bucket = s3_resource.Bucket('my-bucket-name')
# ファイルにダウンロード
bucket.download_file( Key='destination/object/key.txt', Filename='/path/to/save/local/file_res.txt'
)
# ファイルオブジェクトにダウンロード
with open('/path/to/save/local/image_res.jpg', 'wb') as f: bucket.download_fileobj( Key='images/image.jpg', Fileobj=f )
# get_object (メモリ上に読み込み)
s3_object = s3_resource.Object('my-bucket-name', 'text/greeting.txt')
response = s3_object.get()
body = response['Body'].read()
print(body.decode('utf-8')) 

response = s3_client.get_object( Bucket='my-bucket-name', Key='binary/data.bin'
)
# StreamingBody オブジェクトが返る
body_stream = response['Body']
# 全て読み込む
binary_content = body_stream.read()
print(binary_content)
# ストリームとして扱うことも可能
# for chunk in body_stream.iter_chunks(chunk_size=1024):
# process(chunk)
body_stream.close() # 必要に応じてクローズ 

s3_object = s3_resource.Object('my-bucket-name', 'text/greeting.txt')
response = s3_object.get()
# response['Body'] は StreamingBody
body = response['Body'].read()
print(body.decode('utf-8')) # テキストの場合はデコード
response['Body'].close() 

# 単一オブジェクト削除
s3_client.delete_object( Bucket='my-bucket-name', Key='destination/object/key.txt'
)
# 複数オブジェクト削除
response = s3_client.delete_objects( Bucket='my-bucket-name', Delete={ 'Objects': [ {'Key': 'images/image.jpg'}, {'Key': 'binary/data.bin'}, ], 'Quiet': False # Trueにすると成功したオブジェクトは返さない }
)
print(response.get('Deleted'))
print(response.get('Errors')) 

s3_object = s3_resource.Object('my-bucket-name', 'text/greeting.txt')
s3_object.delete()
bucket = s3_resource.Bucket('my-bucket-name')
response = bucket.delete_objects( Delete={ 'Objects': [ {'Key': 'folder/file1.txt'}, {'Key': 'folder/file2.txt'}, ] }
)
print(response.get('Deleted')) 

# GETリクエスト用URL (ダウンロード)
get_url = s3_client.generate_presigned_url( 'get_object', Params={'Bucket': 'my-bucket-name', 'Key': 'private/document.pdf'}, ExpiresIn=3600 # URLの有効期限 (秒)
)
print(f"Download URL: {get_url}")
# PUTリクエスト用URL (アップロード)
put_url = s3_client.generate_presigned_url( 'put_object', Params={'Bucket': 'my-upload-bucket', 'Key': 'user-uploads/new_file.zip'}, ExpiresIn=600, HttpMethod='PUT' # HTTPメソッド指定
)
print(f"Upload URL: {put_url}")
# アップロード例: requests.put(put_url, data=file_content) 

リソースAPIには直接的な generate_presigned_url メソッドはありません。クライアントAPIを使用します。

# リソースオブジェクトからクライアントを取得して使用
s3_object = s3_resource.Object('my-bucket-name', 'private/document.pdf')
presigned_url = s3_object.meta.client.generate_presigned_url( 'get_object', Params={'Bucket': s3_object.bucket_name, 'Key': s3_object.key}, ExpiresIn=3600
)
print(presigned_url) 

# 全てのインスタンス情報取得
response = ec2_client.describe_instances()
for reservation in response['Reservations']: for instance in reservation['Instances']: print(f"ID: {instance['InstanceId']}, State: {instance['State']['Name']}")
# 特定の状態のインスタンスをフィルタリング
response_running = ec2_client.describe_instances( Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)
# タグでフィルタリング
response_tag = ec2_client.describe_instances( Filters=[{'Name': 'tag:Environment', 'Values': ['production']}]
) 

# 全てのインスタンス
for instance in ec2_resource.instances.all(): print(f"ID: {instance.id}, State: {instance.state['Name']}")
# 実行中のインスタンス
for instance in ec2_resource.instances.filter( Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
): print(f"Running ID: {instance.id}")
# タグでフィルタリング
for instance in ec2_resource.instances.filter( Filters=[{'Name': 'tag:Project', 'Values': ['Blue']}]
): print(f"Project Blue ID: {instance.id}") 

instance_id = 'i-xxxxxxxxxxxxxxxxx'
response = ec2_client.describe_instances(InstanceIds=[instance_id])
if response['Reservations']: instance_info = response['Reservations'][0]['Instances'][0] print(f"Private IP: {instance_info.get('PrivateIpAddress')}")
else: print(f"Instance {instance_id} not found.") 

instance_id = 'i-yyyyyyyyyyyyyyyyy'
try: instance = ec2_resource.Instance(instance_id) instance.load() # 情報をロード (APIコール発生) print(f"Instance Type: {instance.instance_type}") print(f"Public DNS: {instance.public_dns_name}")
except ClientError as e: if e.response['Error']['Code'] == 'InvalidInstanceID.NotFound': print(f"Instance {instance_id} not found.") else: raise e 

response = ec2_client.run_instances( ImageId='ami-xxxxxxxxxxxxxxxxx', # AMI ID InstanceType='t2.micro', MinCount=1, MaxCount=1, KeyName='my-key-pair', # キーペア名 SecurityGroupIds=['sg-xxxxxxxxxxxxxxxxx'], # セキュリティグループID SubnetId='subnet-xxxxxxxxxxxxxxxxx', # サブネットID TagSpecifications=[ # タグ付け { 'ResourceType': 'instance', 'Tags': [ {'Key': 'Name', 'Value': 'MyBoto3Instance'}, {'Key': 'Environment', 'Value': 'Development'}, ] }, ] # UserData='#!/bin/bash\nyum update -y' # 起動スクリプト
)
instance_id = response['Instances'][0]['InstanceId']
print(f"Started instance: {instance_id}")
# 起動完了を待つ (Waiter)
waiter = ec2_client.get_waiter('instance_running')
waiter.wait(InstanceIds=[instance_id])
print(f"Instance {instance_id} is now running.") 

instances = ec2_resource.create_instances( ImageId='ami-yyyyyyyyyyyyyyyyy', InstanceType='t3.small', MinCount=1, MaxCount=1, KeyName='my-other-key-pair', SecurityGroupIds=['sg-yyyyyyyyyyyyyyyyy'], SubnetId='subnet-yyyyyyyyyyyyyyyyy', TagSpecifications=[ {'ResourceType': 'instance', 'Tags': [{'Key': 'Name', 'Value': 'MyResourceInstance'}]} ]
)
instance = instances[0] # create_instancesはリストを返す
print(f"Created instance: {instance.id}")
# 起動完了を待つ
instance.wait_until_running()
instance.load() # 状態を更新
print(f"Instance {instance.id} is running with state: {instance.state['Name']}") 

instance_ids = ['i-xxxxxxxxxxxxxxxxx', 'i-zzzzzzzzzzzzzzzzz']
response = ec2_client.stop_instances(InstanceIds=instance_ids)
print(f"Stopping instances: {response['StoppingInstances']}")
# 停止完了を待つ (Waiter)
waiter = ec2_client.get_waiter('instance_stopped')
waiter.wait(InstanceIds=instance_ids)
print(f"Instances {instance_ids} are stopped.") 

instance_id = 'i-yyyyyyyyyyyyyyyyy'
instance = ec2_resource.Instance(instance_id)
response = instance.stop()
print(f"Stopping instance {instance.id}: {response['StoppingInstances']}")
# 停止完了を待つ
instance.wait_until_stopped()
instance.load()
print(f"Instance {instance.id} is stopped with state: {instance.state['Name']}") 

instance_ids = ['i-xxxxxxxxxxxxxxxxx']
ec2_client.reboot_instances(InstanceIds=instance_ids)
print(f"Rebooting instances: {instance_ids}") 

instance_id = 'i-yyyyyyyyyyyyyyyyy'
instance = ec2_resource.Instance(instance_id)
instance.reboot()
print(f"Rebooting instance {instance.id}") 

instance_ids = ['i-xxxxxxxxxxxxxxxxx', 'i-zzzzzzzzzzzzzzzzz']
response = ec2_client.terminate_instances(InstanceIds=instance_ids)
print(f"Terminating instances: {response['TerminatingInstances']}")
# 終了完了を待つ (Waiter)
waiter = ec2_client.get_waiter('instance_terminated')
waiter.wait(InstanceIds=instance_ids)
print(f"Instances {instance_ids} are terminated.") 

instance_id = 'i-yyyyyyyyyyyyyyyyy'
instance = ec2_resource.Instance(instance_id)
response = instance.terminate()
print(f"Terminating instance {instance.id}: {response['TerminatingInstances']}")
# 終了完了を待つ
instance.wait_until_terminated()
# 終了後はload()するとエラーになることが多い
print(f"Instance {instance.id} is terminated.") 

response = dynamodb_client.list_tables()
print(response['TableNames']) 

# リソースオブジェクト自体には list_tables がない
# クライアントAPIを使うか、Sessionからクライアントを取得する
dynamodb_client_from_resource = dynamodb_resource.meta.client
response = dynamodb_client_from_resource.list_tables()
print(response['TableNames']) 

# 型を指定する必要がある (S: String, N: Number, B: Binary, BOOL: Boolean, ...)
response = dynamodb_client.put_item( TableName='my-data-table', Item={ 'userId': {'S': 'user123'}, 'timestamp': {'N': '1678886400'}, 'data': {'M': { # Map型 'message': {'S': 'Hello from client'}, 'is_processed': {'BOOL': False} }}, 'tags': {'L': [ # List型 {'S': 'tag1'}, {'S': 'tag2'} ]} }
)
print(response) 

import decimal # 数値型にはDecimalを使うのが推奨される
table = dynamodb_resource.Table('my-data-table')
response = table.put_item( Item={ 'userId': 'user456', 'timestamp': decimal.Decimal('1678887000'), # Decimalを使用 'data': { 'message': 'Hello from resource', 'is_processed': True }, 'tags': ['tagA', 'tagB'], 'optional_attr': None # Noneは属性として保存されない }
)
print(response) 

response = dynamodb_client.get_item( TableName='my-data-table', Key={ 'userId': {'S': 'user123'}, 'timestamp': {'N': '1678886400'} # パーティションキーとソートキー (あれば) }, ProjectionExpression='userId, data.message' # 取得する属性を指定
)
if 'Item' in response: print(response['Item'])
else: print("Item not found.") 

table = dynamodb_resource.Table('my-data-table')
response = table.get_item( Key={ 'userId': 'user456', 'timestamp': decimal.Decimal('1678887000') }, ProjectionExpression='tags, data.is_processed'
)
if 'Item' in response: item = response['Item'] print(item) # Decimal型は Decimal オブジェクトとして返る print(type(item.get('timestamp'))) # else: print("Item not found.") 

response = dynamodb_client.update_item( TableName='my-data-table', Key={ 'userId': {'S': 'user123'}, 'timestamp': {'N': '1678886400'} }, UpdateExpression='SET data.is_processed = :val1, #tag_attr[0] = :val2 ADD version :inc', # SET, REMOVE, ADD, DELETE ExpressionAttributeNames={ # 予約語を避けるため (#) '#tag_attr': 'tags' }, ExpressionAttributeValues={ # 値をプレースホルダで渡す (:) ':val1': {'BOOL': True}, ':val2': {'S': 'updated_tag'}, ':inc': {'N': '1'} # 数値のインクリメント }, ReturnValues='UPDATED_NEW' # 更新後の値を返す
)
print(response['Attributes']) 

table = dynamodb_resource.Table('my-data-table')
response = table.update_item( Key={ 'userId': 'user456', 'timestamp': decimal.Decimal('1678887000') }, UpdateExpression='SET data.message = :msg REMOVE optional_attr ADD counter :c', ExpressionAttributeValues={ ':msg': 'Updated message via resource', ':c': decimal.Decimal('1') # カウンターアトム操作 }, ReturnValues='ALL_NEW' # 更新後の全属性を返す
)
print(response['Attributes']) 

response = dynamodb_client.delete_item( TableName='my-data-table', Key={ 'userId': {'S': 'user123'}, 'timestamp': {'N': '1678886400'} }, ReturnValues='ALL_OLD' # 削除前の値を返す
)
if 'Attributes' in response: print("Deleted item:", response['Attributes'])
else: print("Item already deleted or not found.") 

table = dynamodb_resource.Table('my-data-table')
response = table.delete_item( Key={ 'userId': 'user456', 'timestamp': decimal.Decimal('1678887000') }, ConditionExpression='attribute_exists(data.message)' # 条件付き削除 # ReturnValues='ALL_OLD'
)
print(response) # 成功時はメタデータのみ、失敗時は例外発生 

# 特定のuserIdを持つ項目をtimestampで降順に取得
response = dynamodb_client.query( TableName='my-data-table', KeyConditionExpression='userId = :uid', # パーティションキーは必須 FilterExpression='data.is_processed = :processed', # 結果に対する追加フィルタ ExpressionAttributeValues={ ':uid': {'S': 'user789'}, ':processed': {'BOOL': True} }, ProjectionExpression='timestamp, data.message', ScanIndexForward=False # Falseで降順、True(デフォルト)で昇順
)
print(response['Items']) 

from boto3.dynamodb.conditions import Key, Attr
table = dynamodb_resource.Table('my-data-table')
# 特定のuserIdで、timestampが特定の値より大きい項目を取得
response = table.query( KeyConditionExpression=Key('userId').eq('user789') & Key('timestamp').gt(decimal.Decimal('1678880000')), FilterExpression=Attr('tags').contains('important') | Attr('data.message').begins_with('urgent'), ProjectionExpression='userId, timestamp, tags'
)
print(response['Items'])
# GSI (Global Secondary Index) を利用する場合
# response_gsi = table.query(
# IndexName='my-gsi-name',
# KeyConditionExpression=Key('gsi_partition_key').eq('value')
# ) 

# テーブル全体をスキャンして特定の属性を持つ項目を取得 (非効率なので注意)
response = dynamodb_client.scan( TableName='my-data-table', FilterExpression='attribute_exists(error_code)', ProjectionExpression='userId, timestamp, error_code'
)
items = response['Items']
# 結果が1MBを超える場合は LastEvaluatedKey を使ってページングが必要
while 'LastEvaluatedKey' in response: response = dynamodb_client.scan( TableName='my-data-table', FilterExpression='attribute_exists(error_code)', ProjectionExpression='userId, timestamp, error_code', ExclusiveStartKey=response['LastEvaluatedKey'] ) items.extend(response['Items'])
print(f"Found {len(items)} items with error_code.") 

from boto3.dynamodb.conditions import Attr
table = dynamodb_resource.Table('my-data-table')
# 特定のタグを含む項目をスキャン
response = table.scan( FilterExpression=Attr('tags').contains('archive')
)
items = response['Items']
# リソースAPIのscanもページングが必要な場合がある
while 'LastEvaluatedKey' in response: response = table.scan( FilterExpression=Attr('tags').contains('archive'), ExclusiveStartKey=response['LastEvaluatedKey'] ) items.extend(response['Items'])
print(f"Found {len(items)} archived items.")