【Python】MySQLからCSVファイルへデータを出力する

■概要

MySQLに格納しているデータについて、全量取得する。
さらに、取得したデータをCSVファイルに出力する。

なお、接続するDB情報は、以下の通り。

■フローチャート

本ツールにおける処理の流れに関して、フローチャートを以下に示す。

▼メイン処理

メイン処理のフローチャートを以下に示す。

▼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を以下に示す。