boto3 で S3 オブジェクトのコピー

boto3 で S3 の操作メモ

バケットに接続

import boto3

s3 = boto3.resource('s3')
bucket_name = "my-bucket"

bucket = s3.Bucket(bucket_name)

prefix の文字列で bucket 内のオブジェクトをフィルタ

prefix = 'myfolder/original.txt'
bucket.objects.filter(Prefix=prefix)

# => s3.ObjectSummary(bucket_name='my-bucket', key=u'myfolder/original.txt')

同一バケット内でオブジェクトのコピー

"""
CopySource は Dict 型で渡す
"""
s3.Object(bucket_name, new_file).copy_from(CopySource={'Bucket': bucket_name, 'Key': original_file_name})

オブジェクトの削除

s3.Object(bucket_name, original_file_name).delete()

フィルタしたオブジェクトを別フォルダにコピーして削除

試していませんが、別のバケットへのコピーでも動きそうです。

import boto3
import time

s3 = boto3.resource('s3')
bucket_name = 'my-bucket'
bucket = s3.Bucket(bucket_name)

new_folder = "new"
old_folder = "old"

for obj in bucket.objects.filter(Prefix=old_folder):
    old_file = obj.key
    object_name = obj.key.split("/")[-1]
    new_file = '%s/%s' % (new_folder, object_name[-1])
    # コピーが成功するまで5回繰り返す
    for i in xrange(1,6):
      res = s3.Object(bucket.name, new_file).copy_from(CopySource={'Bucket': bucket.name, 'Key': old_file})
      if res['ResponseMetadata']['HTTPStatusCode'] == 200:
          break
      else:
          time.sleep(i)
    # 削除が成功するまで5回繰り返す
    for i in xrange(1,6):
      res = obj.delete()
      if res['ResponseMetadata']['HTTPStatusCode'] == 204:
          print "moved %s complete" % new_file
          break
      else:
          time.sleep(i)

オブジェクトをローカルにコピー(ダウンロード)

新たに記事書きました。

rriifftt.hatenablog.com