6
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

mysql-connector-python でMySQL操作(python Mysql接続)

Posted at

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
6
4
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
6
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?