以下是使用 **Python** 连接 MySQL 并完成 `employee` 表的 **创建、插入、更新、删除（CRUD）** 操作的两种主流方式： ✅ **`pymysql`（轻量、原生 SQL、适合学习与简单脚本）** ✅ **`SQLAlchemy`（ORM + Core，支持面向对象操作与数据库抽象，推荐生产使用）** 我们严格遵循你原始需求：建表 → 插入 John Doe → 更新 salary → 删除该记录，并包含**错误处理、事务控制、资源安全释放**等生产级实践。 --- ### ✅ 方式一：使用 `pymysql`（纯 SQL 驱动） > ⚠️ 前置要求：安装驱动 > ```bash > pip install pymysql > ``` ```python import pymysql from pymysql import MySQLError # 🔐 数据库连接配置（请按实际修改） config = { 'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'your_password', # ❗生产环境建议从环境变量读取 'database': 'testdb', # ✅ 请提前创建好该数据库：CREATE DATABASE testdb CHARACTER SET utf8mb4; 'charset': 'utf8mb4', 'autocommit': False # 关键：关闭自动提交，以便手动控制事务 } try: # 🌐 建立连接 connection = pymysql.connect(**config) cursor = connection.cursor() # 1️⃣ 创建 employee 表（含 DROP 避免重复） create_table_sql = """ DROP TABLE IF EXISTS employee; CREATE TABLE employee ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100) NOT NULL, department VARCHAR(100), salary DECIMAL(10, 2) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ # 注意：pymysql 不支持单次执行多条语句（分号分隔），需拆分执行 cursor.execute("DROP TABLE IF EXISTS employee;") cursor.execute(""" CREATE TABLE employee ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100) NOT NULL, department VARCHAR(100), salary DECIMAL(10, 2) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """) # 2️⃣ 插入数据（✅ 参数化防止 SQL 注入！） insert_sql = "INSERT INTO employee (name, department, salary) VALUES (%s, %s, %s)" cursor.execute(insert_sql, ('John Doe', 'HR', 5000.00)) # 3️⃣ 更新数据（✅ 用 WHERE + 参数化） update_sql = "UPDATE employee SET salary = %s WHERE name = %s" cursor.execute(update_sql, (5500.00, 'John Doe')) # 4️⃣ 删除数据（✅ 同样参数化） delete_sql = "DELETE FROM employee WHERE name = %s" cursor.execute(delete_sql, ('John Doe',)) # ✅ 提交事务（所有操作成功才写入数据库） connection.commit() print("✅ CRUD 操作全部成功！") except MySQLError as e: # ❌ 出错则回滚，避免部分执行导致数据不一致 connection.rollback() print(f"❌ 数据库错误：{e}") finally: # 🧹 安全关闭资源 if cursor: cursor.close() if connection: connection.close() ``` **关键点解释：** - `autocommit=False`：启用事务，必须显式调用 `commit()` 或 `rollback()`。 - `cursor.execute(sql, params)`：**参数化查询** —— `%s` 占位符由 pymysql 自动转义，彻底杜绝 SQL 注入（⚠️ 绝对不要用 `f"INSERT ... '{name}'"`！）。 - `DROP TABLE IF EXISTS` + `CREATE TABLE`：确保每次运行都是干净环境（测试友好）。 - `connection.rollback()`：捕获异常后立即回滚，保障原子性。 --- ### ✅ 方式二：使用 `SQLAlchemy`（ORM + Core 混合风格） > ⚠️ 前置要求： > ```bash > pip install sqlalchemy pymysql > ``` #### ✨ 推荐写法：**Core（SQL 表达式语言）+ ORM 元数据定义**（兼顾灵活性与可维护性） ```python from sqlalchemy import create_engine, text, Column, Integer, String, Numeric, MetaData, Table from sqlalchemy.exc import SQLAlchemyError from contextlib import contextmanager # 🔐 连接字符串（格式：mysql+pymysql://user:password@host:port/dbname） DATABASE_URL = "mysql+pymysql://root:your_password@127.0.0.1:3306/testdb?charset=utf8mb4" # 🌐 创建引擎（pool_recycle 防止 MySQL 8 小时超时） engine = create_engine( DATABASE_URL, echo=False, # 设为 True 可打印所有 SQL（调试用） pool_recycle=3600, pool_pre_ping=True ) # 📜 定义 employee 表结构（元数据，用于建表） metadata = MetaData() employee_table = Table( 'employee', metadata, Column('id', Integer, primary_key=True, autoincrement=True), Column('name', String(100), nullable=False), Column('department', String(100)), Column('salary', Numeric(10, 2)) ) def init_table(): """创建/重置 employee 表""" with engine.begin() as conn: # 自动 commit/rollback conn.execute(text("DROP TABLE IF EXISTS employee")) metadata.create_all(conn) def main_crud(): try: with engine.begin() as conn: # ✅ 事务上下文管理器 # 1️⃣ 创建表 init_table() # 2️⃣ 插入（使用 ORM Table 对象 + 字典参数） conn.execute( employee_table.insert(), {"name": "John Doe", "department": "HR", "salary": 5000.00} ) # 3️⃣ 更新（使用 text() 执行带 WHERE 的 SQL，更直观） conn.execute( text("UPDATE employee SET salary = :new_salary WHERE name = :name"), {"new_salary": 5500.00, "name": "John Doe"} ) # 4️⃣ 删除 conn.execute( text("DELETE FROM employee WHERE name = :name"), {"name": "John Doe"} ) print("✅ SQLAlchemy Core CRUD 成功完成！") except SQLAlchemyError as e: print(f"❌ SQLAlchemy 错误：{e}") raise # re-raise for outer handling if needed # 🚀 执行 if __name__ == "__main__": main_crud() ``` #### 💡 Bonus：纯 ORM 风格（定义 Python 类映射表） ```python from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class Employee(Base): __tablename__ = 'employee' id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(100), nullable=False) department = Column(String(100)) salary = Column(Numeric(10, 2)) # 创建会话 Session = sessionmaker(bind=engine) session = Session() try: # 插入 emp = Employee(name="John Doe", department="HR", salary=5000.00) session.add(emp) session.flush() # 获取自增 id（如需） # 更新（通过对象） emp.salary = 5500.00 # 删除（通过对象） session.delete(emp) session.commit() print("✅ ORM 风格 CRUD 完成！") except Exception as e: session.rollback() print(f"❌ ORM 错误：{e}") finally: session.close() ``` **SQLAlchemy 优势总结：** | 特性 | 说明 | |------|------| | **数据库无关性** | 换 PostgreSQL/SQLite 只需改 `DATABASE_URL`，SQL 逻辑不变 | | **连接池 & 自动重连** | `pool_recycle` + `pool_pre_ping` 避免 MySQL `Lost connection` | | **ORM 映射** | 直接操作 Python 对象，无需手写 SQL（适合复杂业务逻辑） | | **Schema 管理** | `metadata.create_all()` 自动建表；配合 Alembic 可做迁移（`ALTER TABLE`） | --- ### 🛑 常见陷阱与避坑指南 | 问题 | 正确做法 | |------|----------| | ❌ `pymysql` 报错 `Packet sequence number wrong` | 检查 `pool_recycle` 和 `pool_pre_ping=True`（MySQL 默认 wait_timeout=28800 秒） | | ❌ 中文乱码 | 连接字符串加 `?charset=utf8mb4`，建表指定 `DEFAULT CHARSET=utf8mb4` | | ❌ `sqlalchemy.exc.InvalidRequestError: This Session's transaction has been rolled back` | 不要复用已 rollback 的 session，用 `sessionmaker()` 新建或 `session.rollback()` 后 `session.expunge_all()` | | ❌ 忘记 `commit()` 导致数据不入库 | 使用 `with engine.begin():` 上下文管理器（自动 commit/rollback） | --- ### 📦 附：一键运行的完整可执行脚本（含依赖检查） ```python # save as mysql_crud_demo.py import sys def check_deps(): try: import pymysql import sqlalchemy except ImportError as e: print(f"❌ 缺少依赖：{e}") print("👉 运行：pip install pymysql sqlalchemy") sys.exit(1) if __name__ == "__main__": check_deps() # 此处插入上面任一完整示例（推荐 pymysql 版本，最简明） # ...（粘贴 pymysql 示例代码） ``` ---