Zenn
🔄

【AWS S3】 Python boto3 を使い操作してみる

2021/09/14に公開
4件
AWS
Docker
Python
S3
#
boto3
tech

はじめに

AWS S3 を API boto3 を使って Bucket 作成から操作してみました。
エラーハンドリング・ログ吐き出しなど荒いのでツッコミどころあるかと。
力不足なので、引き続き実装は継続して続けていきます!!

とりあえず、意図した動作は実装できたので流れを残しておこうかと。

やりたいことイメージ

できたこと

  • S3 バッケット:【生成 / 削除】
  • S3 ファイル:【アップロード / ダウンロード /削除】

後に試したいこと

  • response の値を使って単体テストを書いてみる
  • mock もやってみたい
  • Lambda で定期自動化

動作環境

  • PC: Mac M1
  • Python 3.9.1 (virtual env)
  • Python 3.8.8 (Docker JupyterLab)

リポジトリ

色々試しているリポジトリになるので、本記事で書いた動作以外も含まれています🙏

▼ リポジトリをこちらに作成し直しました!
Github: aws

前提条件

  • AWS アカウントあり
  • AWS IMA作成済

前処理

ローカルで、IMAユーザの設定が必要だったので、AWS CLI をダウンロードしました。
他にも、環境変数に持たせて IMAユーザを呼び出す方法がありそうでしたが、試してませんー

AWS CLI のダウンロード #1

(※ AWS CLI のダウンロードは、Dockerでは Dockrfile に書いたので不要)

  • macOS pkg ファイルをダウンロード

    最新バージョンの AWS CLI をインストール

  • シンボリックリンクを手動で作成 PATH を通す

bash
$ sudo ln -s /folder/installed/aws-cli/aws /usr/local/bin/aws
$ sudo ln -s /folder/installed/aws-cli/aws_completer /usr/local/bin/aws_completer
  • インストールの検証
bash
$ which aws
/usr/local/bin/aws

$ aws --version
aws-cli/2.2.36 Python/3.8.8 Darwin/20.6.0 exe/x86_64 prompt/off

Configuration の設定

AWS CLI をインストールしたので下記コマンドで以前作成していたIAMユーザ(作成方法は割愛)でaws にアクセスできるように設定します。

bash
# 実行コマンド
$ aws configure

# 聞かれる項目を入力
AWS Access Key ID [None]: <Access Key ID> # IAM の　key
AWS Secret Access Key [None]: <Secret Access Key> # IAM の　key
Default region name [None]: ap-northeast-1 # #2を参考にtokyoを選択
Default output format [None]: json # とりあえず json

ツリー構成

  • 以下のファイルで実装しました!
tree
├── api
│   ├── __init__.py
│   ├── controllers
│   │   ├── bucket_controller.py # S3操作
│   │   └── db_controller.py
│   ├── models
│   │   ├── __init__.py
│   │   └── bucket.py
├── config
│   ├── __init__.py
│   ├── const.py # 定数をまとめている
│   └── logger_tool.py
├── data # s3 upload用
│   ├── response.json
│   ├── test.sql
│   ├── user.json
│   └── user.sql
├── log
│   └── bucket.log # S3操作出力ファイル
├── main.py # 実行ファイル
├── tmp # S3 dwonload 用
│   ├── user.json
│   └── user.sql

作成ファイル

model

S3 Bucket を操作する model を作成しました。
基本 Bucket model に機能を追加していく予定です!!

今のところ、下記を実装しました!

  • S3 バッケット:【 生成 / 削除 】
  • S3 ファイル:【 単体: アップロード / ダウンロード / 削除 】

ログ取得、例外処理が甘いと感じておりますが、とりあえず進めます。

logger が多くてみにくい...。

bucket 削除も一括じゃなくて、指定 bucket のみ削除するようにしていきたいと考えています。

api/models/bucket.py
import boto3

from botocore.exceptions import ClientError
from pathlib import Path

