菜鸟教程 – 学的不仅是技术，更是梦想！

技术网站，更是交流网站

Python 3 教程

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

likuolei

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. 安装 MySQL 服务器（如果没有）：
  • Windows/Mac：下载 MySQL Community Server（官网 mysql.com）。
  • Linux：sudo apt install mysql-server (Ubuntu) 或 sudo yum install mysql-server (CentOS)。
  • 启动：mysql -u root -p 测试连接（默认密码为空或 root）。
  • 创建测试数据库：登录 MySQL 后，执行 CREATE DATABASE test_db CHARACTER SET utf8mb4;。
  1. 安装 Python 驱动（必须）：
   # 推荐官方驱动（支持连接池、预编译）
   pip install mysql-connector-python==9.0.0  # 2026 最新稳定版

   # 备用：纯 Python 驱动（无 C 依赖，轻便）
   pip install pymysql==1.1.1
  1. 测试环境：
  • Python 版本：python --version（≥3.8）。
  • MySQL 版本：mysql --version（≥8.0，支持 JSON 等新特性）。
  • 防火墙/权限：确保 Python 主机能访问 MySQL（默认 localhost:3306）。

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. 性能优化 & 最佳实践

优化点做法示例收益参考
连接复用用连接池，避免每次操作新 connect10-50x 加速
批量操作executemany() 而非循环 execute5-20x 快
索引利用SQL 加 WHERE id=xx（id 有索引）查询 100x+
预编译语句cursor.prepare(sql) 复用安全 + 快
读写分离主库写，从库读（config 切换 host）高可用
错误重试用 retry 库自动重连（pip install retry）鲁棒性 ↑
  • 监控：用 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）

错误代码/描述原因 & 解决
Access denied (1045)密码/用户错 → 检查 config
Can’t connect (2003)网络/端口 → ping host, 查防火墙
Unknown database (1049)数据库不存在 → CREATE DATABASE
SQL syntax error (1064)SQL 写错 → 用工具测试 SQL
Lost connection (2013)超时 → 加 ‘connection_timeout’: 300
Too many connections (1040)连接过多 → 用池，加大 max_connections
UnicodeEncodeError编码问题 → 加 charset=’utf8mb4′
OperationalError: 2055缓存 SHA2 → 改 auth_plugin
pymysql vs mysql-connector选 pymysql 如果无 C 扩展需求
Slow query加索引：ALTER TABLE ADD INDEX (col)

调试技巧：

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

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

网站 http://yingjuxia.com
文章已创建 3958

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

相关文章

开始在上面输入您的搜索词，然后按回车进行搜索。按ESC取消。

返回顶部