概要

この記事の応用編です。指定した件数・間隔でデータを自動生成し、MySQLサーバのデータベースに書き込む Python プログラムです。テーブルの初期化、データの読み込み、データの削除も可能となっています。

ローカル環境

macOS Monterey 12.3
python 3.8.12
Azure CLI 2.34.1
mysql Ver 8.0.28 for macos12.2 on x86_64 (Homebrew)

前提条件

Azure環境がすでに用意されていること（テナント/サブスクリプション）
ローカル環境に「azure cli」がインストールされていること
ローカル環境に「mysql」がインストールされていること

事前準備

この記事を実行し、Azure Database for MySQL を Single構成で作成できていること

項目	値
MySQLサーバ名	iturumysql01.mysql.database.azure.com
データベース名	iotdummydb
管理者名	adminadmin@iturumysql01
管理者パスワード	HogeHogeHoge!

PythonプログラムからのMySQLサーバへのDB操作

Pythonプログラム

作成するテーブル名：inventory

mysql_IoTdummy.py

import mysql.connector
from mysql.connector import errorcode
import time
from datetime import date, datetime
import random
import json
import argparse
import string
from faker.factory import Factory
import pandas as pd
import numpy as np
from tqdm import tqdm

# ダミーデータ作成のための Faker の使用
Faker = Factory.create
fake = Faker()
fake = Faker("ja_JP")

# ダミーセクション？(大文字アルファベットを定義)
section = string.ascii_uppercase

# MySQLサーバ接続設定情報
config = {
    'host': 'iturumysql01.mysql.database.azure.com',
    'user': 'adminadmin@iturumysql01',
    'password': 'HogeHogeHoge!',
    'database': 'iotdummydb'
}


# ダミーデータの作成
def iot_json_data(count, proc):
    iot_items = json.dumps({
        'items': [{
            'iot_id': i,                        # IoT id
            'proc': proc,                       # データ生成プロセス名
            'section': random.choice(section),  # IoT機器セクション
            'iot_num': fake.zipcode(),          # IoT機器番号
            'iot_state': fake.prefecture(),     # IoT設置場所
            'val_1': random.uniform(100, 200),  # IoT値−１
            'val_2': random.uniform(50, 90),    # IoT値−２
            'created_at': generate_time()       # データ生成時間
            } 
            for i in range(count)
        ]
    }, ensure_ascii=False).encode('utf-8')
    return iot_items

# ダミーデータの生成時間
def generate_time():
    dt_time = datetime.now()
    gtime = json_trans_date(dt_time)
    return gtime

# date, datetimeの変換関数
def json_trans_date(obj):
    # 日付型を文字列に変換
    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    # 上記以外は対象外.
    raise TypeError ("Type %s not serializable" % type(obj))


# MySQLサーバ接続
def ConnectionToMySQL():
    try:
      conn = mysql.connector.connect(**config)
      print("\nConnection established\n")
    except mysql.connector.Error as err:
      if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
        return 0, "Something is wrong with the user name or password"
      elif err.errno == errorcode.ER_BAD_DB_ERROR:
        return 0, "Database does not exist"
      else:
        return 0, err
    else:
      cursor = conn.cursor()
      return 1, conn


# メイン : ターミナル出力用
def tm_main(count, proc, wait):
    print('ターミナル 出力\n')

    # ダミーデータ生成
    iotjsondata = iot_json_data(count, proc)
    json_dict = json.loads(iotjsondata)

    # ターミナル出力
    for item in json_dict['items']:
        print(item)
        time.sleep(wait)


# メイン : Show Database
def show_db_main():
    print('MySqlサーバへの接続確認')
    code, conn = ConnectionToMySQL()

    if code == 0 :
      print(conn)
    else :
      print(conn)
      conn.close()


# メイン : Init Database
def init_db_main():
    print('データベース・テーブルの新規作成')

    # DBへの接続
    code, conn = ConnectionToMySQL()

    if code == 0 :
      print(conn)
      return
    else :
      cursor = conn.cursor()

    # Drop previous table of same name if one exists
    cursor.execute("DROP TABLE IF EXISTS inventory;")
    print("\tFinished dropping table (if existed).")

    # Create table
    cursor.execute("\
      CREATE TABLE inventory (\
        id int NOT NULL AUTO_INCREMENT PRIMARY KEY, \
        iot_id int NOT NULL, \
        proc VARCHAR(10) NOT NULL, \
        section VARCHAR(2) NOT NULL, \
        iot_num VARCHAR(10) NOT NULL, \
        iot_state VARCHAR(10) NOT NULL, \
        val_1 float NOT NULL, \
        val_2 float NOT NULL, \
        created_at datetime DEFAULT NULL, \
        updated_at datetime DEFAULT NULL \
      );"\
    )
    print("\tFinished creating table.\n")

    # 作成したテーブル情報の取得
    cursor.execute("show columns from inventory;")
    rows = cursor.fetchall()
    # 多次元リストを Pandas DataFrame を利用して表示する
    df = pd.DataFrame(rows, columns = ['Field','Type','Null','Key','Default','Extra'])
    print(df)

    # Cleanup
    cursor.close()
    conn.close()
    print("Done.")


