Python 使用 SQLAlchemy 操作 MySQL 入门与实战

2. Base：所有模型的'模板'

• 作用：定义模型类的基类，自动映射类到数据库表；
• 类比：像'建筑图纸'，所有房子（模型类）都基于这个图纸建造。

创建 Base 的代码：

from sqlalchemy.orm import declarative_base

Base = declarative_base()

3. Session：操作数据库的'把手'

• 作用：负责和数据库交互（新增、查询、更新、删除）；
• 类比：像'遥控器'，通过它控制电视（数据库）的操作。

创建 Session 的代码：

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)  # 创建 Session 工厂（绑定 Engine）
session = SessionLocal()                  # 获取 Session 实例（每次操作都要新建 Session）

总结：Engine→连接池→Session→操作数据库；Base→模型类→表结构。

三、实战 1：定义第一个模型类 ——User 表

现在，我们用 Base 定义一个 User 类，对应 MySQL 的 users 表。

1. 定义模型类

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from datetime import datetime

# 继承 Base 基类
class User(Base):
    __tablename__ = 'users'  # 对应数据库的 users 表（必须小写）
    
    id = Column(Integer, primary_key=True, index=True)  # 主键（自增）
    name = Column(String(50), nullable=False)            # 姓名（非空）
    email = Column(String(100), unique=True, index=True) # 邮箱（唯一，加索引）
    age = Column(Integer)                                # 年龄（可选）
    created_at = Column(DateTime(timezone=True), default=func.now())  # 创建时间（自动填充当前时间）

关键说明：

• __tablename__：模型类对应的表名（必须和数据库中的表名一致，MySQL 默认小写）；
• primary_key=True：标记为主键（自增）；
• nullable=False：字段非空（插入数据时必须传值）；
• unique=True：字段唯一（避免重复邮箱）；
• func.now()：用 MySQL 的 NOW() 函数自动填充当前时间（比 Python 的 datetime.now() 更准确）。

2. 创建表结构

定义好模型类后，需要用 Base 创建表：

# 创建所有继承 Base 的模型对应的表（第一次运行时执行）
Base.metadata.create_all(bind=engine)

注意：

• 这段代码只会创建不存在的表（不会覆盖已有表）；
• 如果修改了模型类（比如新增字段），需要用迁移工具（如 Alembic）更新表结构（本文暂不展开）。

四、实战 2：CRUD 操作

现在，我们用 Session 实现增删改查，全程不用写 SQL！

1. 新增数据（Create）

# 创建 User 实例（相当于数据库中的一行数据）
user = User(name='张三', email='[email protected]', age=18)

# 将实例添加到 Session（相当于'暂存'）
session.add(user)

# 提交 Session（相当于'确认'，将数据写入数据库）
session.commit()

# 刷新实例（获取自增的 id）
session.refresh(user)
print(f"新增用户 ID：{user.id}")

关键说明：

• add() 只是将数据'暂存'到 Session，必须 commit() 才会写入数据库；
• refresh() 会从数据库中获取最新数据（比如自增的 id）。

2. 查询数据（Read）

SQLAlchemy 提供了强大的查询 API，常用的有：

（1）根据主键查询（最常用）

# 根据 id=1 查询用户
user = session.get(User, 1)
if user:
    print(f"用户姓名：{user.name}，邮箱：{user.email}")
else:
    print("用户不存在")

（2）条件查询（比如查年龄≥18 的用户）

from sqlalchemy import select

# 构建查询语句（SELECT * FROM users WHERE age >= 18）
stmt = select(User).where(User.age >= 18)

# 执行查询（返回所有符合条件的用户）
users = session.execute(stmt).scalars().all()
for user in users:
    print(f"ID：{user.id}，姓名：{user.name}")

（3）模糊查询（比如查姓名包含'张'的用户）

stmt = select(User).where(User.name.like('%张%'))
users = session.execute(stmt).scalars().all()

3. 更新数据（Update）

# 先查询要修改的用户
user = session.get(User, 1)

# 修改属性
user.age = 20
user.email = '[email protected]'

# 提交修改（自动生成 UPDATE 语句）
session.commit()

4. 删除数据（Delete）

# 先查询要删除的用户
user = session.get(User, 1)

# 删除实例
session.delete(user)

# 提交删除（自动生成 DELETE 语句）
session.commit()

五、实战 3：关系映射

实际项目中，表之间一定有关系（比如用户和订单的'一对多'），SQLAlchemy 能自动处理这些关系。

1. 定义关联模型（比如 Order 表）

class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey('users.id'))  # 外键（关联 users 表的 id）
    product = Column(String(100), nullable=False)
    price = Column(Integer, nullable=False)
    created_at = Column(DateTime(timezone=True), default=func.now())
    
    # 关系映射：Order→User（多对一）
    user = relationship('User', back_populates='orders')