from config import const, logger_tool


class Bucket(object):
    """
    AWS S3 Bucket 操作する model
    """

    def __init__(self, client: str, bucket_name: str, region: str):
        """ initialization parameter

        params
        ------
            client(str): s3
            bucket_name(str): choice bucket name
            region(str): Region code
        """
        self.client = boto3.client(client)
        self.resource_bucket = boto3.resource(client)
        self.bucket_name = bucket_name
        self.region = region

    def create_bucket(self):
        """ Bucket を作成する処理

        params
        -------
            max_key(int): default: 2
                is_bucket_check fuc args
        """
        logger_tool.info(
            module=__name__,
            action='create',
            status='run',
            bucket_name=self.bucket_name
        )

        try:
            self.client.create_bucket(
                Bucket=self.bucket_name,
                CreateBucketConfiguration={'LocationConstraint': self.region}
            )

        # 作成 Bucket名が既に存在するか判定。即処理停止
        except self.client.exceptions.BucketAlreadyExists as err:
            logger_tool.error(
                module=__name__,
                action='create',
                status=409,
                bucket_name=self.bucket_name,
                ex=err,
                msg="Bucket {} already exists!".format(err.response['Error']['BucketName']),
            )
            raise err

        logger_tool.info(
            module=__name__,
            action='create',
            status=200,
            bucket_name=self.bucket_name
        )

    def upload_data(self, bucket_name: str, upload_data: str):
        """
        S3に 1 データ upload する処理

        params
        ------
            upload_data: ex) sample.png
        """
        key = upload_data.split('/')[-1]

        logger_tool.info(
            module=__name__,
            action='upload',
            status='run',
            bucket_name=bucket_name,
            data=key,
        )

        with open(upload_data, 'rb') as upload_data_file:
            self.resource_bucket.Bucket(bucket_name).put_object(Key=key, Body=upload_data_file)

        logger_tool.info(
            module=__name__,
            action='upload',
            status=200,
            bucket_name=bucket_name,
            data=key,
        )

    def download_data(self, download_data: str) -> None:
        """
        Bucket にあるファイルをダウンロードする

        params
        ------
            download_data(str): S3にある指定データを download する
        """

        # download 用に作成
        Path(const.TMP_PATH).mkdir(exist_ok=True)

        logger_tool.info(
            module=__name__,
            action='download',
            status='run',
            bucket_name=self.bucket_name,
            data=download_data,
        )
        try:
            self.resource_bucket.meta.client.download_file(
                self.bucket_name, download_data, f"tmp/{download_data}"
            )
            logger_tool.info(
                module=__name__,
                action='download',
                status=204,
                bucket_name=self.bucket_name,
                data=download_data,
            )
        except ClientError as ex:
            logger_tool.error(
                module=__name__,
                action='download',
                status=404,
                bucket_name=self.bucket_name,
                ex=ex
            )

    def delete_data(self, bucket_name: str, delete_data: str):
        """
        bucketにあるデータを削除

        params
        ------
            bucket_name(str):  Bucket name
            delete_data(data): 削除するファイル名

        return
        ------
            response: 削除が成功した結果を返すレスポンス parameter
        """

        key = delete_data.split('/')[-1]

        logger_tool.info(
            module=__name__,
            action='delete',
            status='run',
            bucket_name=bucket_name,
            data=key,
        )
        try:
            response = self.client.delete_object(
                Bucket=bucket_name,
                Key=key,
            )
            return response

        except ClientError as ex:
            logger_tool.error(
                module=__name__,
                action='delete data',
                status=404,
                bucket_name=bucket_name,
                ex=ex
            )

        logger_tool.info(
            module=__name__,
            action='delete',
            status=204,
            bucket_name=bucket_name,
            data=key,
        )

    ・・・

    def delete_all_buckets(self):
        """
        全ての bucketを削除する

        return
        ------
            response: bucket を削除した結果を返す response parameter
        """

        buckets = list(self.resource_bucket.buckets.all())

        if buckets:
            logger_tool.info(
                module=__name__,
                action='delete',
                status='run',
                bucket_name=self.bucket_name
            )

            response = [key.delete() for key in self.resource_bucket.buckets.all()]

            logger_tool.info(
                module=__name__,
                action='delete',
                status=204,
                bucket_name=self.bucket_name
            )
            return response