# メイン : Write Database
def write_db_main(count, proc, wait):
    print('データベース・テーブルへのデータ書き込み')

    # DBへの接続
    code, conn = ConnectionToMySQL()

    if code == 0 :
      print(conn)
      return
    else :
      cursor = conn.cursor()

    # ダミーデータ生成
    iotjsondata = iot_json_data(count, proc)
    json_dict = json.loads(iotjsondata)

    # インサート項目の定義
    insert_sql = "INSERT INTO inventory \
                  (iot_id, proc, section, iot_num, iot_state, val_1, val_2, created_at) \
                  values (%s,%s,%s,%s,%s,%s,%s,%s);"

    # プログレスバーで進捗状況を表示
    for item in tqdm(json_dict['items']):
      # データのインサート
      # print(list(item.values()))
      cursor.execute(insert_sql, list(item.values()))
      conn.commit()
      time.sleep(wait)

    # Cleanup
    cursor.close()
    conn.close()
    print("Inserted",count,"row(s) of data.")
    print("Done.")


# メイン : Read Database
def read_db_main():
    print('データベース・テーブルへのデータ読み込み')

    # DBへの接続
    code, conn = ConnectionToMySQL()

    if code == 0 :
      print(conn)
      return
    else :
      cursor = conn.cursor()

    # Read table data
    cursor.execute("SELECT * FROM inventory;")
    rows = cursor.fetchall()
    print(f'Read: {cursor.rowcount}, row(s) of data.')

    # print all rows
    [print(f'Data row = {row}') for row in rows]

    # Cleanup
    cursor.close()
    conn.close()
    print("Done.")


# メイン : Delete Database
def delete_db_main():
    print('データベース・テーブル・データの全削除')

    # DBへの接続
    code, conn = ConnectionToMySQL()

    if code == 0 :
      print(conn)
      return
    else :
      cursor = conn.cursor()

    # Delete table data
    cursor.execute("DELETE FROM inventory;")
    rows = cursor.fetchall()
    print(rows)
    conn.commit()

    # Cleanup
    cursor.close()
    conn.close()
    print("Done.")


# メイン
if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='ダミーデータの自動生成からのDBへの書込み')
    parser.add_argument('--count', type=int, default=5, help='データ作成件数（デフォルト：5件）')
    parser.add_argument('--proc', type=str, default='111', help='データ作成プロセス名（デフォルト：111）')
    parser.add_argument('--mode', type=str, default='tm', help='tm（Terminal(デフォルト)）/ db（db操作）')
    parser.add_argument('--wait', type=float, default=1, help='データ生成間隔（デフォルト：1.0秒）')
    parser.add_argument('--db', type=str, default='show', help='show(デフォルト) / init / write / read / delete')
    args = parser.parse_args()

    start = time.time()

    if (args.mode == 'tm'): 
      tm_main(args.count, args.proc, args.wait)
    elif (args.mode == 'db'):
      if (args.db == 'show'):
        show_db_main()
      elif (args.db == 'init'):
        init_db_main()
      elif (args.db == 'write'):
        write_db_main(args.count, args.proc, args.wait)
      elif (args.db == 'read'):
        read_db_main()
      elif (args.db == 'delete'):
        delete_db_main()
    else :
      print("パラメータ設定を確認ください --help")

    making_time = time.time() - start

    print("")
    print("処理時間:{0}".format(making_time) + " [sec]")
    print("")

プログラムの実行

実行パラメータを指定することにより以下の処理が可能となります
- データ生成
  - パラメータ
    - --mode
      - tm : 生成データのターミナル出力（デフォルト）
      - db : DB操作の実施
    - --proc : 本プログラム実行場所の識別子（デフォルト：'111'）
    - --count : 生成データの件数
    - --wait : データ書込み時の間隔（秒）（デフォルト：1.0秒、最小：0秒）
- DB操作
  - パラメータ（上記「mode」パラメータで「db」を選択することにより有効となります）
    - --db
      - show : データベースへの接続確認（デフォルト）
      - init : テーブルの新規作成（存在していれば削除後に再作成）
      - write : テーブルへの生成データ書き込み
      - read : テーブルからの全件データ読み込み
      - delete : テーブルの全件データ削除

ヘルプ表示

$ python mysql_IoTdummy.py --help
usage: mysql_IoTdummy.py [-h] [--count COUNT] [--proc PROC] [--mode MODE] [--wait WAIT] [--db DB]

ダミーデータの自動生成からのDBへの書込み

optional arguments:
  -h, --help     show this help message and exit
  --count COUNT  データ作成件数（デフォルト：5件）
  --proc PROC    データ作成プロセス名（デフォルト：111）
  --mode MODE    tm（Terminal(デフォルト)）/ db（db操作）
  --wait WAIT    データ生成間隔（デフォルト：1.0秒）
  --db DB        show(デフォルト) / init / write / read / delete

ターミナルへの生成データの出力

$ python mysql_IoTdummy.py       
ターミナル 出力

