■概要
MySQLに格納しているデータについて、全量取得する。
さらに、取得したデータをCSVファイルに出力する。
なお、接続するDB情報は、以下の通り。
| 項目名 | 値 |
| 接続ユーザ名 | user001 |
| 接続パスワード | aaa |
| ホスト | 127.0.0.1 |
| ポート | 3306 |
| スキーマ | test_schema |
| テーブル | test_output_csv_table |
■フローチャート
本ツールにおける処理の流れに関して、フローチャートを以下に示す。
▼メイン処理
メイン処理のフローチャートを以下に示す。
▼mysqlからデータをcsvに出力する関数
mysqlからデータをcsvに出力する関数のフローチャートを以下に示す。
▼Keyを取得する関数
Keyを取得する関数のフローチャートを以下に示す。
▼Mysql接続・実行クラス
Mysql接続・実行クラスのフローチャートを以下に示す。
▼テーブルからデータを取得するクラス
テーブルからデータを取得するクラスのフローチャートを以下に示す。
■サンプルコード
本ツールで実装したサンプルのソースコードを以下に示す。
・output_csv_from_mysql.py
import csv
from datetime import datetime
from connect_mysql import ConnectMysql, SelectData
# 出力ディレクトリパス
OUTPUT_DIR = './output_dir/'
# Keyを取得する関数
def get_dict_keys(list_data):
# 返却用リストの初期化
dict_keys = []
# リストデータ分繰り返し
for data in list_data:
# 辞書データ分繰り返し
for key in data.keys():
# 返却用リストにKeyが存在していない場合
if key not in dict_keys:
# 返却用リストにKeyを追加
dict_keys.append(key)
# 返却用リストを返す
return dict_keys
# mysqlからデータをcsvに出力する関数
def output_csv_from_mysql():
# Mysql接続・実行クラスのインスタンス
sql_connect = ConnectMysql()
# Mysqlに接続する関数実行
connect_info = sql_connect.connector_mysql()
# テーブルからデータを取得するクラスのインスタンス
select_user = SelectData()
# テーブルからデータを取得する関数実行
result_datas = select_user.select_data(connect_info)
# Keyを取得する関数実行
filed_names = get_dict_keys(result_datas)
# 現在時刻取得
now_date = datetime.now()
# 出力するファイル名をセット
output_file_path = OUTPUT_DIR + now_date.strftime('%Y%m%d_%H%M%S') + '.csv'
# 出力ファイルを追記モードでオープン
with open(output_file_path, 'a') as output_file:
# 辞書→csvライブラリの呼び出し
writer = csv.DictWriter(output_file, fieldnames=filed_names)
# ヘッダーを出力
writer.writeheader()
# データ部を出力
writer.writerows(result_datas)
# メイン処理
if __name__ == '__main__':
# mysqlからデータをcsvに出力する関数実行
output_csv_from_mysql()
・connect_mysql.py
import mysql.connector
# MySQLの接続情報
MYSQL_USER = 'user001'
MYSQL_PASSWORD = 'aaa'
MYSQL_HOST = '127.0.0.1'
MYSQL_PORT = '3306'
MYSQL_DATABASE = 'test_schema'
MYSQL_AUTH = 'mysql_native_password'
# SQL文
SQL_SELECT_QUERY = 'SELECT * FROM test_output_csv_table;'
# Mysql接続・実行クラス
class ConnectMysql:
def __init__(self):
self.con = None
# Mysqlに接続する関数
def connector_mysql(self):
try:
# mysqlに接続
self.con = mysql.connector.connect(
user=MYSQL_USER,
password=MYSQL_PASSWORD,
host=MYSQL_HOST,
port=MYSQL_PORT,
database=MYSQL_DATABASE,
auth_plugin=MYSQL_AUTH
)
# 例外処理
except Exception as e:
print('connector_mysql: 例外エラー: ', e)
finally:
return self.con
# テーブルからデータを取得するクラス
class SelectData():
def __init__(self):
self.rows = None
# テーブルからデータを取得する関数
def select_data(self, con):
# 出力形式は辞書型
self.cursor = con.cursor(dictionary=True)
try:
# SQL実行
self.cursor.execute(SQL_SELECT_QUERY)
# 検索結果格納
self.rows = self.cursor.fetchall()
# 例外処理
except Exception as e:
print('select_data: 例外エラー: ', e)
finally:
self.cursor.close()
return self.rows
■実行結果
本ツールを実行したCSVファイルの出力結果を以下に示す。
ID,data1,data2,data3
1,a,aa,aaa
2,b,bb,bbb
3,c,cc,ccc
■参考
本ツールを作成するにあたり、参考とさせていただいたURLを以下に示す。
