【超详细教程】Python 连接 MySQL 全流程实战

Python 连接 MySQL 全流程实战教程（2026版超详细指南）

以下是基于 Python 3.12+ 和 MySQL 8.0+ 的从零到精通实战教程。我会按实际开发流程组织内容：环境准备 → 连接基础 → CRUD 操作 → 高级技巧 → 性能优化 → 完整项目示例 → 常见问题排查。

这个教程假设你有基本的 Python 知识（如变量、函数、异常处理）。我们优先使用 mysql-connector-python（Oracle 官方驱动，推荐），备用 pymysql（纯 Python，轻量）。

为什么学这个？

• MySQL 是最流行的关系型数据库（2026 年市场份额仍超 40%）。
• Python + MySQL 是 Web/App/数据分析的黄金组合（e.g., Django/Flask 后端、数据 ETL）。
• 掌握后，能轻松扩展到其他数据库（如 PostgreSQL via psycopg2）。

0. 环境准备（5 分钟搞定）

1. 基础连接（Hello World 级）

核心：用 connect() 创建连接对象。

import mysql.connector # 或 import pymysql

# 配置（替换为你自己的）
config = {
'host': 'localhost', # 服务器地址（云数据库用 IP/域名）
'port': 3306, # 默认端口
'user': 'root', # 用户名
'password': 'your_password', # 密码
'database': 'test_db', # 数据库名
'charset': 'utf8mb4', # 支持 emoji 等
'auth_plugin': 'mysql_native_password' # 兼容旧客户端
}

# 建立连接
try:
conn = mysql.connector.connect(**config) # pymysql 用 pymysql.connect(**config)
print("连接成功！MySQL 版本：", conn.get_server_info())
except mysql.connector.Error as err:
print(f"连接失败：{err}")
finally:
if 'conn' in locals() and conn.is_connected():
conn.close()
print("连接已关闭。")

输出示例：

连接成功！MySQL 版本： 8.0.35
连接已关闭。

关键点：

• 异常处理：总是用 try-except（常见错误：权限、密码、网络）。
• 关闭连接：防止资源泄漏（用 with 语句更好，后面讲）。
• pymysql 差异：几乎相同，但 pymysql 是线程安全的（多线程首选）。

2. CRUD 操作全解（Create/Read/Update/Delete）

用 游标（Cursor） 执行 SQL。

• 步骤：连接 → 创建游标 → 执行 SQL → 处理结果 → 提交/回滚 → 关闭。

2.1 Create（插入数据）

先建表（只执行一次）：