{'iot_id': 0, 'proc': '111', 'section': 'B', 'iot_num': '080-6847', 'iot_state': '島根県', 'val_1': 140.61912005313204, 'val_2': 72.17919110391685, 'created_at': '2022-04-02T00:25:52.683694'}
{'iot_id': 1, 'proc': '111', 'section': 'A', 'iot_num': '065-8086', 'iot_state': '兵庫県', 'val_1': 186.96659888711127, 'val_2': 65.56487417259912, 'created_at': '2022-04-02T00:25:52.683717'}
{'iot_id': 2, 'proc': '111', 'section': 'G', 'iot_num': '879-6045', 'iot_state': '京都府', 'val_1': 181.3697189155115, 'val_2': 79.14644988820804, 'created_at': '2022-04-02T00:25:52.683730'}
{'iot_id': 3, 'proc': '111', 'section': 'P', 'iot_num': '798-2794', 'iot_state': '京都府', 'val_1': 160.0290794152534, 'val_2': 71.33695982774306, 'created_at': '2022-04-02T00:25:52.683742'}
{'iot_id': 4, 'proc': '111', 'section': 'C', 'iot_num': '026-9269', 'iot_state': '三重県', 'val_1': 193.99969925933686, 'val_2': 85.21973561601754, 'created_at': '2022-04-02T00:25:52.683753'}

処理時間:5.0174829959869385 [sec]

DBへの接続確認

## 正常時
$ python mysql_IoTdummy.py --mode db          
MySqlサーバへの接続確認

Connection established

<mysql.connector.connection_cext.CMySQLConnection object at 0x7ff4589f91f0>

処理時間:0.3275420665740967 [sec]


## 異常時（DBサーバが停止していた場合）
$ python mysql_IoTdummy.py --mode db
MySqlサーバへの接続確認
9030 (28000): This MySQL server is currently stopped. Start the server to establish connectivity.

処理時間:0.22968602180480957 [sec]

テーブルの新規作成

$ python mysql_IoTdummy.py --mode db --db init
データベース・テーブルの新規作成

Connection established

	Finished dropping table (if existed).
	Finished creating table.

        Field            Type Null  Key Default           Extra
0          id      b'int(11)'   NO  PRI    None  auto_increment
1      iot_id      b'int(11)'   NO         None                
2        proc  b'varchar(10)'   NO         None                
3     section   b'varchar(2)'   NO         None                
4     iot_num  b'varchar(10)'   NO         None                
5   iot_state  b'varchar(10)'   NO         None                
6       val_1        b'float'   NO         None                
7       val_2        b'float'   NO         None                
8  created_at     b'datetime'  YES         None                
9  updated_at     b'datetime'  YES         None                
Done.

処理時間:1.008288860321045 [sec]

データの書き込み

## ８件のデータを0.2秒間隔で書き込む
$ python mysql_IoTdummy.py --mode db --db write --count 8 --wait 0.2
データベース・テーブルへのデータ書き込み

Connection established

100%|██████████████████████████████████████████████████████████████████████████████████████████| 8/8 [00:03<00:00,  2.46it/s]
Inserted 8 row(s) of data.
Done.

処理時間:3.45925235748291 [sec]

データの全件読み込み

$ python mysql_IoTdummy.py --mode db --db read                      
データベース・テーブルへのデータ読み込み

Connection established

Read: 8, row(s) of data.
Data row = (1, 0, '111', 'D', '825-0233', '北海道', 111.725, 76.5301, datetime.datetime(2022, 4, 2, 0, 32, 6), None)
Data row = (2, 1, '111', 'W', '143-6705', '新潟県', 196.401, 60.7279, datetime.datetime(2022, 4, 2, 0, 32, 6), None)
Data row = (3, 2, '111', 'V', '230-7952', '福島県', 121.454, 80.9434, datetime.datetime(2022, 4, 2, 0, 32, 6), None)
Data row = (4, 3, '111', 'F', '932-1487', '島根県', 102.066, 80.4288, datetime.datetime(2022, 4, 2, 0, 32, 6), None)
Data row = (5, 4, '111', 'I', '315-0935', '岐阜県', 168.707, 77.4312, datetime.datetime(2022, 4, 2, 0, 32, 6), None)
Data row = (6, 5, '111', 'T', '919-3894', '島根県', 185.376, 51.655, datetime.datetime(2022, 4, 2, 0, 32, 6), None)
Data row = (7, 6, '111', 'Z', '093-1699', '静岡県', 162.089, 66.7628, datetime.datetime(2022, 4, 2, 0, 32, 6), None)
Data row = (8, 7, '111', 'N', '224-5406', '静岡県', 139.609, 63.2027, datetime.datetime(2022, 4, 2, 0, 32, 6), None)
Done.

処理時間:0.33246612548828125 [sec]

データの全件削除

$ python mysql_IoTdummy.py --mode db --db delete
データベース・テーブル・データの全削除

Connection established

[]
Done.

処理時間:0.3926122188568115 [sec]

まとめ

これで、DBのデータ関連の各種テストがやりやすくなるかなぁ、、、、、