PythonでMySQLを操作する方法：PyMySQL入門（初心者向け）

データベースはほとんどのアプリケーションで重要な役割を果たします。

特にWebアプリケーションにおいては、ユーザー情報や投稿データなどを適切に管理するためにMySQLのようなリレーショナルデータベースを利用することが一般的です。

本記事では、PythonからMySQLを操作するためのライブラリであるPyMySQLを取り上げ、インストールから基本的な接続、データのCRUD（作成、読み取り、更新、削除）操作まで、初級者向けにわかりやすく解説します。

各章でサンプルコードを示しながら、実際に手を動かして学べる構成にしています。最後に演習問題を用意しているので、ぜひ挑戦して理解を深めてみてください。

PyMySQLの概要とインストール

PyMySQLとは？

PyMySQLは、純粋なPythonで書かれたMySQLクライアントライブラリです。

C言語で実装されたドライバ（例：MySQLdb）とは異なり、追加のバイナリ（ホスト環境でのコンパイルなど）を必要とせず、Pythonのパッケージとしてインストールするだけで利用できます。

また、Pythonの標準的なDB-API 2.0規格に準拠しているため、ほかのデータベースライブラリと類似のインターフェースで扱えるのも特徴です。

インストール方法

PyMySQLはpipで簡単にインストール可能です。

ターミナル（コマンドプロンプト）を開き、以下のコマンドを実行してください。

pip install PyMySQL

もし仮想環境（venvやvirtualenv）を利用している場合は、仮想環境をアクティベートしてからインストールすることで、他のプロジェクトに影響を与えずに管理できます。

# Unix/Mac の場合
python3 -m venv venv
source venv/bin/activate
pip install PyMySQL

# Windows の場合
python -m venv venv
venv\Scripts\activate
pip install PyMySQL

インストールが成功すると、Pythonスクリプト内で import pymysql が可能になります。

MySQLサーバーへの接続

事前準備

MySQL上で以下のようにして、testdb というデータベースと testuser ユーザーを用意しておくと良いでしょう。

CREATE DATABASE testdb CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;
CREATE USER 'testuser'@'localhost' IDENTIFIED BY 'password123';
GRANT ALL PRIVILEGES ON testdb.* TO 'testuser'@'localhost';
FLUSH PRIVILEGES;

接続の基本コード

PythonからMySQLに接続する最も基本的なコード例は以下の通りです。

import pymysql

# 接続情報
host = 'localhost'
port = 3306
user = 'testuser'
password = 'password123'
database = 'testdb'
charset = 'utf8mb4'

# 接続オブジェクトの取得
connection = pymysql.connect(
    host=host,
    port=port,
    user=user,
    password=password,
    db=database,
    charset=charset,
    cursorclass=pymysql.cursors.DictCursor  # 結果を辞書形式で取得したい場合
)

print("MySQLに接続しました。")

# ※ 必ず finally ブロック等で close() しましょう
connection.close()
print("接続を閉じました。")

host, port, user, password, db（または database）には、自身のMySQLサーバー環境に合わせた値を指定してください。

charset に utf8mb4 を指定することで、絵文字を含む文字列も扱うことができます。

cursorclass=pymysql.cursors.DictCursor をオプションで与えると、SQLの実行結果を辞書（dict）で受け取れるため、コードがわかりやすくなります（デフォルトはタプル形式）。

最後に必ず connection.close() を実行して接続を閉じることが大切です。そうしないと接続が残り続け、サーバー側のリソースを消費してしまいます。

データベースおよびテーブルの作成

データベースへの接続とカーソル取得

接続が確立したら、まずは「カーソル」を取得します。カーソルを通じてSQL文を実行できます。次のようにしてカーソルを作成します。

import pymysql

connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

try:
    with connection.cursor() as cursor:
        # ここにSQL実行を記述する
        pass
finally:
    connection.close()

with connection.cursor() as cursor: と書くと、ブロックを抜ける際にカーソルが自動的にクローズされます。

テーブルの作成

例として、users テーブルを作成するSQLを実行してみます。users テーブルは次のような構造を想定します。

• id (INT, AUTO_INCREMENT, PRIMARY KEY)
• name (VARCHAR(100), NOT NULL)
• email (VARCHAR(255), UNIQUE, NOT NULL)
• created_at (DATETIME, NOT NULL, デフォルトは現在時刻)

import pymysql
from datetime import datetime

connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

