【Python】Boto3を使ってPython でAWS S3を操作する

今回は、AWS のS3 から、必要なデータを（リネイムして）再アップロードしたいという要望があり、Python を用いて自動化したので備忘録として残しておく。

Boto3 とは
認証情報の設定方法（profile での設定）
S3からファイル一覧を取得する
S3から任意のファイルをダウンロードする
S3にファイルをアップロードする
まとめ

Boto3 とは

以前の記事で、今回同様に Python でAWSを操作する際に、AWS CLI を用いて行った記事を紹介した。

serip39.hatenablog.com

この時、今回紹介する、Boto3 の存在を知らなかったので、AWS CLI を力技で Python で動かしていた。しかしながら、Boto3 というステキなライブラリが存在していた。（教えてくれた師匠に感謝。）

aws.amazon.com

Boto3 とは、Python の AWS SDK であり、Boto3 を使用することで、Python のアプリケーション、ライブラリ、スクリプトを AWS の各種サービス（Amazon S3、Amazon EC2、Amazon DynamoDB など）と容易に統合することができる。めちゃくちゃ便利なライブラリである。

boto3.amazonaws.com

pip で簡単にインストールすることができ、すぐに利用することができる。

$ pip install boto3

認証情報の設定方法（profile での設定）

AWS の認証情報に関して、私は、configとcredentialsに情報を書き込み、profileで指定している。 profile 機能に関しては、以前の記事で書いたので参考にして欲しい。

serip39.hatenablog.com

Boto3 で profile を使う場合は、以下のように設定することで使うことができる。

from boto3.session import Session

PROFILE_NAME="profile1"
BUCKET_NAME="test"

session = Session(profile_name=PROFILE_NAME)
s3 = session.resource('s3')
s3bucket = s3.Bucket(BUCKET_NAME)
s3client = s3.meta.client

# clientのみ必要の場合
s3client = session.client('s3')

Session を用いて、認証情報を格納し、サービスクライアント及びリソースを作成できるようにする。

S3からファイル一覧を取得する

S3.Client.list_objects_v2 を使用する。このメソッドは、1回のリクエストで、1000件までしか取得することができない。そのため、response の IsTruncated で途中で中断されているかを確認することができる。'IsTruncated': True の場合には、続きから取得する必要があるので、request のパラメータ StartAfter に前回取得した最後の key を割り当ててあげることで、続きから取得することができる。

from boto3.session import Session

def getFilteredFilenames(file_names=[]):
    if len(file_names) == 0:
        start = ''
    else:
        print(file_names[-1])
        start = file_names[-1]
    
    response = s3client.list_objects_v2(
        Bucket=BUCKET_NAME,
        Prefix=PREFIX,
        StartAfter=start
    )

    if 'Contents' in response:
        file_names = [content['Key'] for content in response['Contents']]
        if 'IsTruncated' in response:
            return getFilteredFilenames(file_names)
    return file_names

if __name__ == '__main__':

    PROFILE_NAME='profile1'
    BUCKET_NAME = 'test'
    PREFIX = 'test'

    session = Session(profile_name=PROFILE_NAME)
    s3client = session.client('s3')

    print(getFilteredFilenames())

S3から任意のファイルをダウンロードする

S3.Bucket.download_fileまたは、S3.Client.download_fileを使用する。

import os, json 
from boto3.session import Session
import time

def getFilteredFilenames(file_names=[]):
     # 省略

def downloadS3(file_name):
    file_path = FILE_PATH + file_name
    s3bucket.upload_file(file_name, file_path)

if __name__ == '__main__':

    PROFILE_NAME='profile1'
    BUCKET_NAME = 'test'
    FILE_PATH='./download/'

    session = Session(profile_name=PROFILE_NAME)
　s3 = session.resource('s3')
　s3bucket = s3.Bucket(BUCKET_NAME)
　s3client = s3.meta.client

    file_names = getFilteredFilenames()

    for file_name in file_names:
        downloadS3(file_name)

S3にファイルをアップロードする

S3.Bucket.upload_file または、S3.Client.upload_fileを使用する。

以下の例では、所定のフォルダにあるファイルを全てS3にアップロードする場合を想定している。（1秒間隔でアップロードをするようにしている。）

import os, json 
from boto3.session import Session
import time

def uploadS3(file_name):
    file_path = FILE_PATH + file_name
    s3bucket.upload_file(file_path, file_name)

def getAllFileNamesInFolder():
    json_files = [pos_json for pos_json in os.listdir(FILE_PATH)]
    return json_files

if __name__ == '__main__':

    PROFILE_NAME='profile1'
    BUCKET_NAME = 'test'
    FILE_PATH='./download/'

    session = Session(profile_name=PROFILE_NAME)
    s3 = session.resource('s3')
    s3bucket = s3.Bucket(BUCKET_NAME)

    file_names = getAllFileNamesInFolder()

    for file_name in file_names:
        uploadS3(file_name)
        time.sleep(1)