ログ吐き出しパラメータを設定

{ key: value } で出力したいパラメータを設定

ログ吐き出しファイルの調整・行数をパラメータに入れてみたかったんですが割愛。
追々できればいいなっと考えています。

config/logger_tool.py
import logging
from pathlib import Path

from config import const

# CREATE DIR
Path(const.LOGFILE_PATH).mkdir(exist_ok=True)

# 表示されるメッセージを設定
FORMAT = '%(asctime)s %(levelname)s:%(message)s'

logging.basicConfig(format=FORMAT, filename=Path(f"{const.LOGFILE_PATH}{const.LOG_FILE}"), level=logging.INFO)


def error(module, action, status, bucket_name: str, ex, msg=None):
    logging.error({
        'module': module,
        'action': action,
        'status': status,
        'bucket name': bucket_name,
        'except': ex,
        'msg': msg,
    })


def info(module, action=None, status=None, bucket_name=None, data=None):
    logging.info({
        'module': module,
        'action': action,
        'status': status,
        'bucket name': bucket_name,
        'data': data,
    })

const

定数としてまとめたかったので、一括にまとめた形です。
厳密には、呼び出した時に、書き換えできちゃうのであしからず。

config/const.py
# 実行は 0: False, 1: True で切り分け

CLIENT = 's3'
BUCKET_NAME = 'testbucket-yyyymmdd'
TOKYO_REGION = 'ap-northeast-1'

LOGFILE_PATH = 'log/'
LOG_FILE = 'bucket.log'
TMP_PATH = 'tmp/'

BUCKET_CREATE = 1
BUCKET_DELETE = 1

UPLOAD = 1
DOWNLOAD = 0
DELETE = 1

因みに、上の設定でスクリプト実行しますと、

  1. S3 にアクセス
  2. バケットを作成
  3. data/ 配下 のデータを アップロード
  4. すぐに、S3 にアップロードした データを削除
  5. バケットを削除

となります。

動きとしては、バケット作成して最後にすぐ消しています。
何やってんの ... っとなりますが、まあ動作確認としては良さそうです。

controller

bucket_controller.py でS3操作をできるように調整しました。
ほんまですと、バッチファイルを作成して定期実行のジョブを組むんでしょうがやった事ないので保留。それか、Lambad で実行するんですかね🤔

また、次回以降試してみるのも面白そうですwww

api/controllers/bucket_controller.py
from config import const
from api.models import bucket


def s3_bucket_management():
    """ aws s3 management """
    management_bucket = bucket.Bucket(client=const.CLIENT, bucket_name=const.BUCKET_NAME, region=const.TOKYO_REGION)

    # TODO: 指定 Directory のファイルを格納する
    data_list = ['user.sql', 'user.json']

    if const.BUCKET_CREATE:
        management_bucket.create_bucket()

    if const.UPLOAD:
        for data in data_list:
            management_bucket.upload_data(bucket_name=const.BUCKET_NAME, upload_data=f"data/{data}")

    if const.DOWNLOAD:
        for data in data_list:
            management_bucket.download_data(download_data=data)

    if const.DELETE:
        for data in data_list:
            management_bucket.delete_data(bucket_name=const.BUCKET_NAME, delete_data=f"data/{data}")

    if const.BUCKET_DELETE:
        management_bucket.delete_all_buckets()

    # 最終的に存在している Bucketを確認する
    management_bucket.print_bucket_name()

main でスクリプト実行

最後に、実行処理を、main.py に書いておけば良さそうです。