create_table_sql = """
CREATE TABLE IF NOT EXISTS users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(255) NOT NULL UNIQUE,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""

try:
    with connection.cursor() as cursor:
        cursor.execute(create_table_sql)
    connection.commit()
    print("テーブル 'users' を作成または既存のまま維持しました。")
finally:
    connection.close()

IF NOT EXISTS を付けることで、既に users テーブルが存在していればスキップし、新たに作成されることを防ぎます。

テーブル作成後に connection.commit() を呼び出してコミットを行うことで、DDL（Data Definition Language）の操作が確定します。

デフォルトエンジンとして InnoDB を指定し、文字コードは utf8mb4 に設定しています。

レコードの挿入（INSERT）

単一レコード挿入

users テーブルに新しいユーザーを追加する例を示します。

プレースホルダーを利用してSQLインジェクションを防ぎながら値を渡す方法を覚えましょう。

import pymysql

connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

insert_sql = "INSERT INTO users (name, email) VALUES (%s, %s);"

try:
    with connection.cursor() as cursor:
        name = "山田 太郎"
        email = "taro.yamada@example.com"
        cursor.execute(insert_sql, (name, email))
    connection.commit()
    print(f"ユーザー '{name}' を挿入しました。挿入されたID: {cursor.lastrowid}")
finally:
    connection.close()

SQL文中の %s がプレースホルダーです。cursor.execute() の第2引数にタプルを渡すことで安全に値をバインドできます。

挿入後、cursor.lastrowid プロパティを参照すると、AUTO_INCREMENTで生成されたIDを取得可能です。

最後に必ず connection.commit() を呼ぶことで、挿入操作が確定してデータベースに反映されます。

複数レコード挿入

複数のユーザーデータをまとめて挿入したい場合は、executemany() を使うと効率的です。

import pymysql

connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

users_data = [
    ("佐藤 花子", "hanako.sato@example.com"),
    ("鈴木 次郎", "jiro.suzuki@example.com"),
    ("高橋 三郎", "saburo.takahashi@example.com")
]

insert_sql = "INSERT INTO users (name, email) VALUES (%s, %s);"

try:
    with connection.cursor() as cursor:
        cursor.executemany(insert_sql, users_data)
    connection.commit()
    print(f"{cursor.rowcount} 件のユーザーを一括挿入しました。")
finally:
    connection.close()

cursor.executemany() にリストやタプルのリストを渡すと、同じSQL文で複数件のレコードをまとめて挿入できます。

cursor.rowcount を参照すると、影響を受けた行数（今回であれば挿入された行数）を取得できます。

バルクインサートは大量データを効率的に処理したい場合にも有効です。

レコードの取得（SELECT）

単一レコードの取得

ユーザーIDを指定して1件だけ取得する例です。

import pymysql

connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

select_sql = "SELECT id, name, email, created_at FROM users WHERE id = %s;"

try:
    with connection.cursor() as cursor:
        user_id = 1
        cursor.execute(select_sql, (user_id,))
        user = cursor.fetchone()  # 1件取得
        if user:
            print("ユーザー情報:", user)
        else:
            print(f"ID {user_id} のユーザーは存在しません。")
finally:
    connection.close()

cursor.fetchone() はクエリの結果から最初の1行を取得します。結果がないときは None を返します。

cursor.fetchmany(size=数値) を使うと指定行数分をタプルのリストで取得できます。

cursor.fetchall() は結果セット全体をタプルのリスト（あるいは設定によっては辞書のリスト）で取得します。ただし大量データを一度に取得するとメモリを圧迫するため、必要に応じて fetchmany() を使うか、ループで fetchone() を利用するとよいでしょう。

複数レコードの取得

全ユーザー情報を取得し、ループで表示する例です。

import pymysql

connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

select_all_sql = "SELECT id, name, email, created_at FROM users;"

try:
    with connection.cursor() as cursor:
        cursor.execute(select_all_sql)
        users = cursor.fetchall()  # 全件取得
        for user in users:
            print(f"ID: {user['id']}, 名前: {user['name']}, メール: {user['email']}, 登録日時: {user['created_at']}")
finally:
    connection.close()

fetchall() で取得した結果をそのままループすることで、全ユーザーを順番に処理できます。

WHERE句やORDER BY句を組み合わせて、フィルタリングやソートを行った結果を取得することももちろん可能です。

レコードの更新（UPDATE）

既存のユーザー情報を更新する例を示します。ここではメールアドレスを変更するコードです。

import pymysql

connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

update_sql = "UPDATE users SET email = %s WHERE id = %s;"

try:
    with connection.cursor() as cursor:
        new_email = "new_email@example.com"
        user_id = 2
        cursor.execute(update_sql, (new_email, user_id))
    connection.commit()
    print(f"ID {user_id} のメールアドレスを '{new_email}' に更新しました。変更件数: {cursor.rowcount}")
finally:
    connection.close()

SET email = %s WHERE id = %s のように、プレースホルダー %s を使って安全に値を渡しています。

更新後に cursor.rowcount を確認すると、実際に更新された行数がわかります。もし該当する行がなければ rowcount は 0 になります。

更新処理でも必ず connection.commit() を呼んで変更を反映させる必要があります。

レコードの削除（DELETE）

ユーザーを削除する例です。ここでは指定したIDのユーザーを削除します。

import pymysql

connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

delete_sql = "DELETE FROM users WHERE id = %s;"

try:
    with connection.cursor() as cursor:
        user_id = 3
        cursor.execute(delete_sql, (user_id,))
    connection.commit()
    print(f"ID {user_id} のユーザーを削除しました。削除件数: {cursor.rowcount}")
finally:
    connection.close()

DELETE FROM users WHERE id = %s のようにして、特定の行を削除しています。

cursor.rowcount によって実際に削除された行数を取得可能です。

万が一 DELETE 文を間違えてWHEREを指定しないと、テーブル内のすべての行が消えてしまうので注意してください。

接続の終了とエラー処理

try–except–finallyによる例外処理

実運用では、SQL実行時にタイムアウトや権限エラー、データ型不一致などの例外が発生することがあります。

そのため、try–except–finally 構文を利用してエラーハンドリングを行い、例外発生時にはロールバックを行った上で接続を確実に閉じるようにしましょう。

import pymysql

connection = None

try:
    connection = pymysql.connect(
        host='localhost',
        port=3306,
        user='testuser',
        password='password123',
        db='testdb',
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor
    )
    with connection.cursor() as cursor:
        # 例: 存在しないテーブルにアクセス
        cursor.execute("SELECT * FROM non_existent_table;")
        data = cursor.fetchall()
        print(data)
    connection.commit()

except pymysql.MySQLError as e:
    # MySQLに関するエラーをキャッチ
    print("MySQLエラーが発生しました:", e)
    if connection:
        connection.rollback()  # エラー発生時はロールバック
finally:
    if connection:
        connection.close()
        print("接続を閉じました。")

pymysql.MySQLError を捕捉することで、MySQL関連のエラーを一括して扱えます。

エラーが発生した場合には connection.rollback() を呼び出し、トランザクションを取り消します。

その後、finally ブロックで connection.close() を実行し、必ず接続を解放します。

演習問題と解答例

最後に、本記事で学んだ内容を確認するための演習問題をいくつか用意します。

各問題に対し、実際にPythonスクリプトとしてファイルを作成し、動作を確認してみてください。

演習問題1：新規テーブルの作成

1. データベース testdb に、次のカラムを持つテーブル products を作成しなさい。
   • product_id (INT, AUTO_INCREMENT, PRIMARY KEY)
   • product_name (VARCHAR(150), NOT NULL)
   • price (DECIMAL(10,2), NOT NULL)
   • stock (INT, NOT NULL, デフォルトは 0)
   • created_at (TIMESTAMP, デフォルトは CURRENT_TIMESTAMP)
2. テーブルが正常に作成されたか確認するSELECT文を実行し、Python上で出力しなさい。

import pymysql

# 接続情報（適宜変更）
connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

create_table_sql = """
CREATE TABLE IF NOT EXISTS products (
    product_id INT AUTO_INCREMENT PRIMARY KEY,
    product_name VARCHAR(150) NOT NULL,
    price DECIMAL(10,2) NOT NULL,
    stock INT NOT NULL DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""

select_table_sql = "SHOW TABLES LIKE 'products';"

try:
    with connection.cursor() as cursor:
        # テーブル作成
        cursor.execute(create_table_sql)
        connection.commit()
        print("テーブル 'products' を作成または既存のまま維持しました。")

        # テーブル存在確認
        cursor.execute(select_table_sql)
        result = cursor.fetchone()
        if result:
            print("テーブル 'products' が存在します。")
        else:
            print("テーブル 'products' は存在しません。")
finally:
    connection.close()

演習問題2：データの挿入と取得

1. products テーブルに次のデータを追加しなさい。
   • 商品A：価格 1500.00、在庫 30
   • 商品B：価格 2500.50、在庫 15
   • 商品C：価格 500.00、在庫 100
2. products テーブルから全てのレコードを取得し、コンソールに次の形式で出力しなさい。 [product_id] product_name (price: ¥価格, stock: 在庫個)
   例: [1] 商品A (price: ¥1500.00, stock: 30個)

import pymysql

# 接続情報（適宜変更）
connection = pymysql.connect(
    host='localhost',
    port=3306,
    user='testuser',
    password='password123',
    db='testdb',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

# 挿入用SQL
insert_sql = "INSERT INTO products (product_name, price, stock) VALUES (%s, %s, %s);"

# 追加するデータ
products_to_insert = [
    ("商品A", 1500.00, 30),
    ("商品B", 2500.50, 15),
    ("商品C", 500.00, 100)
]

select_all_sql = "SELECT product_id, product_name, price, stock FROM products;"

try:
    with connection.cursor() as cursor:
        # 複数レコード一括挿入
        cursor.executemany(insert_sql, products_to_insert)
    connection.commit()
    print(f"{len(products_to_insert)} 件のレコードを 'products' テーブルに挿入しました。")

    with connection.cursor() as cursor:
        cursor.execute(select_all_sql)
        products = cursor.fetchall()
        for prod in products:
            print(f"[{prod['product_id']}] {prod['product_name']} (price: ¥{prod['price']:.2f}, stock: {prod['stock']}個)")
finally:
    connection.close()

演習問題3：条件付きクエリと更新

1. products テーブルのうち、価格が 2000 円以上の商品を取得し、取得結果をリストとして返す関数 get_expensive_products(min_price) を定義しなさい。
2. 関数を呼び出し、min_price=2000 を渡して取得した商品の情報をコンソールに表示しなさい。
3. products テーブルのうち、在庫が 20 個以下の商品の在庫を一律で 50 個に更新しなさい。更新後、更新件数をコンソールに表示しなさい。

import pymysql

# 接続情報（適宜変更）
DB_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'testuser',
    'password': 'password123',
    'db': 'testdb',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

def get_expensive_products(min_price):
    """
    指定した価格以上の商品を取得する関数
    :param min_price: 取得する商品の最低価格（float）
    :return: 条件を満たす商品情報の辞書リスト
    """
    connection = None
    try:
        connection = pymysql.connect(**DB_CONFIG)
        with connection.cursor() as cursor:
            sql = "SELECT product_id, product_name, price, stock FROM products WHERE price >= %s;"
            cursor.execute(sql, (min_price,))
            result = cursor.fetchall()
            return result
    except pymysql.MySQLError as e:
        print("エラーが発生しました:", e)
        return []
    finally:
        if connection:
            connection.close()

def update_low_stock(threshold, new_stock):
    """
    在庫が threshold 以下の商品の在庫を new_stock に更新する関数
    :param threshold: 在庫の閾値（int）
    :param new_stock: 更新後の在庫数（int）
    :return: 更新件数（int）
    """
    connection = None
    try:
        connection = pymysql.connect(**DB_CONFIG)
        with connection.cursor() as cursor:
            sql = "UPDATE products SET stock = %s WHERE stock <= %s;"
            cursor.execute(sql, (new_stock, threshold))
            updated_count = cursor.rowcount
        connection.commit()
        return updated_count
    except pymysql.MySQLError as e:
        print("エラーが発生しました:", e)
        if connection:
            connection.rollback()
        return 0
    finally:
        if connection:
            connection.close()

if __name__ == "__main__":
    # 演習3-1: 価格が2000円以上の商品を取得
    expensive_items = get_expensive_products(2000)
    if expensive_items:
        print("価格が¥2000以上の商品一覧：")
        for item in expensive_items:
            print(f"- [{item['product_id']}] {item['product_name']} (price: ¥{item['price']:.2f}, stock: {item['stock']}個)")
    else:
        print("条件を満たす商品はありませんでした。")

    # 演習3-2: 在庫が20以下の商品の在庫を50に更新
    changed_count = update_low_stock(threshold=20, new_stock=50)
    print(f"在庫が20以下の商品を {changed_count} 件、在庫50に更新しました。")

演習問題4：レコードの削除とトランザクション

1. products テーブルから、価格が 1000 円未満の商品をすべて削除しなさい。
2. 削除処理はトランザクションとして扱い、削除件数をコンソールに表示し、最後にコミットするかロールバックするかを判断するコードを実装しなさい。
   • 削除件数が 2 件以上の場合：コミットし、”削除を確定しました。削除件数: X 件” と表示。
   • 削除件数が 0 件の場合：ロールバックし、”削除対象がありませんでした。処理を取り消しました。” と表示。
   • それ以外（1 件のみ削除された場合など）：コミットし、”削除件数 X 件。正常に削除されました。” と表示。

import pymysql

# 接続情報（適宜変更）
DB_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'testuser',
    'password': 'password123',
    'db': 'testdb',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

def delete_low_price_items(price_threshold):
    """
    price_threshold 未満の商品を削除し、処理をコミット/ロールバックする関数
    :param price_threshold: 削除判定となる価格の閾値（float）
    """
    connection = None
    try:
        connection = pymysql.connect(**DB_CONFIG)
        with connection.cursor() as cursor:
            sql = "DELETE FROM products WHERE price < %s;"
            cursor.execute(sql, (price_threshold,))
            deleted_count = cursor.rowcount

        # 条件に応じてコミットまたはロールバック
        if deleted_count >= 2:
            connection.commit()
            print(f"削除を確定しました。削除件数: {deleted_count} 件")
        elif deleted_count == 0:
            connection.rollback()
            print("削除対象がありませんでした。処理を取り消しました。")
        else:
            connection.commit()
            print(f"削除件数 {deleted_count} 件。正常に削除されました。")

    except pymysql.MySQLError as e:
        print("エラーが発生しました:", e)
        if connection:
            connection.rollback()
            print("エラー発生時のためロールバックしました。")
    finally:
        if connection:
            connection.close()

if __name__ == "__main__":
    # 価格1000円未満の商品を削除
    delete_low_price_items(1000.00)

まとめ

本記事では、PythonからMySQLを操作するためのPyMySQLライブラリの使い方を初級者向けに解説しました。主なポイントは次のとおりです。

1. PyMySQLのインストール
   • pip install PyMySQL を実行するだけで導入可能。
   • 標準的なDB-API 2.0に準拠しており、Pythonの辞書形式で結果を取得できる。
2. 接続とカーソルの取得
   • pymysql.connect() で接続し、with connection.cursor() でカーソルを取得。
   • charset=’utf8mb4′ を指定すると、絵文字を含む文字列も扱える。
3. データベース操作（DDL）
   • テーブル作成は CREATE TABLE IF NOT EXISTS を使う。
   • DDL実行後は必ず connection.commit() を行って確定する。
4. データ操作（DML）
   • レコードの挿入（INSERT）：cursor.execute() または cursor.executemany() を利用。
   • レコードの取得（SELECT）：fetchone(), fetchall(), fetchmany() を使い分ける。
   • レコードの更新（UPDATE）：cursor.rowcount で更新件数を確認し、connection.commit() で確定。
   • レコードの削除（DELETE）：同様に cursor.rowcount を取得し、connection.commit()／rollback() を使い分ける。
5. トランザクションと例外処理
   • try–except–finally を使い、例外時には rollback()、正常時には commit() を呼び、最後に close() して接続を解放する。
   • pymysql.MySQLError でMySQL関連のエラーを一括キャッチできる。
6. 演習問題での理解確認
   • 各種演習問題を通して、テーブル作成、データの挿入・取得・更新・削除、それぞれの判定条件やトランザクション管理を実践的に学んだ。

今後のステップ

• ORM（Object-Relational Mapping）ライブラリへの移行
  PyMySQLのような低レイヤーでのSQL操作に慣れたら、DjangoのORMやSQLAlchemyなどを活用し、より抽象度の高い形でデータベースを扱う方法も学ぶと効率的です。
• パフォーマンスチューニング
  大量データを扱う際にはバルクインサート、多段インデックスの設定、クエリの最適化などを検討するとよいでしょう。
• セキュリティ対策
  実運用環境では、SQLインジェクション対策や通信の暗号化、適切な権限設定などをしっかり行い、安全なアプリケーション開発を目指しましょう。

本記事を通じて、PythonとPyMySQLを使ったMySQL操作の基本をしっかり習得し、今後の開発にお役立ていただければ幸いです。

ぜひ演習問題にも取り組んで、実際にコードを書いて動作を確認しながら学習を進めてみてください。