クイックスタート：PythonでMySQLに接続する（5分）
まず2つのライブラリをインストールします：
pip install PyMySQL mysql-connector-pythonPyMySQLで最もシンプルな接続コードはこちらです：
import pymysql
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
try:
with conn.cursor() as cursor:
cursor.execute("SELECT VERSION()")
result = cursor.fetchone()
print(f"MySQL version: {result['VERSION()']}")
finally:
conn.close()動きましたか？よし、ここまでは準備です。しかし本番環境では、このようなコードはピーク時にさまざまな問題を引き起こします。
本番環境でMySQL接続を行う際の本当の問題点
私の本番データベースはMySQL 8.0で約50GBのデータを扱っています。以前はリクエストごとに新しいコネクションを作成して終了後に閉じていましたが、日中になるにつれてシステムが遅くなっていきました。プロファイリングで判明したのは、レスポンス時間のほとんどがクエリではなく、TCPコネクションの確立とMySQLの認証ハンドシェイクに費やされていたということです。
よく見られる3つの問題点：
- コネクションのオーバーヘッド：リクエストごとに新しいコネクションを作成すると20〜100msかかり、1分間に1,000リクエストあれば深刻な問題になります。
- SQLインジェクション：クエリに文字列を直接結合するのは非常に危険なセキュリティホールです。
- 不適切なTransaction処理：途中でエラーが発生してもrollbackがなければ、データが破損する可能性があります。
解決策1：Connection Pool — 毎回作成する代わりにコネクションを再利用する
コネクションを繰り返し開閉する代わりに、Connection Poolはコネクションのグループを常時保持して再利用します。mysql-connector-pythonであれば、Poolのセットアップは数行だけです：
from mysql.connector import pooling
db_pool = pooling.MySQLConnectionPool(
pool_name="mypool",
pool_size=5, # ワークロードに応じて調整
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
use_unicode=True
)
def get_connection():
return db_pool.get_connection()
# 使用例
conn = get_connection()
try:
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT id, name FROM users WHERE active = %s", (1,))
rows = cursor.fetchall()
print(rows)
finally:
conn.close() # Poolにコネクションを返す（実際には閉じない）注意：pool_size=5は小〜中規模のアプリケーションに適しています。トラフィックが多い場合は10〜20に増やしますが、MySQLの設定でmax_connectionsも確認が必要です。
PyMySQLとSQLAlchemy Poolの組み合わせ
PyMySQLを使っていてConnection Poolが欲しい場合、SQLAlchemyと組み合わせる方法が私のよく使うアプローチです：
from sqlalchemy import create_engine, text
engine = create_engine(
"mysql+pymysql://myuser:mypassword@localhost/mydb?charset=utf8mb4",
pool_size=5,
max_overflow=10, # 必要時は最大15コネクションまで許可
pool_timeout=30, # Poolからコネクションを取得するまで最大30秒待機
pool_recycle=1800, # 30分後にコネクションをリサイクル（"MySQL has gone away"を防ぐ）
pool_pre_ping=True # 使用前にコネクションの生存確認を行う
)
with engine.connect() as conn:
result = conn.execute(
text("SELECT id, name FROM users WHERE active = :active"),
{"active": 1}
)
for row in result:
print(row)pool_pre_ping=Trueは私が必ず有効にするパラメータです。MySQLサーバーの再起動や長時間のアイドル後にコネクションが切れていても、自動的に確認して再接続してくれます。
解決策2：Prepared Statements — 高速かつ安全
Prepared Statementsは2つの問題を同時に解決します：SQLインジェクションの防止と、同じクエリを繰り返し実行する際のパフォーマンス向上です。
仕組み：クエリをデータなしでMySQLに送り、MySQLが一度だけパースとコンパイルを行います。その後はパラメータとしてデータだけを渡します。MySQLはそのデータをSQLコードとして解釈することはありません。
import pymysql
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
# 正しい方法 — パラメータ化クエリを使用
def find_user(email):
with conn.cursor() as cursor:
sql = "SELECT id, name, email FROM users WHERE email = %s AND active = %s"
cursor.execute(sql, (email, 1)) # %sはプレースホルダー（文字列フォーマットではない）
return cursor.fetchone()
# 誤った方法 — 文字列を直接結合（SQLインジェクション！）
def find_user_unsafe(email):
with conn.cursor() as cursor:
sql = f"SELECT * FROM users WHERE email = '{email}'" # 危険！
cursor.execute(sql)
return cursor.fetchone()find_user_unsafe関数にemail = "' OR '1'='1"を入力すると、データベース内の全ユーザーが返されてしまいます。これは実際のコードベースでも今なお見かける、典型的なSQLインジェクションです。
mysql-connector-pythonでのTrue Prepared Statements
import mysql.connector
conn = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb'
)
cursor = conn.cursor(dictionary=True, prepared=True) # prepared=Trueでサーバーサイドキャッシュを有効化
sql = "SELECT id, name FROM orders WHERE user_id = %s AND status = %s"
cursor.execute(sql, (123, "completed"))
orders = cursor.fetchall()
cursor.close()
conn.close()prepared=TrueパラメータはMySQLサーバーレベルでtrue prepared statementsを有効にします。サーバーが実行計画をキャッシュするため、大規模なデータベースでは2回目以降のクエリが大幅に速くなります。
解決策3：Context Managerを使った安全なTransaction
Transactionはデータの整合性を保証します：すべて成功するか、初期状態にrollbackするかのどちらかです。顧客から引き落としが完了したのに注文が作成されないまま処理が途中で失敗するシステムを実際に見たことがあります。Transactionを正しく使わないことは本当のリスクです。
import pymysql
from contextlib import contextmanager
@contextmanager
def db_transaction(conn):
"""自動的にcommitまたはrollbackを行う。"""
try:
yield conn
conn.commit()
except Exception as e:
conn.rollback()
raise e
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
autocommit=False # autocommitを無効にしてTransactionを手動管理
)
def create_order(user_id, product_id, quantity, price):
with db_transaction(conn):
with conn.cursor() as cursor:
# ステップ1：在庫を減らす
cursor.execute(
"UPDATE products SET stock = stock - %s WHERE id = %s AND stock >= %s",
(quantity, product_id, quantity)
)
if cursor.rowcount == 0:
raise ValueError("在庫が不足しています")
# ステップ2：注文を作成
cursor.execute(
"INSERT INTO orders (user_id, product_id, quantity, total_price) VALUES (%s, %s, %s, %s)",
(user_id, product_id, quantity, price * quantity)
)
order_id = cursor.lastrowid
# ステップ3：ログを記録
cursor.execute(
"INSERT INTO transaction_logs (order_id, action, created_at) VALUES (%s, %s, NOW())",
(order_id, "ORDER_CREATED")
)
return order_id # commitが成功した場合のみreturnどのステップで例外が発生しても — ネットワークエラーであれ制約違反であれ — db_transactionが自動的にすべてをrollbackします。在庫を減らしたのに注文が存在しない状況は絶対に起きません。
発展：executemany()を使ったBatch Insert
数千行をinsertする場合、execute()を1行ずつ呼び出すのは非常に遅いです。executemany()を使ってbatch insertを行いましょう：
data = [
("[email protected]", "Alice", 1),
("[email protected]", "Bob", 1),
("[email protected]", "Charlie", 0),
# ... 数千行
]
try:
with conn.cursor() as cursor:
sql = "INSERT INTO users (email, name, active) VALUES (%s, %s, %s)"
cursor.executemany(sql, data)
conn.commit()
print(f"Inserted {cursor.rowcount} rows")
except Exception as e:
conn.rollback()
print(f"Error: {e}")
finally:
conn.close()50GBの本番データベースでは、executemany()で10,000行をimportすると、execute()のループより15〜20倍速くなります。data migrationを実行する際にその差が最も顕著です。
運用経験から学んだ実践的なTips
1. コネクションのハングを防ぐためタイムアウトを必ず設定する
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
connect_timeout=5, # 接続タイムアウト：5秒
read_timeout=30, # 読み取りタイムアウト：30秒
write_timeout=30 # 書き込みタイムアウト：30秒
)2. 例外の種類に応じたエラーハンドリング
import pymysql
try:
with conn.cursor() as cursor:
cursor.execute("INSERT INTO users (email) VALUES (%s)", ("[email protected]",))
conn.commit()
except pymysql.err.IntegrityError as e:
conn.rollback()
if e.args[0] == 1062: # Duplicate entry
print("このメールアドレスはすでに登録されています")
else:
raise
except pymysql.err.OperationalError as e:
conn.rollback()
print(f"データベース接続エラー：{e}")
raise3. PyMySQL vs mysql-connector-python 簡単比較
- PyMySQL：Pure Pythonで、C extensionが不要、deployも簡単で、ほとんどのユースケースに適しています。欠点：大量のデータを返すクエリでは若干遅くなります。
- mysql-connector-python：Oracleの公式ドライバで、オプションのC extension、server-sideのtrue prepared statements、built-in connection poolをサポートしています。トラフィックの高い本番環境に適しています。
私はFlaskのウェブアプリにはPyMySQL + SQLAlchemy、大量データを扱うETLスクリプトにはprepared statementsのパフォーマンスが優れるmysql-connector-pythonを使っています。ユースケースに応じて選んでください — 正しく使えばどちらも十分な性能を発揮します。