2. 在 User 表中添加反向关系

class User(Base):
    __tablename__ = 'users'
    # 之前的字段...
    
    # 反向关系：User→Order（一对多）
    orders = relationship('Order', back_populates='user')

3. 测试关系映射

# 创建用户
user = User(name='李四', email='[email protected]', age=25)
session.add(user)
session.commit()

# 创建订单（自动关联 user_id）
order1 = Order(user_id=user.id, product='手机', price=5999)
order2 = Order(user_id=user.id, product='电脑', price=9999)
session.add_all([order1, order2])
session.commit()

# 查询用户的所有订单（通过 user.orders）
user = session.get(User, user.id)
for order in user.orders:
    print(f"订单 ID：{order.id}，商品：{order.product}，价格：{order.price}")

关键说明：

• ForeignKey('users.id')：Order 表的 user_id 关联 User 表的 id（外键约束）；
• relationship：定义模型之间的关系（比如 User 的 orders 属性会自动查询该用户的所有订单）；
• back_populates：双向关联（User 的 orders 和 Order 的 user 互相引用）。

六、进阶优化与注意事项

1. 连接池配置（提升性能）

Engine 默认开启连接池（大小 5），可以通过 create_engine 的参数优化：

engine = create_engine(
    'mysql+pymysql://root:root123@localhost:3306/test_db',
    pool_size=10,           # 连接池大小（默认 5）
    max_overflow=20,        # 超出连接池后的最大连接数（默认 10）
    pool_timeout=30,        # 获取连接的超时时间（默认 30 秒）
    pool_recycle=1800       # 连接的存活时间（1800 秒=30 分钟，避免连接失效）
)

2. 避免 SQL 注入（安全第一）

SQLAlchemy 的 query 和 execute 方法会自动处理 SQL 注入（用参数化查询），禁止手动拼接 SQL！

错误示例（会被 SQL 注入）：

name = "张三'; DROP TABLE users; --"
session.execute(f"SELECT * FROM users WHERE name='{name}'")  # 危险！

正确示例（参数化查询）：

from sqlalchemy import text

name = "张三'; DROP TABLE users; --"
# 用 text() 和参数化查询
stmt = text("SELECT * FROM users WHERE name=:name")
session.execute(stmt, {"name": name})  # 安全！

3. 事务处理（避免数据不一致）

SQLAlchemy 的 Session 默认开启事务，必须 commit() 才会提交，rollback() 可以回滚：

try:
    user = User(name='王五', email='[email protected]')
    session.add(user)
    # 模拟错误（比如除以 0）
    1 / 0
    session.commit()
except Exception as e:
    session.rollback()  # 回滚事务（取消之前的操作）
    print(f"错误：{e}")

七、常见问题解答

1. 连接报错 'Access denied for user 'root'@'localhost''

• 原因：密码错误或用户没有权限；
• 解决：检查连接字符串中的密码，或在 MySQL 中给 root 用户授权：

GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY 'root123' WITH GRANT OPTION;
FLUSH PRIVILEGES;

2. 新增数据时报 'Duplicate entry '[email protected]' for key 'email''

• 原因：邮箱重复（unique=True 约束）；
• 解决：确保插入的邮箱唯一，或在代码中处理重复错误。

3. 查询结果为空，但数据库中有数据

• 原因：index=True 需要先创建索引（或模型类的 __tablename__ 写错）；
• 解决：检查模型类的 __tablename__ 是否和数据库表名一致，或重新创建表（Base.metadata.drop_all() → Base.metadata.create_all()）。

八、总结

SQLAlchemy 是 Python 中最常用的 ORM 框架之一，其优势包括：

• 跨数据库支持：支持 MySQL、PostgreSQL、SQLite 等，切换数据库只需修改连接字符串；
• 强大的查询 API：支持复杂查询（联表、分组、排序），不用写 SQL；
• 事务管理：自动处理事务，避免数据不一致；
• 性能优化：连接池、索引、延迟加载等功能提升性能。

适用场景：

• 中大型项目（需要复杂的数据库操作）；
• 多数据库切换（比如从 SQLite 迁移到 MySQL）；
• 需要事务管理和安全的项目（比如电商、金融）。

九、后续学习建议

• 学习 Alembic：数据库迁移工具（修改模型类后，自动更新表结构）；
• 学习 SQLAlchemy Core：直接写 SQL 表达式（比 ORM 更灵活）；
• 学习 查询优化：比如 joinedload（预加载关联数据，避免 N+1 查询）。

经验总结：ORM 是工具，不是银弹——简单的查询可以用 ORM，复杂的查询（比如多表联查、统计）建议用 SQLAlchemy Core 或原生 SQL，平衡开发效率和性能。