main.py
import api.controllers.bucket_controller

if __name__ == '__main__':
    api.controllers.bucket_controller.s3_bucket_management()

実行結果

main.py を実行し結果はこんな感じです!

bucket.log
# IMA ユーザ で設定した認証情報を確認
2021-09-18 23:48:34,441 INFO:Found credentials in shared credentials file: ~/.aws/credentials

# Bucketを生成
2021-09-18 23:49:47,531 INFO:{'module': 'api.models.bucket', 'action': 'create', 'status': 'run', 'bucket name': 'testbucket-yyyymmdd', 'data': None}
2021-09-18 23:50:28,226 INFO:{'module': 'api.models.bucket', 'action': 'create', 'status': 200, 'bucket name': 'testbucket-yyyymmdd', 'data': None}

# data/ 配下に格納していた指定データを S3 に upload
2021-09-18 23:50:28,231 INFO:{'module': 'api.models.bucket', 'action': 'upload', 'status': 'run', 'bucket name': 'testbucket-yyyymmdd', 'data': 'user.sql'}
2021-09-18 23:50:28,627 INFO:{'module': 'api.models.bucket', 'action': 'upload', 'status': 200, 'bucket name': 'testbucket-yyyymmdd', 'data': 'user.sql'}
2021-09-18 23:50:28,627 INFO:{'module': 'api.models.bucket', 'action': 'upload', 'status': 'run', 'bucket name': 'testbucket-yyyymmdd', 'data': 'user.json'}
2021-09-18 23:50:28,768 INFO:{'module': 'api.models.bucket', 'action': 'upload', 'status': 200, 'bucket name': 'testbucket-yyyymmdd', 'data': 'user.json'}

# そしてすぐさまデータ削除
2021-09-18 23:50:34,111 INFO:{'module': 'api.models.bucket', 'action': 'delete', 'status': 'run', 'bucket name': 'testbucket-yyyymmdd', 'data': 'user.sql'}
2021-09-18 23:50:34,380 INFO:{'module': 'api.models.bucket', 'action': 'delete', 'status': 'run', 'bucket name': 'testbucket-yyyymmdd', 'data': 'user.json'}

# んで、Bucket を削除
2021-09-18 23:50:35,003 INFO:{'module': 'api.models.bucket', 'action': 'delete', 'status': 'run', 'bucket name': 'testbucket-yyyymmdd', 'data': None}
2021-09-18 23:50:35,682 INFO:{'module': 'api.models.bucket', 'action': 'delete', 'status': 204, 'bucket name': 'testbucket-yyyymmdd', 'data': None}

ログの中身をみていきます。

意図した流れで処理実行できているようです。

気になる点は、INFO:api.models.logger_tool とlogger_tool のファイル名でログが吐かれているところ!
ここは書き出されるファイル名でログ吐き出しできるように調整したいです。

改善必須です。

【追記 2021/09/19】

logging.basicConfig で format 指定したらログ出力はカスタマイズできますよ。

のコメントを参考にファイル名・時間表示のログ出力を調整させて頂きました!ありがとうございます🙏

まとめ

まだまだ、実用では使えないレベルですが、S3 の操作が軽く理解できたので良かったです。

次は、大量データの一括アップロード など試してみたいですね。

私のアウトプットを込めた内容になっております。
もっとこう書いた方が良いなど、気軽にコメント頂けると幸いです🙏

ありがとうございました!!

参考

GitHubで編集を提案

Discussion

jkobaxjkobax

logging.basicConfig で format 指定したらログ出力はカスタマイズできますよ。

is_bucket_check() は bucket_exists() とかにしても良さそうですね。
https://qiita.com/yskszk/items/5a7f99c974773f03a82a

なかがわなかがわ

コメントありがとうございます!

is_bucket_check() は bucket_exists() とかにしても良さそうですね。

そうですね。命名は bucet_exists()が適切ですね!!
ログ出力のカスタマイズも合わせて試してみます!