PythonでMySQLを操作する方法を解説します。

使用ライブラリ(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()

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

sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
cur.execute(sql, ('webmaster@python.org', 'very-secret'))
 
# コミットしてトランザクション実行
connection.commit()

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

MySQL接続部分を共通化

♯ MySQL接続（DBと接続）

mysql_connector.py

# coding:utf-8
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 (mydb.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 (mydb.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 (mydb.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())

sql文を実行

test.py

from mysql_connector import MysqlConnector

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

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

# insert
connector.insert("INSERT INTO target_users(`id`, `target_name`, `uuid`, `target_email`) VALUES('999', 'サッカー', '9f0da26e-e91b-11eb-8baf-00155dfdbfdf', 'email@example.org');")

# update
connector.update("UPDATE target_users SET target_name = 'ラグビー' WHERE id = 1;")

# delete
connector.delete("DELETE FROM target_users WHERE id = 29;")

#  テーブル作成
table = 'test_table'
connector.fetch("DROP TABLE IF EXISTS `%s`;" %table)
connector.fetch(
    """
    CREATE TABLE IF NOT EXISTS `%s` (
   id int(11) NOT NULL AUTO_INCREMENT,
   env_name varchar(255) NOT NULL,
   comment text DEFAULT NULL,
   created_at datetime NOT NULL DEFAULT current_timestamp(),
   update_at datetime DEFAULT NULL ON UPDATE current_timestamp(),
   delete_at datetime DEFAULT NULL,
   PRIMARY KEY (id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8
   """ %table)


# 接続確認
connector.debug_conn()

# 切断
connector.close()
connector = MysqlConnector() 

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

# insert
connector.insert("INSERT INTO target_users(`id`, `target_name`, `uuid`, `target_email`) VALUES('999', 'サッカー', '9f0da26e-e91b-11eb-8baf-00155dfdbfdf', 'email@example.org');")

# update
connector.update("UPDATE target_users SET target_name = 'ラグビー' WHERE id = 1;")

# delete
connector.delete("DELETE FROM target_users WHERE id = 29;")

#  テーブル作成
table = 'test_table'
connector.fetch("DROP TABLE IF EXISTS `%s`;" %table)
connector.fetch(
    """
    CREATE TABLE IF NOT EXISTS `%s` (
   id int(11) NOT NULL AUTO_INCREMENT,
   env_name varchar(255) NOT NULL,
   comment text DEFAULT NULL,
   created_at datetime NOT NULL DEFAULT current_timestamp(),
   update_at datetime DEFAULT NULL ON UPDATE current_timestamp(),
   delete_at datetime DEFAULT NULL,
   PRIMARY KEY (id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8
   """ %table)


# 接続確認
connector.debug_conn()

# 切断
connector.close()

以下コマンドを実行して結果確認

python test.py