mysql-connector-python

筆者の環境

• python 3.9

mysql-connector-pythonとは

Pythonプログラムからデータベースにアクセスするためのライブラリの一つ

mysql-connectorのインストール

$ pip3 install mysql-connector-python

基本的な使用方法

import mysql.connector as mydb

# コネクションの作成
import conn = mydb.connect(
    host='localhost',
    port='3306',
    user='root',
    password='password',
    database='dbname'
)

# コネクションが切れた時に再接続してくれるよう設定
conn.ping(reconnect=True)

# DB操作用にカーソルを作成
cur = conn.cursor()

テーブル作成

table = 'test_table'
cur.execute("DROP TABLE IF EXISTS `%s`;", table)
cur.execute(
    """
    CREATE TABLE IF NOT EXISTS `%s` (
    `id` int auto_increment primary key,
    `name` varchar(50) not null,
    `price` int(11) not null
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci
    """, table)
)

データ挿入

cur.execute("INSERT INTO test_table VALUES (1, 'BTC', 10200)")

# プレースホルダを利用して挿入
cur.execute("INSERT INTO test_table VALUES (2, 'ETH', %s)", (5000, ))
cur.execute("INSERT INTO test_table VALUES (%s, %s, %s)", (3 ,'XEM', 2500))
cur.execute("INSERT INTO test_table VALUES (%(id)s, %(name)s, %(price)s)", {'id': 4, 'name': 'XRP', 'price': 1000})

# executemanyで複数データを一度に挿入
records = [
  (5, 'MONA', 3000),
  (6, 'XP', 1000),
]
cur.executemany("INSERT INTO test_table VALUES (%s, %s, %s)", records)

データ取得

cur.execute("SELECT * FROM test_table ORDER BY id ASC")

# 全てのデータを取得
rows = cur.fetchall()

for row in rows:
    print(row)

# 出力結果
"""
(1, 'BTC', 10200)
(2, 'ETH', 5000)
(3 ,'XEM', 2500)
(4, 'XRP', 1000)
(5, 'MONA', 3000)
(6, 'XP', 1000)
"""

# 1件取得
cur.execute("SELECT * FROM test_table WHERE name=%s", ('BTC', ))
print(cur.rowcount)
print(cur.fetchone())

# 出力結果
"""
1
(1, 'BTC', 10200)
"""

データ更新

# UPDATE
cur.execute('UPDATE test_table SET name=%s WHERE id=1', ('ビットコイン',))
cur.execute('UPDATE test_table SET name=%s WHERE id=%s', ('イーサリアム', 2))

データ削除

# DELETE
cur.execute('DELETE FROM test_table WHERE id = 3')

最後

# カーソルとコネクションを閉じる
cur.close()
conn.close()

DB接続・操作専用のクラスを作成(ここから本趣旨)

共通化をすべき理由:

- MySQLとの接続が必要な時に、毎度、下記のコードを書いて、接続情報を生成するのは非常に手間

- DB設定の修正を行う際に、関連するコードを全て修正しなければならないので、かなり苦痛、修正漏れとか出しそう

- そもそもエンジニアとして、同じことを2回するのがいや

# コネクションの作成
conn = mydb.connect(
    host='localhost',
    port='3306',
    user='root',
    password='password',
    database='dbname'
)
...

ということで、いざ、MySQLの操作用のクラスを作成

コード(DB機能提供側)

(mysql_connector.py)

import mysql.connector as mydb
import sys

#
#    DBアクセス管理クラス
#
class MysqlConnector:

    # -----------------------------------
    # コンストラクタ
    #
    # コネクションを取得し、クラス変数にカーソルを保持する。
    # -----------------------------------
    def __init__(self, host='localhost', port='3306', user='root', password='password', database='dbname'):

        try:
            self.conn = mydb.connect(
                host=host,
                port=port,
                user=user,
                password=password,
                database=database
            )
            # コネクションの設定
            self.conn.ping(reconnect=True)
            self.conn.autocommit = False

            # カーソル情報をクラス変数に格納
            self.cur = self.conn.cursor()

            # 接続できているかどうか確認
            # print(self.conn.is_connected())

        except (mydb.errors.ProgrammingError) as e:
            print(e)
            sys.exit(1)

    # -----------------------------------
    # クエリの実行
    #
    # クエリを実行し、取得結果を呼び出し元に通知する。
    # -----------------------------------
    def fetch(self, sql):
        try:
            self.cur.execute(sql)
            rows = self.cur.fetchall()
            return rows
        except mydb.Error as e:
            print(e)
            sys.exit(1)

    # -----------------------------------
    # インサートの実行
    #
    # インサートを実行する。
    # -----------------------------------
    def insert(self, sql):
        try:
            self.cur.execute(sql)
            self.conn.commit()

        except (mysql.connector.errors.ProgrammingError) as e:
            self.conn.rollback()
            print(e)
            sys.exit(1)


    # -----------------------------------
    # アップデートの実行
    #
    # アップデートを実行する。
    # -----------------------------------
    def update(self, sql):
        try:
            self.cur.execute(sql)
            self.conn.commit()
        except (mysql.connector.errors.ProgrammingError) as e:
            self.conn.rollback()
            print(e)
            sys.exit(1)

    def delete(self, sql):
        try:
            self.cur.execute(sql)
            self.conn.commit()
        except (mysql.connector.errors.ProgrammingError) as e:
            self.conn.rollback()
            print(e)
            sys.exit(1)

    def close(self):
        self.cur.close()
        self.conn.close()

    def debug_conn(self):
        print(self.conn.is_connected())

コード(機能使用側)

接続

from yourmodulename.mysql_connector import MysqlConnector


connector = MysqlConnector() # インスタンスを生成

ホスト名やユーザー名を、デフォルトから変更するなら、インスタンス作成時に引数に設定する

connector = MysqlConnector(host='myhost', port='myport', user='myuser', password='mypassword', database='mydb')

SELECT

rows = connector.fetch("SELECT * FROM universities limit 5;")
# 確認
for row in rows:
    print(row)

insert

connector.insert("INSERT INTO circles(`id`, `name`, `university_id`) VALUES('1', 'サッカー', '1');")

update

connector.update("UPDATE circles SET name = 'ラグビー' WHERE id = 1;")

delete

connector.delete("DELETE FROM circles WHERE id = 1;")

接続確認

connector.debug_conn()

最後

connector.close()

最後に

コード(DB機能提供側) は、再利用して楽する為に、丸コピして、使ってみてください。

また、python-mysql-connectorは導入が簡単な反面、再利用の設計を考えると、工数かけて設計しないと行けなくなるので、簡易的なプロジェクト以外の場合は、pythonで使えるORMの使用を検討するべきだということがわかりました。