python - Amazon S3 boto - フォルダを削除するには?

Question

s3 に「test」という名前のフォルダーを作成し、「test_1.jpg」、「test_2.jpg」を「test」にプッシュしました。

boto を使用してフォルダ「test」を削除するにはどうすればよいですか?

Answer 1

これが2018年（ほぼ2019年）バージョンです：

s3 = boto3.resource('s3')
bucket = s3.Bucket('mybucket')
bucket.objects.filter(Prefix="myprefix/").delete()

Answer 2

S3にはフォルダはありません。代わりに、キーはフラットな名前空間を形成します。ただし、名前にスラッシュが含まれるキーは、AWSコンソールを含む一部のプログラムで特に表示されます（たとえば、Amazon S3 boto-フォルダーの作成方法を参照してください）。

「ディレクトリ」を削除する代わりに、プレフィックスと削除でファイルを一覧表示することができます（そして削除する必要があります）。本質的に：

for key in bucket.list(prefix='your/directory/'):
    key.delete()

ただし、このページの他の達成された回答は、より効率的なアプローチを特徴としています。

---

プレフィックスはダミー文字列検索を使用して検索されることに注意してください。プレフィックスがだった場合your/directory、つまり、末尾にスラッシュが追加されていない場合、プログラムは喜んで削除しyour/directory-that-you-wanted-to-remove-is-definitely-not-t‌​his-oneます。

詳細については、「S3botoリストキーがディレクトリキーを返すことがある」を参照してください。

Answer 3

パトリックのソリューションのわずかな改善。ご存じかもしれませんが、 と の両方list_objects()にdelete_objects()オブ​​ジェクト制限が 1000 あります。これが、リストをページ分割し、チャンクで削除する必要がある理由です。これは非常に普遍的であり、サブディレクトリ/パスを削除するために与えるPrefixことができますpaginator.paginate()

client = boto3.client('s3', **credentials)
paginator = client.get_paginator('list_objects_v2')
pages = paginator.paginate(Bucket=self.bucket_name)

delete_us = dict(Objects=[])
for item in pages.search('Contents'):
    delete_us['Objects'].append(dict(Key=item['Key']))

    # flush once aws limit reached
    if len(delete_us['Objects']) >= 1000:
        client.delete_objects(Bucket=bucket, Delete=delete_us)
        delete_us = dict(Objects=[])

# flush rest
if len(delete_us['Objects']):
    client.delete_objects(Bucket=bucket, Delete=delete_us)

Answer 4

S3 バケットでバージョニングが有効になっている場合:

s3 = boto3.resource('s3')
bucket = s3.Bucket('mybucket')
bucket.object_versions.filter(Prefix="myprefix/").delete()

Answer 5

私が行ったように、オブジェクトの内容でフィルタリングする必要がある場合、以下はロジックの青写真です。

def get_s3_objects_batches(s3: S3Client, **base_kwargs):
    kwargs = dict(MaxKeys=1000, **base_kwargs)
    while True:
        response = s3.list_objects_v2(**kwargs)
        # to yield each and every file: yield from response.get('Contents', [])
        yield response.get('Contents', [])
        if not response.get('IsTruncated'):  # At the end of the list?
            break
        continuation_token = response.get('NextContinuationToken')
        kwargs['ContinuationToken'] = continuation_token


def your_filter(b):
   raise NotImplementedError()


session = boto3.session.Session(profile_name=profile_name)
s3client = session.client('s3')
for batch in get_s3_objects_batches(s3client, Bucket=bucket_name, Prefix=prefix):
    to_delete = [{'Key': obj['Key']} for obj in batch if your_filter(obj)]
    if to_delete:
        s3client.delete_objects(Bucket=bucket_name, Delete={'Objects': to_delete})