介绍 Python 中使用 SQLAlchemy 框架连接 MySQL 数据库的完整流程。涵盖环境搭建（Docker MySQL）、核心组件（Engine、Base、Session）解析、模型定义、CRUD 操作（增删改查）、表关系映射（一对多）以及性能优化与安全避坑建议。适合零基础开发者快速上手 ORM 开发。
作为 Python 生态中常用的 ORM 框架，SQLAlchemy 几乎是中大型 Python 项目的'标配'——它能帮你用 Python 类操作 MySQL 数据库，简化数据库操作。
本文针对初学者，从环境搭建→核心概念→CRUD 实战→关系映射，一步步讲清如何用 SQLAlchemy 连接 MySQL。全程代码可复制，旨在帮助读者掌握基础用法。
在写代码前，必须先搞定 3 件事：安装 MySQL、安装 Python 库、配置连接。
如果不想折腾 MySQL 的安装（比如 Windows 下的环境变量、权限问题），直接用 Docker 跑 MySQL——1 条命令就能启动：
# 启动 MySQL 8.0 容器（密码设为 root123，端口映射 3306）
docker run -d --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root123 mysql:8.0
验证是否启动成功：
docker ps # 看 mysql 容器是否在运行
需要两个库：
pymysql：Python 连接 MySQL 的驱动；sqlalchemy：核心 ORM 框架。用 pip 安装：
pip install pymysql sqlalchemy
先写一段代码，验证能否连接到 MySQL：
from sqlalchemy import create_engine # MySQL 连接字符串格式：mysql+pymysql://用户名：密码@主机：端口/数据库名 # 注意：首次连接需要先在 MySQL 中创建数据库（比如 test_db） engine = create_engine('mysql+pymysql://root:root123@localhost:3306/test_db') # 测试连接（无报错则成功） with engine.connect() as conn: print("连接成功！")
关键说明：
test_db 需要先在 MySQL 中创建（用 Navicat 或命令行 CREATE DATABASE test_db;）；host 填 localhost，密码填你安装时设置的密码。SQLAlchemy 的核心是 '用对象封装数据库操作'，你需要先理解 3 个关键组件：
创建 Engine 的代码：
from sqlalchemy import create_engine # 连接字符串格式：mysql+pymysql://用户名：密码@主机：端口/数据库名 engine = create_engine( 'mysql+pymysql://root:root123@localhost:3306/test_db', echo=True, # 打印执行的 SQL（调试用，上线关闭） pool_size=5, # 连接池大小（默认 5） max_overflow=10 # 超出连接池后的最大连接数（默认 10） )
创建 Base 的代码：
from sqlalchemy.orm import declarative_base # 创建模型基类（所有模型类都要继承它） Base = declarative_base()
创建 Session 的代码：
from sqlalchemy.orm import sessionmaker # 创建 Session 工厂（绑定 Engine） SessionLocal = sessionmaker(bind=engine) # 获取 Session 实例（每次操作都要新建 Session） session = SessionLocal()
总结：Engine→连接池→Session→操作数据库；Base→模型类→表结构。
现在，我们用 Base 定义一个 User 类，对应 MySQL 的 users 表。
from sqlalchemy import Column, Integer, String, DateTime from sqlalchemy.sql import func from datetime import datetime # 继承 Base 基类 class User(Base): __tablename__ = 'users' # 对应数据库的 users 表（必须小写） # 字段定义（Column 是表字段的装饰器） 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() 更准确）。定义好模型类后，需要用 Base 创建表：
# 创建所有继承 Base 的模型对应的表（第一次运行时执行） Base.metadata.create_all(bind=engine)
注意：
现在，我们用 Session 实现增删改查，全程不用写 SQL！
# 创建 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）。SQLAlchemy 提供了强大的查询 API，常用的有：
# 根据 id=1 查询用户 user = session.get(User, 1) if user: print(f"用户姓名：{user.name}，邮箱：{user.email}") else: print("用户不存在")
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}")
stmt = select(User).where(User.name.like('%张%')) users = session.execute(stmt).scalars().all()
# 先查询要修改的用户 user = session.get(User, 1) # 修改属性 user.age = 20 user.email = '[email protected]' # 提交修改（自动生成 UPDATE 语句） session.commit()
# 先查询要删除的用户 user = session.get(User, 1) # 删除实例 session.delete(user) # 提交删除（自动生成 DELETE 语句） session.commit()
实际项目中，表之间一定有关系（比如用户和订单的'一对多'），SQLAlchemy 能自动处理这些关系。
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') # back_populates 指向 User 的 orders 属性
class User(Base): __tablename__ = 'users' # 之前的字段... # 反向关系：User→Order（一对多） orders = relationship('Order', back_populates='user') # back_populates 指向 Order 的 user 属性
# 创建用户 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 互相引用）。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 分钟，避免连接失效） )
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}) # 安全！
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}")
解决：检查连接字符串中的密码，或在 MySQL 中给 root 用户授权：
GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY 'root123' WITH GRANT OPTION;
FLUSH PRIVILEGES;
unique=True 约束）；index=True 需要先创建索引（或模型类的 __tablename__ 写错）；__tablename__ 是否和数据库表名一致，或重新创建表（Base.metadata.drop_all()→Base.metadata.create_all()）。SQLAlchemy 是Python 中最常用的 ORM 框架，它的优势是：
适用场景：
joinedload（预加载关联数据，避免 N+1 查询）。最后，经验总结：'ORM 是工具，不是银弹'—— 简单的查询可以用 ORM，复杂的查询（比如多表联查、统计）建议用 SQLAlchemy Core 或原生 SQL，平衡开发效率和性能。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online