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