Python(Boto3)を利用してAmazon S3にファイルをアップロードする【後編】

はじめに

こんにちは、クラウド事業部の丸山です。

前編に引き続き、今回はより実用的内容としてPythonからDBデータを操作し、DBファイルをAmazon S3へとアップロードします。
個人でデータベース操作を行いたい場合、サーバーを用意するのは手間がかかりますよね。
AWSなどのクラウド上でデータベースを作成するのは手軽ですが、コストが気になることもあります。
そこで、ローカル環境で利用できるデータベースと低コストで利用できるAmazon S3を組み合わせて、
この問題を解決する方法を提案します。

前編では基本的なファイルのアップロード方法について記載しています。 techblog.ap-com.co.jp

どんなひとに読んで欲しい

AWSに興味がある方
Pythonを使ってAWSと連携する方法に興味がある方
冗長性のあるDBデータを扱いたいが、コストを抑えたい方

www.sqlite.org

docs.python.org

qiita.com

sqlite3とは

SQLiteは、ローカル環境で簡易的に利用できるデータベースです。
サーバーを必要とせず、組み込み型データベースとして機能します。
スマホアプリや軽量アプリケーションの開発によく利用されており、その簡便さと操作性、そして高速なパフォーマンスに定評があります。

前提

Python3をインストール済みの環境で実施。
AWS CLIをインストール済みの環境で実施。
※今回はAWS CLIの設定にてプロファイルを設定して、プログラムコードの中に認証情報を持たないセキュアな作りとします。
SQLの知識、sqliteの基本的な操作については関連記事などを参考に操作できるものとします。

手順

AWS CLIのセットアップ

AWS CLIがインストールされていない場合はこちらを参考にインストールを実施。 docs.aws.amazon.com

プロファイル設定

以下のコマンドを利用しAWS CLI用のプロファイルを作成します。

(.venv) user@MachineName Directory % aws configure --profile sample-user
AWS Access Key ID [None]: YOUR_ACCESS_KEY #実際に取得したアクセスキーを入力する。
AWS Secret Access Key [None]: YOUR_SECRET_KEY #実際に取得したアクセスキーを入力する。
Default region name [None]: ap-northeast-1
Default output format [None]: 
(.venv) user@MachineName Directory %

コード処理の作成

今回は簡単なコードとして、プログラム実行でデータを作成し、そのままアップロードする例をコードを想定します。
ここではランダムな数値を作成して、SQLite3にINSERTする機能と、作成したデータを削除する機能のみを実装しています。
また、どちらの処理後にもS3へのアップロードを確認するプロンプトを表示する簡単な構造とします。

全文を表示

import sqlite3
import string
import random
import datetime
import boto3

def upload_to_s3(file_name, bucket, object_name=None):
    """
    ファイルをS3バケットにアップロードする関数

    :param file_name: アップロードするファイル
    :param bucket: アップロード先のS3バケット
    :param object_name: S3オブジェクト名。指定しない場合はfile_nameが使用される
    :return: アップロードが成功すればTrue、失敗すればFalse
    """
    if object_name is None:
        object_name = file_name

    session = boto3.session.Session(profile_name='sample-user')
    s3_client = session.client('s3')

    try:
        s3_client.upload_file(file_name, bucket, object_name)
        print(f"Successfully uploaded {file_name} to {bucket}/{object_name}")
    except Exception as e:
        print(f"S3 Upload Error: {e}")
        return False
    return True

def generate_random_data():
    chars = string.ascii_letters
    while True:
        data = ''.join(random.choice(chars) for _ in range(10))
        yield data

def main():
    try:
        conn = sqlite3.connect('random_data.db')
        cursor = conn.cursor()

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS data(
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                random_data TEXT,
                creation_date TEXT)
        """)

        while True:
            menu_option = input(
                "\n選択肢:\n1. データ作成\n2. データ削除\nq. 終了\n番号を選んでください: ")

            if menu_option in ["q", "quit"]:
                print("終了します。")
                break

            elif menu_option == "1":
                num_records = int(input("作成するランダムデータの数を入力してください: "))
                generator = generate_random_data()
                current_date = datetime.datetime.now().strftime('%Y-%m-%d')

                for _ in range(num_records):
                    random_data = next(generator)
                    cursor.execute("INSERT INTO data (random_data, creation_date) VALUES (?, ?)",
                                   (random_data, current_date))
                conn.commit()

                cursor.execute(
                    "SELECT * FROM data ORDER BY id DESC LIMIT ?", (num_records,))
                new_records = cursor.fetchall()
                print(f"新しく作成されたデータ: {new_records}")

                if input("S3にアップロードしますか？ (y/n): ").lower() == 'y':
                    upload_to_s3('random_data.db', 'db-upload-bucket')
                    print("データベースのバックアップが完了しました。")

            elif menu_option == "2":
                cursor.execute("SELECT COUNT(*) FROM data")
                total_records = cursor.fetchone()[0]
                print(f"現在のデータ件数: {total_records}")

                delete_option = input("全てのデータを削除しますか？ (y/n): ").lower()
                if delete_option == 'y':
                    cursor.execute("DELETE FROM data")
                    conn.commit()
                    print("全てのデータを削除しました。")

                    if input("データベースをバックアップしますか？ (y/n): ").lower() == 'y':
                        upload_to_s3('random_data.db', 'db-upload-bucket')
                        print("データベースのバックアップが完了しました。")
                else:
                    print("データの削除をキャンセルしました。")

    except Exception as e:
        print(f"エラーが発生しました: {e}")
    finally:
        conn.close()

if __name__ == "__main__":
    main()

実行イメージ

～作成時～

(.venv) user@MachineName Directory % python3 /Users/user/Directory/sample.py

選択肢:
1. データ作成
2. データ削除
q. 終了
番号を選んでください: 1
作成するランダムデータの数を入力してください: 1
新しく作成されたデータ: [(1, 'wltPhJMpzX', '2024-05-18')]
S3にアップロードしますか？ (y/n): y
Successfully uploaded random_data.db to db-upload-bucket/random_data.db
データベースのバックアップが完了しました。

～削除時～

現在のデータ件数: 1
全てのデータを削除しますか？ (y/n): y
全てのデータを削除しました。
データベースをバックアップしますか？ (y/n): y
Successfully uploaded random_data.db to db-upload-bucket/random_data.db
データベースのバックアップが完了しました。

(.venv) user@MachineName Directory %

実行後

アップロードされていることが確認できます。

おわりに

SQLiteは、少しだけデータベースを操作したいけれどコストが気になる際に、気軽に利用できる方法です。
SQLiteをクラスタリングする方法もありますが、環境構築の手間がかかります。
そのような場合、Amazon S3にデータをそのままアップロードできる簡易的なバックアップシステムを備えておくことで、
データの保管についても安心できます。

お知らせ

APCはAWS Advanced Tier Services(アドバンストティアサービスパートナー)認定を受けております。

その中で私達クラウド事業部はAWSなどのクラウド技術を活用したSI/SESのご支援をしております。
www.ap-com.co.jp

https://www.ap-com.co.jp/service/utilize-aws/

また、一緒に働いていただける仲間も募集中です！
今年もまだまだ組織規模拡大中なので、ご興味持っていただけましたらぜひお声がけください。

APC 技術ブログ

株式会社エーピーコミュニケーションズの技術ブログです。