# 建表 SQL（示例：用户表）
create_table_sql = """
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
age INT,
email VARCHAR(100) UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""

conn = mysql.connector.connect(**config)
cursor = conn.cursor()
cursor.execute(create_table_sql)
conn.commit() # DDL 也需 commit
cursor.close()
conn.close()

插入单条/多条：

conn = mysql.connector.connect(**config)
cursor = conn.cursor()

# 单条插入
insert_sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
data = ("Alice", 28, "alice@example.com")
cursor.execute(insert_sql, data)

# 多条插入（高效）
multi_data = [
("Bob", 35, "bob@example.com"),
("Charlie", 22, "charlie@example.com")
]
cursor.executemany(insert_sql, multi_data)

conn.commit() # 必须 commit，否则数据不保存
print(f"插入 {cursor.rowcount} 条记录。")
cursor.close()
conn.close()

安全提示：用 %s 占位符防 SQL 注入（预编译）。

2.2 Read（查询数据）

conn = mysql.connector.connect(**config)
cursor = conn.cursor(dictionary=True) # 返回 dict，便于访问 (e.g., row['name'])

# 查询所有
select_all_sql = "SELECT * FROM users ORDER BY age DESC"
cursor.execute(select_all_sql)
rows = cursor.fetchall() # 全部结果
for row in rows:
print(row) # e.g., {'id': 1, 'name': 'Alice', …}

# 查询单条/带条件
select_one_sql = "SELECT name, age FROM users WHERE id = %s"
cursor.execute(select_one_sql, (1,))
row = cursor.fetchone() # 第一条
print("单条：", row)

# 分页（LIMIT + OFFSET）
paged_sql = "SELECT * FROM users LIMIT %s OFFSET %s"
cursor.execute(paged_sql, (5, 0)) # 前5条，从0开始
rows = cursor.fetchall()

cursor.close()
conn.close()

输出示例：

{'id': 2, 'name': 'Bob', 'age': 35, 'email': 'bob@example.com', 'created_at': datetime.datetime(2026, 1, 21, 18, 0, 0)}
单条： {'name': 'Alice', 'age': 28}

进阶：用 buffered=True 游标处理大结果集。

2.3 Update（更新数据）

conn = mysql.connector.connect(**config)
cursor = conn.cursor()

update_sql = "UPDATE users SET age = %s WHERE name = %s"
cursor.execute(update_sql, (30, "Alice"))

conn.commit()
print(f"更新 {cursor.rowcount} 条记录。")
cursor.close()
conn.close()

2.4 Delete（删除数据）

conn = mysql.connector.connect(**config)
cursor = conn.cursor()

delete_sql = "DELETE FROM users WHERE age < %s"
cursor.execute(delete_sql, (25,))

conn.commit()
print(f"删除 {cursor.rowcount} 条记录。")
cursor.close()
conn.close()

事务管理：多操作用 conn.start_transaction() 和 conn.rollback()（出错回滚）。

3. 用 with 语句管理连接/游标（推荐实践）

简化 finally/close 逻辑。

import mysql.connector

with mysql.connector.connect(**config) as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT COUNT(*) FROM users")
count = cursor.fetchone()[0]
print(f"总记录数：{count}")
# with 结束自动 commit/close
# conn 也自动关闭

4. 高级技巧（生产级应用）

4.1 连接池（高并发必备）

mysql-connector 支持内置池。

from mysql.connector import pooling

pool = pooling.MySQLConnectionPool(
pool_name="mypool",
pool_size=10, # 最大连接数
**config
)

# 获取连接
conn = pool.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users LIMIT 1")
print(cursor.fetchone())
cursor.close()
conn.close() # 归还到池

优势：复用连接，减少开销（Web App 首选）。

4.2 存储过程/批量操作

调用存储过程：

cursor.callproc("my_stored_proc", (arg1, arg2))
for result in cursor.stored_results():
print(result.fetchall())

批量插入（10万+ 数据）： 用 executemany() + 事务。

4.3 Pandas 集成（数据分析神器）

import pandas as pd
import mysql.connector

conn = mysql.connector.connect(**config)
df = pd.read_sql("SELECT * FROM users WHERE age > 25", con=conn)
print(df.head())

# 写回数据库
df.to_sql("users_backup", con=conn, if_exists="replace", index=False)
conn.close()

2026 提示：Pandas 3.0+ 支持 Arrow backend，读写更快。

4.4 异步连接（asyncio，高吞吐）

用 aiomysql（基于 pymysql）。

pip install aiomysql

import asyncio
import aiomysql

async def main():
pool = await aiomysql.create_pool(**config)
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT * FROM users")
rows = await cur.fetchall()
print(rows)
pool.close()
await pool.wait_closed()

asyncio.run(main())

5. 性能优化 & 最佳实践

• 监控：用 conn.cmd_query("SHOW STATUS LIKE 'Threads%'") 查连接数。
• 安全：用环境变量存密码（os.getenv(“DB_PASS”)）。
• 日志：加 conn.autocommit = False 手动控制。

6. 完整项目实战：简单用户管理系统

项目结构：

user_manager/
├── config.py # 配置
├── db.py # 数据库操作
└── main.py # 入口

config.py：

DB_CONFIG = {
'host': 'localhost',
'user': 'root',
'password': 'your_password',
'database': 'test_db',
'charset': 'utf8mb4'
}

db.py：

import mysql.connector
from config import DB_CONFIG

def get_connection():
return mysql.connector.connect(**DB_CONFIG)

def add_user(name, age, email):
with get_connection() as conn:
with conn.cursor() as cursor:
sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
cursor.execute(sql, (name, age, email))
conn.commit()

def get_users():
with get_connection() as conn:
with conn.cursor(dictionary=True) as cursor:
cursor.execute("SELECT * FROM users")
return cursor.fetchall()

# 类似地加 update/delete

main.py：

from db import add_user, get_users

add_user("David", 40, "david@example.com")
users = get_users()
for user in users:
print(user)

运行：python main.py 输出：所有用户列表。

扩展：加 Flask API（/users GET/POST）变成 Web 服务。

7. 常见问题排查（Top 10）

调试技巧：

• 打印 err.errno / err.msg。
• 用 MySQL Workbench 测试 SQL。
• 日志：加 mysql.connector.logging 配置。

恭喜！你现在能独立用 Python 操作 MySQL 了～ 有具体部分想深入？如 ORM（SQLAlchemy）、云数据库（AWS RDS）、大数据集成（PySpark + MySQL）？告诉我，我继续展开！