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.")