在 Python 中连接 MySQL 数据库，最常用、官方推荐且维护活跃的库是 **`mysql-connector-python`**（Oracle 官方驱动）或更流行的开源社区驱动 **`PyMySQL`**（纯 Python 实现，兼容 MySQL 协议）以及高性能的 **`mysqlclient`**（C 扩展，基于 MySQL C API，是 Django 默认推荐）。此外，现代项目常通过 **SQLAlchemy**（ORM）或 **aiomysql**（异步）进行更高层封装。 下面我将为你提供 **4 种主流方式的完整、可运行示例**，并说明适用场景、安装命令、连接原理及关键注意事项。 --- ### ✅ 方式 1：使用 `PyMySQL`（推荐入门 & 轻量项目） > ✅ 纯 Python、无编译依赖、安装简单、兼容性好（支持 Python 3.7+）、语法接近原生 MySQLdb > ❌ 性能略低于 `mysqlclient`（但对大多数 Web/API 应用足够） #### 🔧 安装： ```bash pip install PyMySQL ``` #### 🐍 连接与查询示例： ```python import pymysql # 1. 建立连接（请替换为你的实际配置） connection = pymysql.connect( host='localhost', # 数据库主机地址 user='root', # 用户名 password='your_pass', # 密码（生产环境建议从环境变量读取！） database='test_db', # 数据库名（可选，不填则需手动 SELECT） port=3306, # MySQL 默认端口 charset='utf8mb4', # 推荐：支持 emoji 和完整 UTF-8 cursorclass=pymysql.cursors.DictCursor # 返回字典而非元组，更易用 ) try: # 2. 创建游标对象 with connection.cursor() as cursor: # 3. 执行 SQL 查询（防 SQL 注入：务必用参数化查询！） cursor.execute("SELECT id, name FROM users WHERE age > %s", (18,)) result = cursor.fetchall() # 获取所有结果 print(result) # 输出示例: [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}] # 4. 提交事务（仅 INSERT/UPDATE/DELETE 需要；SELECT 不需要） # connection.commit() finally: # 5. 确保连接关闭（推荐用 with 或 try/finally） connection.close() ``` 📌 **关键说明**： - `cursorclass=pymysql.cursors.DictCursor` → 让 `fetchone()`/`fetchall()` 返回 `dict`，字段名即 key，避免下标错误。 - `%s` 是 PyMySQL 的**占位符**（不是 Python 字符串格式化！），自动转义防止 SQL 注入。 - 使用 `with connection.cursor() as cursor:` 可自动关闭游标（但**不关闭连接**），所以外层仍需 `close()` 或用上下文管理器包装连接（见进阶技巧）。 --- ### ✅ 方式 2：使用 `mysql-connector-python`（Oracle 官方驱动） > ✅ 官方支持、功能完整（如连接池、X Protocol）、文档规范 > ❌ 包体积较大、部分旧版本有 Unicode bug（建议 ≥ 8.0.33） #### 🔧 安装： ```bash pip install mysql-connector-python ``` #### 🐍 示例代码： ```python import mysql.connector from mysql.connector import Error try: connection = mysql.connector.connect( host='localhost', user='root', password='your_pass', database='test_db', charset='utf8mb4', autocommit=True # 自动提交，省去 manual commit() ) if connection.is_connected(): cursor = connection.cursor(dictionary=True) # 同样返回 dict cursor.execute("SELECT * FROM users LIMIT 3") records = cursor.fetchall() for row in records: print(f"ID: {row['id']}, Name: {row['name']}") except Error as e: print("数据库连接错误：", e) finally: if connection.is_connected(): cursor.close() connection.close() print("MySQL 连接已关闭") ``` ⚠️ 注意：`autocommit=True` 适合只读或简单写操作；事务控制需设为 `False` 并手动 `commit()`/`rollback()`。 --- ### ✅ 方式 3：使用 `mysqlclient`（Django 生产首选，高性能） > ✅ C 扩展，性能最优；Django 默认后端；支持连接池（需搭配 `django-db-geventpool` 等） > ❌ Windows 安装可能需预编译 wheel 或 VS Build Tools（Mac/Linux 更友好） #### 🔧 安装（Linux/macOS）： ```bash pip install mysqlclient ``` > Windows 用户推荐：`pip install --only-binary=all mysqlclient` 或从 [Christoph Gohlke 的非官方 wheel](https://www.lfd.uci.edu/~gohlke/pythonlibs/#mysqlclient) 下载 `.whl` #### 🐍 简单使用（接口类似 MySQLdb）： ```python import MySQLdb # 或 from MySQLdb import connect conn = MySQLdb.connect( host='localhost', user='root', passwd='your_pass', db='test_db', charset='utf8mb4' ) cursor = conn.cursor(MySQLdb.cursors.DictCursor) cursor.execute("SELECT * FROM users") print(cursor.fetchall()) conn.close() ``` 💡 提示：`mysqlclient` 是 `PyMySQL` 的“精神继承者”，API 高度兼容，可无缝替换（只需改 import）。 --- ### ✅ 方式 4：使用 SQLAlchemy（ORM + 连接池，企业级推荐） > ✅ 统一数据库抽象、内置连接池、事务管理、ORM 映射、支持多种方言（MySQL/PostgreSQL/SQLite） > ❌ 学习曲线稍陡，轻量脚本可能“杀鸡用牛刀” #### 🔧 安装： ```bash pip install sqlalchemy pymysql # 使用 PyMySQL 作为底层驱动 ``` #### 🐍 ORM + 原生 SQL 混合示例： ```python from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker # 创建带连接池的引擎（5 连接池 + 10 最大 overflow） engine = create_engine( "mysql+pymysql://root:your_pass@localhost:3306/test_db?charset=utf8mb4", pool_size=5, max_overflow=10, echo=False # 设为 True 可打印所有 SQL（调试用） ) # 方式 A：原生 SQL 查询（推荐简单场景） with engine.connect() as conn: result = conn.execute(text("SELECT * FROM users WHERE id = :user_id"), {"user_id": 1}) for row in result: print(row) # 方式 B：ORM（需定义模型类，此处略 —— 如需我可补充） ``` ✅ `create_engine` 自动管理连接池，`engine.connect()` 返回可复用的连接，无需手动 close（上下文管理自动处理）。 --- ### ⚠️ 重要安全与工程实践提醒： | 项目 | 建议 | |------|------| | 🔐 密码管理 | ❌ 硬编码密码 → ✅ 使用 `os.getenv("DB_PASSWORD")` 或 `python-decouple` / `dotenv` 加载 `.env` 文件 | | 🧩 连接复用 | 避免每次请求都 `connect()` → ✅ 使用连接池（SQLAlchemy / `pymysql.connections.ConnectionPool`） | | 🧹 资源释放 | 必须 `close()` 游标和连接，或用 `with` 语句确保释放（尤其在异常时） | | 🌐 字符集 | 务必设 `charset='utf8mb4'`（不是 `utf8`！MySQL 的 `utf8` 是阉割版，不支持 emoji） | | 🛡️ SQL 注入 | 所有用户输入必须用参数化查询（`%s` / `:name`），**绝不拼接字符串**！ | --- ### 💡 进阶技巧：带连接池的 PyMySQL 封装（生产可用） ```python import pymysql from pymysql import connections from pymysql.cursors import DictCursor class MySQLPool: def __init__(self, **config): self.config = config self.pool = connections.Pool(**config, max_connections=10) def get_connection(self): return self.pool.get_connection() # 使用 db = MySQLPool( host='localhost', user='root', password='your_pass', database='test_db', charset='utf8mb4', cursorclass=DictCursor ) conn = db.get_connection() try: with conn.cursor() as cur: cur.execute("SELECT NOW()") print(cur.fetchone()) finally: conn.close() # 归还到连接池，非真正关闭 ``` ---