コード

main.py

import boto3
from InquirerPy import inquirer

s3 = boto3.resource('s3')

def list_buckets():
    return [bucket.name for bucket in s3.buckets.all()]

def list_objects(bucket_name, prefix=''):
    bucket = s3.Bucket(bucket_name)
    all_objects = [obj.key for obj in bucket.objects.filter(Prefix=prefix)]
    if prefix:
        all_objects = [obj.replace(prefix, '', 1) for obj in all_objects]
    return [obj for obj in all_objects if obj != ''] 

def download_file(bucket_name, file_key, download_path):
    s3.meta.client.download_file(bucket_name, file_key, download_path)
    print(f'File downloaded at {download_path}')

def select_and_download():
    selected_bucket = inquirer.select(
        message="Select an S3 bucket:",
        choices=list_buckets(),
    ).execute()
    
    current_prefix = ''
    while True:
        objects = list_objects(selected_bucket, current_prefix)
        if not objects:
            print("No files found.")
            break

        selected_obj = current_prefix + inquirer.select(
            message="Select a file or a directory:",
            choices=objects,
        ).execute()

        if selected_obj.endswith('/'):  # Directory
            current_prefix = selected_obj
        else:  # File
            download_path = input(f'Enter the download path for {selected_obj}: ')
            download_file(selected_bucket, selected_obj, download_path)
            break

if __name__ == "__main__":
    select_and_download()

not75743

動作の様子

not75743

各種バージョン

python 3.10.9
boto3 1.28.7
botocore 1.31.7
inquirerpy 0.3.4

not75743

流れ

s3バケットを選択する
s3バケットのオブジェクトを選択する。ファイルを選べばダウンロードへ、フォルダを選択したらそのフォルダを起点にして再度オブジェクトを選択する
ダウンロードする際のファイル名を指定する
ダウンロードできたことを確認する

not75743

参考

not75743

ポイント①

def list_objects(bucket_name, prefix=''):
    bucket = s3.Bucket(bucket_name)
    all_objects = [obj.key for obj in bucket.objects.filter(Prefix=prefix)]
    if prefix:
        all_objects = [obj.replace(prefix, '', 1) for obj in all_objects]
    return [obj for obj in all_objects if obj != '']

指定したs3バケット内のオブジェクトを一覧出力する関数です。
if prefixとしているのは、各オブジェクトを一覧表示した際に、ディレクトリ名を省略するためです。
このようにしないと、参照しているディレクトリ名が延々表示されます。

not75743

ポイント②

def select_and_download():
    selected_bucket = inquirer.select(
        message="Select an S3 bucket:",
        choices=list_buckets(),
    ).execute()

inquirer.selectのchoiceで選択肢として表示する内容を指定し、executeで質問を投げかけます。
今回はlist_buckets()関数を使用してs3バケットを一覧表示するようにしています。

not75743

ポイント③

    while True:
        objects = list_objects(selected_bucket, current_prefix)
        if not objects:
            print("No files found.")
            break

        selected_obj = current_prefix + inquirer.select(
            message="Select a file or a directory:",
            choices=objects,
        ).execute()

        if selected_obj.endswith('/'):  # Directory
            current_prefix = selected_obj
        else:  # File
            download_path = input(f'Enter the download path for {selected_obj}: ')
            download_file(selected_bucket, selected_obj, download_path)
            break

while Trueとすることで、breakが出現するまでコードをループするようにしています。
breakはファイルが見つからない場合、ファイルをダウンロードした場合に現れますので、それまではディレクトリの中身を一覧表示する動作を続けます。
結果的にディレクトリを深ぼっていく形となります。

not75743