MySQLの接続、クエリの実行方法【Python】

Pythonを用いて、MySQLへ接続・クエリを実行する方法を紹介します。

MySQLのインスト―ル、初期設定

MySQLの公式サイトからインストーラを入手して、インストールを行います。

インストールが終わったら、MySQLへ接続してデータベースを作成します。

CREATE DATABASE データベース名;

必要に応じてユーザを作成、権限を付与します。

CREATE USER 'ユーザ名'@'localhost' IDENTIFIED BY 'パスワード';
GRANT ALL PRIVILEGES ON データベース名.* TO 'ユーザ名'@'localhost';
FLUSH PRIVILEGES;

ライブラリのインストール

PythonでMySQLを扱うためのライブラリをインストールします。

pip install mysql-connector-python

実装方法

共通クラス、関数の作成

MySQL接続に関するクラスを作成して、処理の共通化を行います。

接続情報は自身の環境に合わせて修正、データ取得や更新等の処理を実装します。

 import mysql.connector
 from mysql.connector import Error
 
 class MySQLHandler:
     def __init__(self, host, user, password, database):
         self.host = host
         self.user = user
         self.password = password
         self.database = database
         self.connection = None
         self.cursor = None
 
     def connect(self):
         """MySQL に接続"""
         try:
             self.connection = mysql.connector.connect(
                 host=self.host,
                 user=self.user,
                 password=self.password,
                 database=self.database
             )
             if self.connection.is_connected():
                 self.cursor = self.connection.cursor()
                 print("MySQL connected")
         except Error as e:
             print(f"MySQL connection error: {e}")
 
     def close(self):
         """MySQL の接続を閉じる"""
         try:
             if self.connection and self.connection.is_connected():
                 if self.cursor:
                     self.cursor.close()
                 self.connection.close()
                 print("MySQL closed")
         except Error as e:
             print(f"MySQL close error: {e}")
 
     def execute_query(self, query, params=None):
         """クエリを実行"""
         try:
             self.connect()
             self.cursor.execute(query, params)
             self.connection.commit()
         except Error as e:
             print(f"MySQL query error: {e}")
         finally:
             self.close()
 
     def fetch_all(self, query, params=None):
         """データを取得"""
         try:
             self.connect()
             self.cursor.execute(query, params)
             return self.cursor.fetchall()
         except Error as e:
             print(f"MySQL fetch error: {e}")
         finally:
             self.close()
 
 # MySQLHandler のインスタンスを作成
 db_handler = MySQLHandler(
     host="localhost",
     user="ユーザ名",
     password="パスワード",
     database="データベース名"
 )

-----select,update,delete等の必要な関数を実装-----
 # データ取得
 def select_data(table):
     select_query = "SELECT * FROM " + table
     rows = db_handler.fetch_all(select_query)
     
     return rows

メイン処理から呼び出し

前述で作成した接続処理をインポートして、データを取得します。

配列で取得されたデータをコンソール出力して、正常に取得出来ているか確認します。

from connection import select_data

rows = select_data()
for row in rows:
     print(row)