Python 与 MySQL 的现代连接之道：PyMySQL 实战与 2026 技术演进

# 使用 with 语句确保游标在使用后被正确关闭
with conn.cursor() as cursor:
# 将参数作为第二个元组传递
cursor.execute(sql, (username,))
result = cursor.fetchone()
return result

# 测试
user = get_user_by_name("Alice")
print(f"查询到的用户: {user}")
CODEBLOCK_b785c847python
from dbutils.pooled_db import PooledDB

# 全局初始化连接池（通常在应用启动时执行一次）
# 注意：连接池是线程安全的
db_pool = PooledDB(
creator=pymysql, # 指定底层驱动
maxconnections=20, # 最大连接数：根据数据库服务器配置调整
mincached=5, # 初始化时创建的空闲连接数
maxcached=10, # 最大空闲连接数
blocking=True, # 当连接池满时，是否阻塞等待而不是报错
ping=1, # 2026 最佳实践：定期 ping 检查连接有效性，防止掉线
host=config.host,
user=config.user,
password=config.password,
database=config.database,
charset=config.charset,
cursorclass=DictCursor
)

def execute_transaction(sql: str, params: tuple):
"""
从连接池获取连接，执行事务，并自动归还连接。
这里的 conn.close() 不会真正关闭 TCP 连接，而是将其归还给池子。
"""
conn = db_pool.connection()
try:
with conn.cursor() as cursor:
affected_rows = cursor.execute(sql, params)
conn.commit() # 显式提交事务
return affected_rows
except Exception as e:
conn.rollback() # 发生错误必须回滚
print(f"❌ 事务执行失败: {e}")
# 在生产环境，这里应该通过 Prometheus 或 Sentry 上报错误
raise
finally:
conn.close() # 归还连接

# 模拟并发操作
for i in range(10):
execute_transaction("UPDATE stats SET count = count + 1 WHERE id = %s", (1,))
CODEBLOCK_314c5cb8python
import json

# 模拟从 LLM (如 GPT-5 或 Claude 4) 获取的修复指令
# 在实际应用中，你会调用 OpenAI API，并 response_format={"type": "json_object"}
ai_generated_payload = {
"table": "orders",
"action": "update",
"set": {"status": "cancelled", "remark": "Auto-fixed by AI Agent 01"},
"where": {"id": 1024, "amount": -99.9}
}

def execute_ai_intent(intent: dict):
"""
安全地执行 AI 生成的数据库意图。
这里包含两层安全检查：
1. 白名单检查表名
2. 强制参数化查询
"""
allowed_tables = ["orders", "users", "logs"] # AI 只能操作这些表
table = intent.get("table")

if table not in allowed_tables:
raise PermissionError("AI Agent 试图访问非授权表，已拦截。")

# 构造安全的 SQL
set_clause = ", ".join([f"{k} = %s" for k in intent["set"].keys()])
where_clause = " AND ".join([f"{k} = %s" for k in intent["where"].keys()])
sql = f"UPDATE {table} SET {set_clause} WHERE {where_clause}"

# 合并参数
params = tuple(intent["set"].values()) + tuple(intent["where"].values())

print(f"🤖 AI 生成的 SQL: {sql}
参数: {params}")

conn = db_pool.connection()
try:
with conn.cursor() as cursor:
rows = cursor.execute(sql, params)
conn.commit()
return rows
except Exception as e:
conn.rollback()
# 将错误反馈给 AI，让它自我修正
return f"Error: {e}"

# 测试 AI 执行
execute_ai_intent(ai_generated_payload)
CODEBLOCK_b7e67de9python
conn = pymysql.connect(
...,
# 强制连接使用 UTC，与 Python 和 Docker 容器保持一致
init_command="SET time_zone=‘+00:00‘;"
)
CODEBLOCK_4677560fpythonndef export_large_table():
conn = db_pool.connection()
# 使用 SSCursor (服务端游标)
cursor = conn.cursor(pymysql.cursors.SSCursor)

cursor.execute("SELECT * FROM massive_log_table")

# 这里的循环不会把内存撑爆
for row in cursor:
# 处理每一行，例如写入 CSV 或 Kafka
process_row(row)