[備忘録] MySQL Connector PythonでMySQLを操作してみた

はじめに

前回はGoogle ColabでPyMySQLを使ってMySQLを操作しましたが、今回はMySQLの公式Pythonライブラリであるmysql-connector-pythonを試してみます。

環境

• Google Colab
• MySQL（前回記事でセットアップ済み）
• mysql-connector-python

セットアップ

MySQLの起動（前回記事から）

# MySQLサービス開始
!service mysql start

MySQL Connector Pythonのインストール

!pip install mysql-connector-python

基本的な使い方

接続とデータベース作成

import mysql.connector
from mysql.connector import Error

# 接続設定
config = {
    'host': 'localhost',
    'user': 'root',
    'password': 'password',
    'charset': 'utf8mb4'
}

try:
    # 接続
    connection = mysql.connector.connect(**config)
    cursor = connection.cursor()
    
    # データベース作成
    cursor.execute("CREATE DATABASE IF NOT EXISTS sample_db")
    cursor.execute("USE sample_db")
    
    print("データベース接続成功")
    
except Error as e:
    print(f"エラー: {e}")

テーブル作成とCRUD操作

try:
    # テーブル作成
    create_table = """
    CREATE TABLE IF NOT EXISTS products (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(100) NOT NULL,
        price DECIMAL(10,2),
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    """
    cursor.execute(create_table)
    
    # データ挿入（複数）
    products = [
        ("ノートPC", 89800),
        ("マウス", 2980),
        ("キーボード", 5800)
    ]
    
    insert_query = "INSERT INTO products (name, price) VALUES (%s, %s)"
    cursor.executemany(insert_query, products)
    connection.commit()
    
    print(f"{cursor.rowcount}件のデータを挿入しました")
    
except Error as e:
    print(f"エラー: {e}")

データ取得

try:
    # 全データ取得
    cursor.execute("SELECT * FROM products")
    results = cursor.fetchall()
    
    print("登録済み商品:")
    for row in results:
        print(f"ID: {row[0]}, 商品名: {row[1]}, 価格: ¥{row[2]:,}")
    
    # 条件付き検索
    cursor.execute("SELECT * FROM products WHERE price > %s", (5000,))
    expensive_products = cursor.fetchall()
    
    print("\n高額商品:")
    for product in expensive_products:
        print(f"{product[1]}: ¥{product[2]:,}")
        
except Error as e:
    print(f"エラー: {e}")

実践的な機能

トランザクション処理

try:
    # トランザクション開始
    connection.start_transaction()
    
    # 複数の操作をまとめて実行
    cursor.execute("UPDATE products SET price = price * 0.9 WHERE id = %s", (1,))
    cursor.execute("INSERT INTO products (name, price) VALUES (%s, %s)", 
                  ("セール商品", 1980))
    
    # 全てコミット
    connection.commit()
    print("トランザクション完了")
    
except Error as e:
    # エラー時はロールバック
    connection.rollback()
    print(f"トランザクションエラー: {e}")

接続プール使用

from mysql.connector import pooling

# 接続プール設定
pool_config = {
    'pool_name': 'mypool',
    'pool_size': 5,
    'pool_reset_session': True,
    'host': 'localhost',
    'user': 'root',
    'password': 'password',
    'database': 'sample_db'
}

try:
    # プール作成
    pool = pooling.MySQLConnectionPool(**pool_config)
    
    # プールから接続取得
    pool_connection = pool.get_connection()
    pool_cursor = pool_connection.cursor()
    
    pool_cursor.execute("SELECT COUNT(*) FROM products")
    count = pool_cursor.fetchone()[0]
    print(f"商品数: {count}件")
    
    # 接続をプールに戻す
    pool_connection.close()
    
except Error as e:
    print(f"プールエラー: {e}")

プリペアードステートメント

try:
    # プリペアードステートメント
    prep_stmt = "SELECT * FROM products WHERE price BETWEEN %s AND %s"
    
    # パラメータを安全に渡す
    cursor.execute(prep_stmt, (3000, 10000))
    results = cursor.fetchall()
    
    print("価格帯別商品:")
    for product in results:
        print(f"{product[1]}: ¥{product[2]:,}")
        
except Error as e:
    print(f"エラー: {e}")

finally:
    # 接続クローズ
    if connection.is_connected():
        cursor.close()
        connection.close()
        print("MySQL接続を閉じました")

便利な機能

辞書形式での結果取得

try:
    connection = mysql.connector.connect(**config, database='sample_db')
    
    # 辞書カーソル使用
    dict_cursor = connection.cursor(dictionary=True)
    dict_cursor.execute("SELECT * FROM products LIMIT 2")
    
    results = dict_cursor.fetchall()
    for row in results:
        print(f"商品: {row['name']}, 価格: ¥{row['price']:,}")
        
except Error as e:
    print(f"エラー: {e}")

バッチ処理

# 大量データの効率的な挿入
batch_data = [(f"商品{i}", i * 1000) for i in range(1, 101)]

try:
    cursor.executemany(
        "INSERT INTO products (name, price) VALUES (%s, %s)", 
        batch_data
    )
    connection.commit()
    print(f"{len(batch_data)}件のデータを一括挿入")
    
except Error as e:
    print(f"バッチ処理エラー: {e}")

エラーハンドリング

def safe_database_operation():
    connection = None
    try:
        connection = mysql.connector.connect(**config, database='sample_db')
        cursor = connection.cursor()
        
        # 何らかの処理
        cursor.execute("SELECT * FROM products")
        return cursor.fetchall()
        
    except mysql.connector.Error as db_error:
        print(f"データベースエラー: {db_error}")
        return None
        
    except Exception as e:
        print(f"予期しないエラー: {e}")
        return None
        
    finally:
        if connection and connection.is_connected():
            cursor.close()
            connection.close()

# 使用例
results = safe_database_operation()
if results:
    print(f"{len(results)}件のデータを取得")

まとめ

mysql-connector-pythonは、公式ライブラリならではの安定性と機能の豊富さが特徴です。接続プールやトランザクション制御、柔軟な設定オプション、充実したエラーハンドリングなど、実践的な開発に必要な機能が揃っています。

興味ありましたらGoogle Colabで試してみてください。

参考情報