Quick start: Kết nối MySQL với Python trong 5 phút
Cài hai thư viện trước:
pip install PyMySQL mysql-connector-pythonĐây là đoạn kết nối cơ bản nhất với PyMySQL:
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()Chạy được rồi? Tốt. Nhưng trong môi trường thật, code kiểu này sẽ gây ra đủ loại vấn đề vào giờ cao điểm.
Vấn đề thật sự khi kết nối MySQL trong production
Database production của mình chạy MySQL 8.0 với khoảng 50GB data. Hồi trước mỗi request tạo một connection mới và đóng sau khi xong — hệ thống cứ chậm dần vào ban ngày. Sau khi profile lại mới thấy: phần lớn thời gian response bị ăn vào việc tạo TCP connection và MySQL authentication handshake, chứ không phải query chậm.
Ba vấn đề phổ biến nhất:
- Connection overhead: Tạo connection mới mỗi request tốn 20–100ms, nhân với 1000 request/phút là vấn đề nghiêm trọng.
- SQL Injection: Ghép string trực tiếp vào query là lỗ hổng bảo mật cực kỳ nguy hiểm.
- Transaction thiếu cẩn thận: Dữ liệu có thể bị corrupt nếu lỗi xảy ra giữa chừng mà không có rollback.
Giải pháp 1: Connection Pool — dùng lại connection thay vì tạo mới
Thay vì đóng/mở connection liên tục, Connection Pool giữ sẵn một nhóm connection để tái sử dụng. Với mysql-connector-python, setup pool chỉ cần vài dòng:
from mysql.connector import pooling
db_pool = pooling.MySQLConnectionPool(
pool_name="mypool",
pool_size=5, # Điều chỉnh theo workload
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
use_unicode=True
)
def get_connection():
return db_pool.get_connection()
# Sử dụng
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() # Trả connection về pool, không đóng thậtLưu ý: pool_size=5 phù hợp cho ứng dụng nhỏ đến vừa. Với traffic cao hơn, tăng lên 10–20 nhưng cũng cần kiểm tra max_connections trong MySQL config.
PyMySQL với SQLAlchemy Pool
Nếu bạn dùng PyMySQL và muốn có connection pool, kết hợp với SQLAlchemy là cách mình hay dùng nhất:
from sqlalchemy import create_engine, text
engine = create_engine(
"mysql+pymysql://myuser:mypassword@localhost/mydb?charset=utf8mb4",
pool_size=5,
max_overflow=10, # Cho phép tối đa 15 connection khi cần
pool_timeout=30, # Chờ tối đa 30s lấy connection từ pool
pool_recycle=1800, # Recycle connection sau 30 phút (tránh "MySQL has gone away")
pool_pre_ping=True # Kiểm tra connection còn sống trước khi dùng
)
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 là tham số mình luôn bật — nó tự động kiểm tra và reconnect nếu connection đã chết sau khi MySQL server restart hoặc idle quá lâu.
Giải pháp 2: Prepared Statements — vừa nhanh vừa an toàn
Prepared Statements giải quyết đồng thời hai vấn đề: ngăn SQL Injection và tăng performance khi chạy cùng một query nhiều lần.
Nguyên lý: Query gửi lên MySQL trước (không có data), MySQL parse và compile một lần. Sau đó chỉ truyền data vào qua parameter riêng biệt — MySQL không bao giờ interpret data đó như SQL code.
import pymysql
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
# ĐÚNG — Dùng parameterized query
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 là placeholder, không phải string format!
return cursor.fetchone()
# SAI — Ghép string trực tiếp (SQL Injection!)
def find_user_unsafe(email):
with conn.cursor() as cursor:
sql = f"SELECT * FROM users WHERE email = '{email}'" # NGUY HIỂM!
cursor.execute(sql)
return cursor.fetchone()Hàm find_user_unsafe với input email = "' OR '1'='1" sẽ trả về toàn bộ user trong database. Đây là SQL Injection kinh điển mà mình thấy vẫn còn xuất hiện trong codebase thực tế.
True Prepared Statements với mysql-connector-python
import mysql.connector
conn = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb'
)
cursor = conn.cursor(dictionary=True, prepared=True) # prepared=True bật server-side cache
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()Tham số prepared=True bật true prepared statements ở MySQL server level — server cache execution plan, query lần sau nhanh hơn đáng kể với database lớn.
Giải pháp 3: Transaction an toàn với Context Manager
Transaction đảm bảo tính toàn vẹn: hoặc tất cả thành công, hoặc rollback về trạng thái ban đầu. Mình đã từng chứng kiến hệ thống bị lỗi giữa chừng khi trừ tiền khách hàng rồi mà chưa kịp tạo order — không dùng transaction đúng cách là rủi ro thật sự.
import pymysql
from contextlib import contextmanager
@contextmanager
def db_transaction(conn):
"""Tự động commit hoặc 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 # Tắt autocommit để tự control transaction
)
def create_order(user_id, product_id, quantity, price):
with db_transaction(conn):
with conn.cursor() as cursor:
# Bước 1: Trừ tồn kho
cursor.execute(
"UPDATE products SET stock = stock - %s WHERE id = %s AND stock >= %s",
(quantity, product_id, quantity)
)
if cursor.rowcount == 0:
raise ValueError("Không đủ hàng trong kho")
# Bước 2: Tạo order
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
# Bước 3: Ghi log
cursor.execute(
"INSERT INTO transaction_logs (order_id, action, created_at) VALUES (%s, %s, NOW())",
(order_id, "ORDER_CREATED")
)
return order_id # Chỉ return khi commit thành côngNếu bất kỳ bước nào raise exception — dù là network error hay constraint violation — db_transaction tự động rollback toàn bộ. Không bao giờ xảy ra tình trạng trừ kho rồi mà không có order.
Nâng cao: Batch insert với executemany()
Khi cần insert hàng nghìn row, gọi execute() từng dòng rất chậm. Dùng executemany() để batch insert:
data = [
("[email protected]", "Alice", 1),
("[email protected]", "Bob", 1),
("[email protected]", "Charlie", 0),
# ... hàng nghìn row
]
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()Với database production 50GB data, executemany() import 10,000 row nhanh hơn khoảng 15–20 lần so với loop execute() từng row — sự khác biệt rõ nhất khi chạy data migration.
Tips thực tế từ kinh nghiệm vận hành
1. Luôn đặt timeout để tránh connection treo
conn = pymysql.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydb',
charset='utf8mb4',
connect_timeout=5, # Timeout kết nối: 5 giây
read_timeout=30, # Timeout đọc: 30 giây
write_timeout=30 # Timeout ghi: 30 giây
)2. Xử lý lỗi theo từng loại exception
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("Email đã tồn tại trong hệ thống")
else:
raise
except pymysql.err.OperationalError as e:
conn.rollback()
print(f"Lỗi kết nối database: {e}")
raise3. So sánh nhanh PyMySQL vs mysql-connector-python
- PyMySQL: Pure Python, không cần C extension, dễ deploy, phù hợp cho hầu hết use case. Nhược điểm: chậm hơn một chút khi query trả về lượng data lớn.
- mysql-connector-python: Official driver từ Oracle, có C extension tùy chọn, hỗ trợ true prepared statements server-side, built-in connection pool. Phù hợp cho production với traffic cao.
Mình dùng PyMySQL + SQLAlchemy cho web app Flask, còn mysql-connector-python cho các script ETL cần xử lý data lớn vì performance tốt hơn với prepared statements. Tùy use case mà chọn — cả hai đều đủ tốt khi dùng đúng cách.
