Python3 MySQL(PyMySQL)
Connecting, parameterised queries, transactions, and pooling in a real service.
Key PointsNever build SQL with f-strings — always pass parameters; commit explicitly and roll back on error.
Install and Connect
Bash
pip install PyMySQL
# 或更完整的 mysqlclient（需要编译环境）Python
import pymysql
conn = pymysql.connect(
host="127.0.0.1", port=3306,
user="app", password="secret", database="shop",
charset="utf8mb4", autocommit=False,
)Query with Parameters
Python
with conn.cursor() as cur:
# 正确：占位符由驱动转义
cur.execute("SELECT id, name FROM users WHERE age > %s AND city = %s", (18, "Beijing"))
for uid, name in cur.fetchall():
print(uid, name)
# 错误示范：SQL 注入
# cur.execute(f"SELECT * FROM users WHERE name = '{name}'")%s is PyMySQL's placeholder, not Python string formatting. Passing a tuple as the second argument is what makes escaping work.Insert and Transactions
Python
try:
with conn.cursor() as cur:
cur.executemany(
"INSERT INTO orders (user_id, amount) VALUES (%s, %s)",
[(1, 99.5), (2, 20.0)],
)
new_id = cur.lastrowid
conn.commit()
except Exception:
conn.rollback()
raisePooling
A raw connection per request does not scale. Use a pool (DBUtils, SQLAlchemy) so connections are reused and bounded.
Python
from dbutils.pooled_db import PooledDB
import pymysql
POOL = PooledDB(
creator=pymysql, maxconnections=20, blocking=True,
host="127.0.0.1", user="app", password="secret", database="shop",
charset="utf8mb4", autocommit=False,
)
conn = POOL.connection()
try:
with conn.cursor() as cur:
cur.execute("SELECT 1")
finally:
conn.close() # 归还到池，不是真正关闭ORM Alternative
Python
from sqlalchemy import create_engine, text
engine = create_engine("mysql+pymysql://app:secret@127.0.0.1/shop", pool_size=20)
with engine.connect() as conn:
rows = conn.execute(text("SELECT id, name FROM users WHERE age > :age"), {"age": 18})
for r in rows:
print(r.id, r.name)Set
charset="utf8mb4" — plain utf8 in MySQL is 3-byte and breaks emoji and some CJK characters.安装与连接
Bash
pip install PyMySQL
# 或更完整的 mysqlclient（需要编译环境）Python
import pymysql
conn = pymysql.connect(
host="127.0.0.1", port=3306,
user="app", password="secret", database="shop",
charset="utf8mb4", autocommit=False,
)参数化查询
Python
with conn.cursor() as cur:
# 正确：占位符由驱动转义
cur.execute("SELECT id, name FROM users WHERE age > %s AND city = %s", (18, "Beijing"))
for uid, name in cur.fetchall():
print(uid, name)
# 错误示范：SQL 注入
# cur.execute(f"SELECT * FROM users WHERE name = '{name}'")%s 是 PyMySQL 的占位符，不是 Python 字符串格式化。把元组作为第二个参数传进去，转义才会生效。插入与事务
Python
try:
with conn.cursor() as cur:
cur.executemany(
"INSERT INTO orders (user_id, amount) VALUES (%s, %s)",
[(1, 99.5), (2, 20.0)],
)
new_id = cur.lastrowid
conn.commit()
except Exception:
conn.rollback()
raise连接池
每个请求新建连接无法扩展。用连接池（DBUtils、SQLAlchemy）复用并限制连接数。
Python
from dbutils.pooled_db import PooledDB
import pymysql
POOL = PooledDB(
creator=pymysql, maxconnections=20, blocking=True,
host="127.0.0.1", user="app", password="secret", database="shop",
charset="utf8mb4", autocommit=False,
)
conn = POOL.connection()
try:
with conn.cursor() as cur:
cur.execute("SELECT 1")
finally:
conn.close() # 归还到池，不是真正关闭ORM 方案
Python
from sqlalchemy import create_engine, text
engine = create_engine("mysql+pymysql://app:secret@127.0.0.1/shop", pool_size=20)
with engine.connect() as conn:
rows = conn.execute(text("SELECT id, name FROM users WHERE age > :age"), {"age": 18})
for r in rows:
print(r.id, r.name)连接参数写
charset="utf8mb4" —— MySQL 里的 utf8 是 3 字节，表情符号和部分中日韩字符会出错。