PythonのBoto3ライブラリを使ってAWS S3からオブジェクトをダウンロードする方法

PythonのBoto3ライブラリを使用すると、AWS S3に保存されたオブジェクトを簡単にローカル環境へダウンロードできます。本記事では、指定したS3パスからオブジェクトを取得し、既定のローカルパスまたは任意のパスに保存する関数の実装方法を解説します。既存ファイルの上書き可否を制御するオプションや、404エラーなどの例外処理もあわせて紹介します。

例として、「Bucket_1」バケット内の「testfolder」に格納された「test.zip」をダウンロードするケースを取り上げます。

解決のためのアプローチとアルゴリズム

ステップ1 − 例外処理のためにboto3とbotocoreの例外クラス（ClientError）をインポートします。

ステップ2 − pathlibモジュールからPathをインポートし、ファイル名のチェックに使用できるようにします。

ステップ3 − 関数download_object_from_s3は、s3_path（S3上のパス）、localpath（ローカルの保存先）、overwrite_existing_file（既存ファイルの上書き可否）の3つの引数を受け取ります。

ステップ4 − s3_pathが「s3://bucket_name/key」というAWS標準形式で渡されているかを検証します。localpathのデフォルト値はNone、overwrite_existing_fileのデフォルト値はTrueです。ユーザーがこれらの値を明示的に渡せば、指定したローカルパスへのダウンロードも可能です。

ステップ5 − Boto3ライブラリを使ってAWSセッションを作成します。

ステップ6 − S3用のAWSリソースオブジェクトを生成します。

ステップ7 − S3パスを分割し、ルートとなるバケット名とダウンロード対象のオブジェクトパスを切り出します。

ステップ8 − overwrite_existing_fileがFalseに設定されており、かつ指定されたローカルパスに同名ファイルがすでに存在する場合は、何も処理を行いません。

ステップ9 − 上記以外の場合（いずれかの条件が成立しない場合）はオブジェクトをダウンロードします。localpathが指定されていればそこへ、指定がなければデフォルトのパスへ保存します。

ステップ10 − レスポンスコードに基づいて例外をハンドリングし、ファイルが正常にダウンロードされたかどうかを確認します。

ステップ11 − ダウンロード中に予期しない問題が発生した場合に備え、汎用的な例外処理も実装します。

コード例

AWS S3からファイルをダウンロードするには、以下のコードを使用します。

import boto3
from botocore.exceptions import ClientError
from pathlib import Path

def download_object_from_s3(s3path, localPath=None,
overwrite_existing_file=True):

   if 's3://' not in s3path:
      print('指定されたパスは有効なS3パスではありません。')
      raise Exception('指定されたパスは有効なS3パスではありません。')

   session = boto3.session.Session()
   s3_resource = session.resource('s3')

   s3_tokens = s3path.split('/')
   bucket_name = s3_tokens[2]
   object_path = ""
   filename = s3_tokens[len(s3_tokens) - 1]
   print('ファイル名: ' + filename)

   if len(s3_tokens) > 4:
      for tokn in range(3, len(s3_tokens) - 1):
         object_path += s3_tokens[tokn] + "/"
      object_path += filename
   else:
      object_path += filename
   print('オブジェクト: ' + object_path)
   try:
      if not overwrite_existing_file and Path.is_file(filename):
         pass
      else:
         if localPath is None:
            s3_resource.meta.client.download_file(bucket_name, object_path, filename)
         else:
            s3_resource.meta.client.download_file(bucket_name, object_path, localPath + '/' + filename)
      print('ファイル名: ' + filename)
      return filename
   except ClientError as error:
      if error.response['Error']['Code'] == '404':
         print(s3path + " ファイルが見つかりません: ")
         raise Exception(s3path + " ファイルが見つかりません: ")
   except Exception as error:
      print("s3ヘルパーのdownload_object関数で予期しないエラーが発生しました: " + error.__str__())
      raise Exception("s3ヘルパーのdownload_object関数で予期しないエラーが発生しました: " + error.__str__())

#デフォルトのローカルパスにダウンロード
print(download_object_from_s3("s3://Bucket_1/testfolder/test.zip"))
#指定したパスにダウンロード
print(download_object_from_s3("s3://Bucket_1/testfolder/test.zip","C://AWS"))
#S3にファイルが存在しない場合
print(download_object_from_s3("s3://Bucket_1/testfolder/abc.zip"))

出力結果

#デフォルトのローカルパスにダウンロードした場合
Filename: test.zip
object: testfolder/test.zip
Filename: test.zip

#指定したパスにダウンロードした場合
Filename: test.zip
object: testfolder/test.zip
Filename: test.zip

#S3にファイルが存在しない場合
Filename: abc.zip
object: testfolder/abc.zip
s3://Bucket_1/testfolder/abc.zip File not found:
botocore.exceptions.ClientError: An error occurred (404) when calling
the HeadObject operation: Not Found

補足： ダウンロード先のデフォルトパスは、この関数が記述されているディレクトリです。ローカルパスを指定しなかった場合、ファイルは同じディレクトリ内にダウンロードされます。

たとえば、この関数が「S3_class」というクラス内に記述されており、そのクラスが「C://AWS/src/S3_class」に配置されている場合、「test.zip」は「C://AWS/src/test.zip」にダウンロードされます。