1. SQLAlchemy 是什么？ #

• SQLAlchemy 是 Python 最流行的 ORM（对象关系映射） 框架，用 Python 类和对象操作数据库，而不是手写 SQL。

1.1 为什么用 SQLAlchemy？ #

• 开发效率：复杂查询用 Python 表达，逻辑更清晰、可复用。
• 易维护：表结构变更主要改模型类，不必大面积改 SQL。
• 跨数据库：SQLAlchemy 支持多种数据库；本教程统一使用 MySQL。
• 更安全：参数自动绑定，降低 SQL 注入风险。
• 生态完整：事务、关系映射、迁移（Alembic）、异步 ORM 等。

2. 前置知识 #

2.1 什么是 ORM？ #

• ORM 把数据库表映射为 Python 类，行映射为对象，列映射为属性。
• 表 users → 类 User；一行 → User 实例；列 name → user.name。

2.2 数据库基础术语 #

• 表（Table）：存储结构
• 行（Row）：一条记录
• 列（Column）：字段
• 主键（Primary Key）：唯一标识
• 外键（Foreign Key）：关联其他表

3. 安装与环境准备 #

3.1 安装 SQLAlchemy #

使用 pip 安装 SQLAlchemy 与 MySQL 驱动：

# 安装 SQLAlchemy 与 MySQL 同步驱动
pip install "sqlalchemy" pymysql

3.2 MySQL 环境准备 #

请先安装并启动 MySQL，创建数据库：

# 创建数据库
CREATE DATABASE aidb
# 设置字符集 utf8mb4，支持中文
  CHARACTER SET utf8mb4
# 设置排序规则
  COLLATE utf8mb4_unicode_ci;

连接 URL 格式（下文示例统一使用，请按本地环境修改用户名、密码）：

# 同步连接 URL 格式说明
mysql+pymysql://用户名:密码@主机:端口/数据库名?charset=utf8mb4

# 连接 URL
mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4

异步 ORM 需额外安装：

# 安装异步扩展与 MySQL 异步驱动
pip install "sqlalchemy[asyncio]" aiomysql

4. 核心概念：Engine、Base、Session #

4.1 Engine（引擎） #

• Engine 负责数据库连接和 SQL 方言；echo=True 可打印执行的 SQL。

# 从 sqlalchemy 导入 create_engine
from sqlalchemy import create_engine

# 定义 MySQL 连接 URL（按本地修改用户名、密码、主机、库名）
DATABASE_URL = (
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4"
)
# 创建引擎：echo 打印 SQL，pool_pre_ping 检测连接可用性
engine = create_engine(DATABASE_URL, echo=True, pool_pre_ping=True)

• pool_pre_ping=True：连接前检测可用性，避免 MySQL 空闲断连后报错。
• URL 中 charset=utf8mb4：支持中文及 emoji。

4.2 Base（声明式基类） #

• 所有 ORM 模型继承 Base（基于 DeclarativeBase）。

# 从 sqlalchemy.orm 导入 DeclarativeBase
from sqlalchemy.orm import DeclarativeBase

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

4.3 Session（会话） #

• Session 是 ORM 操作入口；with 退出时关闭会话，未提交事务会回滚。
• 写入数据库须 commit()，或使用 session.begin() 自动提交。

# 引入 Integer、String、Column 以定义数据库字段类型
from sqlalchemy import Integer, String, Column
# 引入声明式基类和 Session 对象
from sqlalchemy.orm import DeclarativeBase,Session
# 引入创建数据库引擎的方法
from sqlalchemy import create_engine

# 定义声明式基类，用于所有 ORM 模型继承
class Base(DeclarativeBase):
    pass

# 定义用户模型 User，继承 Base
class User(Base):
    # 指定表名为 users
    __tablename__ = "users"
    # 定义 id 主键字段，类型为 Integer
    id = Column(Integer, primary_key=True)
    # 定义 name 字段，类型为 String(50)，长度 50
    name = Column(String(50))
    # 
    city: Mapped[str] = mapped_column(String(50))


# 定义数据库连接字符串，指定 MySQL 用户名、密码、主机、数据库及字符集
DATABASE_URL = "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4"

# 创建数据库引擎，echo=True 打印执行的 SQL，pool_pre_ping 检查连接可用性
engine = create_engine(DATABASE_URL, echo=True, pool_pre_ping=True)
# 
Base.metadata.drop_all(engine)
# 创建所有模型定义的表（仅首次运行需要）
Base.metadata.create_all(engine)

# 新建一个 User 实例，name 字段为 "Alice"
user = User(name="Alice")

# 使用 with 上下文管理器来开启一个 Session 会话
with Session(engine) as session:
    # 将新用户对象 user 添加到 session 会话中（处于待提交状态）
    session.add(user)
    # 提交 session，将新增用户写入数据库
    session.commit()

自动提交写法：

# 自动提交写法一：Session+begin
# 使用 with 语句开启一个 Session 会话
with Session(engine) as session:
    # 在 Session 会话下，进入一个自动事务块
    with session.begin():
        # 向会话中添加对象 user，此时为待提交状态
        session.add(user)

# 自动提交写法二：sessionmaker + begin
# 从 sqlalchemy.orm 导入 sessionmaker 工厂函数
from sqlalchemy.orm import sessionmaker
# 创建 Session 工厂，指定绑定的引擎
SessionFactory = sessionmaker(bind=engine)
# 使用工厂的 begin() 方法自动管理事务与 Session 生命周期
with SessionFactory.begin() as session:
    # 添加 user 对象到会话，退出 with 时自动提交
    session.add(user)

5. 定义数据模型 #

• 每个 Model 类 对应一张表；带 Mapped[...] 注解的属性对应列。
• 模型需：class X(Base)、__tablename__、Mapped + mapped_column。
• 建表：Base.metadata.create_all(engine)。

5.1 基本模型定义 #

# 从 datetime 导入 datetime
from datetime import datetime
# 从 sqlalchemy 导入 create_engine, String, Boolean, DateTime
from sqlalchemy import create_engine, String, Boolean, DateTime
# 从 sqlalchemy.orm 导入 DeclarativeBase, Mapped, mapped_column
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义 Product 模型，继承 Base
class Product(Base):
# 指定映射的数据库表名
    __tablename__ = "products"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列定义（非空等约束见参数）
    name: Mapped[str] = mapped_column(String(100), nullable=False)
# 价格字段，浮点数
    price: Mapped[float]
# 列字段定义
    is_available: Mapped[bool] = mapped_column(default=True)
# 列字段定义
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<Product(id={self.id}, name='{self.name}', price={self.price})>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据模型在数据库中创建表
Base.metadata.create_all(engine)

5.2 常用列类型 #

• Integer / Float / Boolean：数值与布尔
• String(n)：定长字符串；Text：长文本
• DateTime / Date：日期时间 / 仅日期

# 从 datetime 导入 datetime, date
from datetime import datetime, date
# 从 sqlalchemy 导入 String, Integer, Float, Boolean, DateTime, Text, Date
from sqlalchemy import String, Integer, Float, Boolean, DateTime, Text, Date
# 从 sqlalchemy.orm 导入 Mapped, mapped_column
from sqlalchemy.orm import DeclarativeBase,Mapped, mapped_column
# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass
# 定义 Example 模型，继承 Base
class Example(Base):
# 指定映射的数据库表名
    __tablename__ = "examples"
# 主键列
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
# 列定义（非空等约束见参数）
    title: Mapped[str] = mapped_column(String(255), nullable=False)
# 列字段定义
    price: Mapped[float] = mapped_column(Float)
# 列字段定义
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
# 列字段定义
    created_at: Mapped[datetime] = mapped_column(DateTime)
# 列字段定义
    content: Mapped[str] = mapped_column(Text)
# 列字段定义
    birthday: Mapped[date] = mapped_column(Date)

5.3 列参数 #

参数	作用
primary_key=True	主键
nullable=False	非空
unique=True	唯一
default=值	Python 侧默认值
index=True	建索引
server_default=text("...")	数据库侧默认值

# 从 datetime 导入 datetime
from datetime import datetime
# 从 sqlalchemy 导入 String, Integer, Float, Boolean, DateTime, text
from sqlalchemy import String, Integer, Float, Boolean, DateTime, text
# 从 sqlalchemy.orm 导入 Mapped, mapped_column
from sqlalchemy.orm import DeclarativeBase,Mapped, mapped_column
# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass
# 定义 Product 模型，继承 Base
class Product(Base):
# 指定映射的数据库表名
    __tablename__ = "products"
# 主键列
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
# 列定义（非空等约束见参数）
    name: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
# 列定义（非空等约束见参数）
    price: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
# 列字段定义
    in_stock: Mapped[bool] = mapped_column(Boolean, default=True)
# 列字段定义
    created_at: Mapped[datetime] = mapped_column(
# DateTime 列，数据库默认当前时间
        DateTime, server_default=text("CURRENT_TIMESTAMP")
    )

6. 基本 CRUD 操作 #

• CRUD 在 Session 中执行；须 commit() 或使用 session.begin() 才会写入数据库。
• echo=True 时控制台会打印实际 SQL，便于调试。

6.1 创建（Create） #

• session.add(obj) 添加单条；session.add_all(list) 批量添加。
• commit() 前主键通常为 None；提交后数据库生成 id 并回填到对象。

# 从 sqlalchemy 导入 String, Integer, create_engine
from sqlalchemy import String, Integer, create_engine
# 从 sqlalchemy.orm 导入 DeclarativeBase, Session, Mapped, mapped_column
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义 User 模型，继承 Base
class User(Base):
# 指定映射的数据库表名
    __tablename__ = "users"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列字段定义
    name: Mapped[str] = mapped_column(String(50))
# 年龄字段，整型
    age: Mapped[int]
    city: Mapped[str] = mapped_column(String(50))

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据模型在数据库中创建表
Base.metadata.create_all(engine)

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 将对象加入会话（待提交）
    session.add(User(name="张三", age=25))
# 提交事务，写入数据库
    session.commit()

# 批量将对象加入会话
    session.add_all([
# 创建模型实例
        User(name="李四", age=30),
# 创建模型实例
        User(name="王五", age=28),
# 创建模型实例
        User(name="赵六", age=35),
    ])
# 提交事务，写入数据库
    session.commit()

6.2 读取（Read） #

写法	说明
session.scalars(select(User)).all()	查全部，返回 User 列表
.first()	首条或 None
session.get(User, 1)	按主键查
.where(User.age < 30)	条件过滤
.order_by(desc(User.age)).limit(2)	排序与分页
session.execute(select(func.count(User.id))).scalar_one()	聚合计数

• 查 ORM 实体 用 session.scalars(select(User))。
• 查 多列 / 聚合 用 session.execute(select(...))，结果为 Row 元组。

# 从 sqlalchemy 导入 select, desc, func
from sqlalchemy import select, desc, func

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行查询并取全部结果
    all_users = session.scalars(select(User)).all()
# 执行查询并取第一条结果
    first = session.scalars(select(User)).first()
# 执行：by_id = session.get(User, 1)
    by_id = session.get(User, 1)
# 执行查询并取第一条结果
    zhang = session.scalars(select(User).where(User.name == "张三")).first()
# 执行查询并取全部结果
    young = session.scalars(select(User).where(User.age < 30)).all()
# 执行查询并取全部结果
    top2 = session.scalars(select(User).order_by(desc(User.age)).limit(2)).all()
# 执行：count = session.execute(select(func.count(User.id))).scalar_one()
    count = session.execute(select(func.count(User.id))).scalar_one()

scalars 是 SQLAlchemy 查询 API 中 Session 的一个方法，用于执行 select 查询并返回「单表 ORM 实体」的对象结果流（iterator）。

• 用法：session.scalars(select(User))
• 它比 session.execute 方便，直接生成 ORM 实例列表，无需再 row[0] 取值。
• 适用于只查一个 ORM 模型（单表）的场景，返回的就是 User 实例。
• 调用 .all() 得到结果列表，.first() 得到第一条或 None。

示例：

all_users = session.scalars(select(User)).all()  # 得到 User 实例列表
first_user = session.scalars(select(User)).first()  # 得到第一个 User 或 None

而 session.execute(select(...)) 返回的是 SQL 查询行 Row 对象（元组），通常用于多列/聚合等复杂查询，需 row[0] 取值。

# scalars vs execute：ORM 查询结果取值方式对比
# 导入需要的 SQLAlchemy 相关模块
from sqlalchemy import Integer, String, create_engine, func, select, text
# 导入 ORM 基础类、类型、Session、映射列等
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

# 定义基础类，所有 ORM 实体都需要继承这个 Base
class Base(DeclarativeBase):
    pass

# 定义 User 表的 ORM 映射
class User(Base):
    # 设置表名为 users
    __tablename__ = "users"

    # 定义 id 列，为主键，整型
    id: Mapped[int] = mapped_column(primary_key=True)
    # 定义 name 列，字符串类型，最大长度 50
    name: Mapped[str] = mapped_column(String(50))
    # 定义 age 列，整型
    age: Mapped[int]

    # 定义对象的字符串表示，方便输出和调试
    def __repr__(self) -> str:
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

# 创建数据库连接引擎
engine = create_engine(
    # 数据库连接 URL（MySQL+pymysql 驱动）
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    # 开启 SQL 输出
    echo=True,
    # 开启预连接测试
    pool_pre_ping=True,
)

# 使用连接引擎设置/初始化数据库
with engine.begin() as conn:
    # 先关闭外键约束，方便表删除
    conn.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
    # 如果有 addresses 表则删除
    conn.execute(text("DROP TABLE IF EXISTS addresses"))
    # 如果有 users 表则删除
    conn.execute(text("DROP TABLE IF EXISTS users"))
    # 恢复外键约束
    conn.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
    # 根据 ORM 定义创建所有表
    Base.metadata.create_all(conn)

# 向 users 表插入三条用户数据
with Session(engine) as session:
    # 添加多个 User 实例到会话
    session.add_all([
        User(name="张三", age=25),
        User(name="李四", age=30),
        User(name="王五", age=28),
    ])
    # 提交到数据库
    session.commit()

# 演示查询不同方式获取 ORM 结果
with Session(engine) as session:
    # 1. 推荐用 scalars 直接返回 ORM 实例列表
    users = session.scalars(select(User)).all()
    # 输出查询到的所有用户
    print(users)

    # 2. 用 execute 时，结果为 Row，需要 row[0] 取出 ORM 实例
    rows = session.execute(select(User)).all()
    # 用列表推导获得 ORM 实例集合
    users2 = [row[0] for row in rows]
    # 输出另一种方式得到的用户
    print(users2)

    # 聚合/多列查询时建议用 execute
    count = session.execute(select(func.count(User.id))).scalar_one()
    # 输出用户总数
    print(count)

简要说明：

• session.scalars(select(User)) 推荐用于查单表 ORM 实体，直接得到对象，无需解包。
• session.execute(select(User)) 返回 Row，需 row[0] 取 ORM 实例。
• 多列/聚合必须用 execute，返回 Row 或标量结果。

6.3 更新（Update） #

• 改单条：查出对象 → 改属性 → commit()。
• 批量改：session.execute(update(User).where(...).values(...))，不加载对象到内存。

# 从 sqlalchemy 导入 select, update
from sqlalchemy import select, update

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行查询并取第一条结果
    user = session.scalars(select(User).where(User.name == "张三")).first()
# 执行：if user:
    if user:
# 执行：user.age = 26
        user.age = 26
# 提交事务，写入数据库
        session.commit()

# 执行批量 UPDATE 语句
    session.execute(update(User).where(User.age > 30).values(age=31))
# 提交事务，写入数据库
    session.commit()

6.4 删除（Delete） #

• 删单条：session.delete(obj) + commit()。
• 批量删：session.execute(delete(User).where(...))，用 .rowcount 获取影响行数。

# 从 sqlalchemy 导入 select, delete
from sqlalchemy import select, delete

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行查询并取第一条结果
    user = session.scalars(select(User).where(User.name == "张三")).first()
# 执行：if user:
    if user:
# 从会话中标记删除对象
        session.delete(user)
# 提交事务，写入数据库
        session.commit()

# 获取 SQL 影响行数
    deleted = session.execute(delete(User).where(User.age < 25)).rowcount
# 提交事务，写入数据库
    session.commit()
# 打印输出
    print(f"批量删除 {deleted} 条")

7. 一对多和多对一 #

• 一个 User 可有多个 Address；每个 Address 只属于一个 User。
• 实现：ForeignKey + relationship() + back_populates。
• 写入时给 user.addresses = [...]，一次 commit() 即可连带保存子表。

关键词	说明
ForeignKey	声明外键
relationship()	定义关系属性
back_populates	双向访问
cascade	级联删除等（如 all, delete-orphan）

# 从 sqlalchemy 导入 create_engine, String, Integer, ForeignKey, select
from sqlalchemy import create_engine, String, Integer, ForeignKey, select
# 从 sqlalchemy.orm 导入 DeclarativeBase, Session, Mapped, mapped_column, relationship
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column, relationship

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义 User 模型，继承 Base
class User(Base):
# 指定映射的数据库表名
    __tablename__ = "users"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列字段定义
    name: Mapped[str] = mapped_column(String(50))
# 年龄字段，整型
    age: Mapped[int]
# 字段定义：addresses: Mapped[list["Address"]] = relationship(
    addresses: Mapped[list["Address"]] = relationship(
# 执行："Address", back_populates="user", cascade="all, delete-orphan"
        "Address", back_populates="user", cascade="all, delete-orphan"
    )

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<User(id={self.id}, name='{self.name}')>"

# 定义 Address 模型，继承 Base
class Address(Base):
# 指定映射的数据库表名
    __tablename__ = "addresses"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 外键列
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
# 列字段定义
    email: Mapped[str] = mapped_column(String(100))
# 字段定义：user: Mapped[User] = relationship("User", back_populates="addresses")
    user: Mapped[User] = relationship("User", back_populates="addresses")

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<Address(id={self.id}, email='{self.email}')>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据模型在数据库中创建表
Base.metadata.create_all(engine)

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行：user = User(name="赵六", age=35, addresses=[
    user = User(name="赵六", age=35, addresses=[
# 创建模型实例
        Address(email="zhaoliu@example.com"),
# 创建模型实例
        Address(email="zl@company.com")
    ])
# 将对象加入会话（待提交）
    session.add(user)
# 提交事务，写入数据库
    session.commit()
# 打印输出
    print("创建用户和地址成功")

# 执行查询并取第一条结果
    user = session.scalars(select(User).where(User.name == "赵六")).first()
# 打印输出
    print(f"\n用户：{user.name}")
# 打印输出
    print("地址列表：")
# 执行：for address in user.addresses:
    for address in user.addresses:
# 打印输出
        print(f"  - {address.email}")

# 通过地址查用户
# 执行查询并取第一条结果
    address = session.scalars(select(Address).where(Address.email == "zhaoliu@example.com")).first()
# 打印输出
    print(f"\n地址：{address.email}")
# 打印输出
    print(f"所属用户：{address.user.name}")

• cascade 是 relationship 的参数，用于控制父对象与子对象在进行操作（如删除、保存）时的联动行为。
  • 例如：cascade="all, delete-orphan" 表示所有操作都级联，并且删除“孤儿”对象（即从父对象关系中移除时也会一并删除）。
  • 示例：

    addresses = relationship(
        "Address",
        back_populates="user",
        cascade="all, delete-orphan"
    )

  • 这样设置后，如果删除一个用户，对应的所有地址也会自动被删除，不需手动删除关联的 Address 实例。

8. 多对多关系 #

• 例：学生 ↔ 课程，需中间 关联表（Table，不映射为 ORM 类）。
• relationship(..., secondary=关联表) 声明多对多；back_populates 双向访问。

关键词	说明
Table	中间关联表
secondary	指定关联表
relationship()	多对多关系

# 从 sqlalchemy 导入 create_engine, String, Integer, Table, ForeignKey, select, Column
from sqlalchemy import create_engine, String, Integer, Table, ForeignKey, select, Column
# 从 sqlalchemy.orm 导入 DeclarativeBase, Session, Mapped, mapped_column, relationship
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column, relationship

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义多对多中间关联表（非 ORM 类）
student_course = Table(
# 表名为student_course
# 中间表表名
    "student_course",
# 绑定到 Base 的元数据
    Base.metadata,
# 定义中间表的一列（含外键与主键）
    Column("student_id", Integer, ForeignKey("students.id"), primary_key=True),
# 定义中间表的一列（含外键与主键）
    Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True),
)

# 定义 Student 模型，继承 Base
class Student(Base):
# 指定映射的数据库表名
    __tablename__ = "students"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列字段定义
    name: Mapped[str] = mapped_column(String(50))
# 与Course模型建立多对多关系，通过student_course中间表，反向引用为students
# 字段定义：courses: Mapped[list["Course"]] = relationship(
    courses: Mapped[list["Course"]] = relationship(
# 执行："Course", secondary=student_course, back_populates="students"
        "Course", secondary=student_course, back_populates="students"
    )

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<Student(id={self.id}, name='{self.name}')>"

# 定义 Course 模型，继承 Base
class Course(Base):
# 指定映射的数据库表名
    __tablename__ = "courses"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列字段定义
    title: Mapped[str] = mapped_column(String(100))
# 与Student模型建立多对多关系，通过student_course中间表，反向引用为courses
# 字段定义：students: Mapped[list["Student"]] = relationship(
    students: Mapped[list["Student"]] = relationship(
# 执行："Student", secondary=student_course, back_populates="courses"
        "Student", secondary=student_course, back_populates="courses"
    )

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<Course(id={self.id}, title='{self.title}')>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据模型在数据库中创建表
Base.metadata.create_all(engine)

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行：course1 = Course(title="高等数学")
    course1 = Course(title="高等数学")
# 执行：course2 = Course(title="计算机基础")
    course2 = Course(title="计算机基础")
# 执行：student1 = Student(name="小明", courses=[course1, course2])
    student1 = Student(name="小明", courses=[course1, course2])
# 执行：student2 = Student(name="小红", courses=[course2])
    student2 = Student(name="小红", courses=[course2])
# 批量将对象加入会话
    session.add_all([student1, student2])
# 提交事务，写入数据库
    session.commit()
# 打印输出
    print("创建学生和课程成功")

# 执行查询并取第一条结果
    stu = session.scalars(select(Student).where(Student.name=="小明")).first()
# 打印输出
    print(f"\n学生：{stu.name}")
# 打印输出
    print("所选课程：")
# 执行：for c in stu.courses:
    for c in stu.courses:
# 打印输出
        print(f"  - {c.title}")

# 执行查询并取第一条结果
    cour = session.scalars(select(Course).where(Course.title=="计算机基础")).first()
# 打印输出
    print(f"\n课程：{cour.title}")
# 打印输出
    print("选修学生：")
# 执行：for s in cour.students:
    for s in cour.students:
# 打印输出
        print(f"  - {s.name}")

9. 复杂条件与聚合 #

9.1 复杂条件查询 #

• 多条件：and_(...)、or_(...)
• 模糊：User.name.like("张%")
• 列表：User.city.in_(["北京", "上海"])
• 区间：User.age.between(28, 32)

# 从 sqlalchemy 导入 create_engine, String, and_, or_, select
from sqlalchemy import create_engine, String, and_, or_, select
# 从 sqlalchemy.orm 导入 DeclarativeBase, Session, Mapped, mapped_column
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义 User 模型，继承 Base
class User(Base):
# 指定映射的数据库表名
    __tablename__ = "users"
# id字段，主键，自增长
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# name字段，最大长度50
# 列字段定义
    name: Mapped[str] = mapped_column(String(50))
# age字段
# 年龄字段，整型
    age: Mapped[int]
# city字段，最大长度50
# 列字段定义
    city: Mapped[str] = mapped_column(String(50))

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<User(id={self.id}, name='{self.name}', age={self.age}, city='{self.city}')>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据ORM模型生成数据库表
# 根据模型在数据库中创建表
Base.metadata.create_all(engine)

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 构建用户数据列表
# 查询或组装用户数据
    users = [
# 创建模型实例
        User(name="张三", age=25, city="北京"),
# 创建模型实例
        User(name="李四", age=30, city="上海"),
# 创建模型实例
        User(name="王五", age=28, city="北京"),
# 创建模型实例
        User(name="赵六", age=35, city="广州"),
# 执行：]
    ]
# 批量将对象加入会话
    session.add_all(users)
# 提交事务，写入数据库
    session.commit()

# 执行查询并取全部结果
    result1 = session.scalars(select(User).where(and_(User.age > 25, User.city == "北京"))).all()
# 打印输出
    print("年龄大于25且城市为北京的用户：")
# 执行：for user in result1:
    for user in result1:
# 打印输出
        print(user)

# 执行查询并取全部结果
    result2 = session.scalars(select(User).where(or_(User.age < 25, User.city == "上海"))).all()
# 打印输出
    print("\n年龄小于25或城市为上海的用户：")
# 执行：for user in result2:
    for user in result2:
# 打印输出
        print(user)

# 执行查询并取全部结果
    result3 = session.scalars(select(User).where(User.name.like("张%"))).all()
# 打印输出
    print("\n姓名以'张'开头的用户：")
# 执行：for user in result3:
    for user in result3:
# 打印输出
        print(user)

# 执行查询并取全部结果
    result4 = session.scalars(select(User).where(User.city.in_(["北京", "上海"]))).all()
# 打印输出
    print("\n城市为北京或上海的用户：")
# 执行：for user in result4:
    for user in result4:
# 打印输出
        print(user)

# 执行查询并取全部结果
    result5 = session.scalars(select(User).where(User.age.between(28, 32))).all()
# 打印输出
    print("\n年龄在28到32之间的用户：")
# 执行：for user in result5:
    for user in result5:
# 打印输出
        print(user)

9.2 聚合查询 #

• func.count / avg / max / min / sum 做统计。
• group_by(User.city) 分组聚合。

# 从 sqlalchemy 导入 create_engine, String, Integer, select, func
from sqlalchemy import create_engine, String, Integer, select, func
# 从 sqlalchemy.orm 导入 DeclarativeBase, Session, Mapped, mapped_column
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义 User 模型，继承 Base
class User(Base):
# 指定映射的数据库表名
    __tablename__ = "users"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列字段定义
    name: Mapped[str] = mapped_column(String(50))
# 年龄字段，整型
    age: Mapped[int]
# 列字段定义
    city: Mapped[str] = mapped_column(String(50))

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据ORM模型创建数据表（如果不存在则创建）
# 根据模型在数据库中创建表
Base.metadata.create_all(engine)

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行：avg_age = session.execute(select(func.avg(User.age))).scalar_one()
    avg_age = session.execute(select(func.avg(User.age))).scalar_one()
# 打印输出
    print(f"平均年龄：{avg_age:.2f}")

# 执行：max_age = session.execute(select(func.max(User.age))).scalar_one()
    max_age = session.execute(select(func.max(User.age))).scalar_one()
# 打印输出
    print(f"最大年龄：{max_age}")

# 执行：min_age = session.execute(select(func.min(User.age))).scalar_one()
    min_age = session.execute(select(func.min(User.age))).scalar_one()
# 打印输出
    print(f"最小年龄：{min_age}")

# 执行：total_age = session.execute(select(func.sum(User.age))).scalar_one()
    total_age = session.execute(select(func.sum(User.age))).scalar_one()
# 打印输出
    print(f"年龄总和：{total_age}")

# 执行：user_count = session.execute(select(func.count(User.id))).scalar_one()
    user_count = session.execute(select(func.count(User.id))).scalar_one()
# 打印输出
    print(f"用户总数：{user_count}")

# 按城市对用户进行分组统计，每个城市对应用户数量和平均年龄
# 执行：city_stats = session.execute(
    city_stats = session.execute(
# 执行：select(
        select(
# 执行：User.city,
            User.city,
# 执行：func.count(User.id).label("count"),
            func.count(User.id).label("count"),
# 执行：func.avg(User.age).label("avg_age")
            func.avg(User.age).label("avg_age")
# 执行：).group_by(User.city)
        ).group_by(User.city)
# 执行：).all()
    ).all()
# 打印输出
    print("\n按城市分组统计：")
# 执行：for city, count, avg_age in city_stats:
    for city, count, avg_age in city_stats:
# 打印输出
        print(f"  {city}: {count}人, 平均年龄{avg_age:.2f}")

10. 事务管理 #

事务：一组要么全成功、要么全失败的数据库操作。

• with Session(engine) as session: — 管理会话关闭；不会自动 commit，需手动 session.commit()，或在内部使用 session.begin()
• with session.begin(): 或 with SessionFactory.begin() as session: — 块正常结束时自动 commit，异常时 rollback

10.1 基本事务操作 #

# 从 sqlalchemy 导入 create_engine, String
from sqlalchemy import create_engine, String
# 从 sqlalchemy.orm 导入 DeclarativeBase, Session, Mapped, mapped_column
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义 User 模型，继承 Base
class User(Base):
# 指定映射的数据库表名
    __tablename__ = "users"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列字段定义
    name: Mapped[str] = mapped_column(String(50))
# 年龄字段，整型
    age: Mapped[int]

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据模型类创建所有数据表
# 根据模型在数据库中创建表
Base.metadata.create_all(engine)

# 捕获数据库操作异常
try:
# 使用上下文管理器开启 Session
    with Session(engine) as session:
# 执行：user1 = User(name="用户1", age=20)
        user1 = User(name="用户1", age=20)
# 执行：user2 = User(name="用户2", age=25)
        user2 = User(name="用户2", age=25)
# 批量将对象加入会话
        session.add_all([user1, user2])
# 提交事务，写入数据库
        session.commit()
# 打印输出
        print("事务提交成功")
# 处理异常并输出错误信息
except Exception as e:
# 打印输出
    print(f"发生错误：{e}")

说明：

• with Session(engine) as session: 退出时会关闭会话；若事务未提交，会自动 rollback
• 使用手动 commit() 时，应用 try-except 捕获异常并视情况 rollback

10.2 自动提交事务（session.begin()） #

无需自己封装事务类，SQLAlchemy提供 session.begin() 上下文，在块正常结束时自动 commit：

# 从 sqlalchemy 导入 create_engine, String
from sqlalchemy import create_engine, String
# 从 sqlalchemy.orm 导入 DeclarativeBase, Session, sessionmaker, Mapped, mapped_column
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker, Mapped, mapped_column

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义 User 模型，继承 Base
class User(Base):
# 指定映射的数据库表名
    __tablename__ = "users"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列字段定义
    name: Mapped[str] = mapped_column(String(50))
# 年龄字段，整型
    age: Mapped[int]

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据模型类创建所有数据表
# 根据模型在数据库中创建表
Base.metadata.create_all(engine)

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 开启事务块，正常退出时自动 commit
    with session.begin():
# 执行：user1 = User(name="用户1", age=20)
        user1 = User(name="用户1", age=20)
# 执行：user2 = User(name="用户2", age=25)
        user2 = User(name="用户2", age=25)
# 批量将对象加入会话
        session.add_all([user1, user2])

# 创建 sessionmaker 会话工厂
SessionFactory = sessionmaker(engine)
# 会话工厂 + 自动提交事务
with SessionFactory.begin() as session:
# 批量将对象加入会话
    session.add_all([
# 创建模型实例
        User(name="用户3", age=30),
# 创建模型实例
        User(name="用户4", age=35),
    ])
# 退出时自动 commit 并 close

11. 关系加载策略 #

ORM 默认对 relationship() 使用 lazy（懒加载）：访问 user.addresses 时才发 SQL 去查关联数据。单独查一个对象没问题；但若先查出 N 个用户，再逐个访问 addresses，就会额外产生 N 次查询，即常见的 N+1 查询问题。

SQLAlchemy 通过查询时的 加载选项（loader options） 提前指定关联数据如何加载，避免隐式懒加载。

策略	函数	适用场景
lazy	默认	偶尔访问关联；列表循环时易 N+1
selectinload	selectinload()	推荐：集合关系（一对多、多对多），两条 SQL
joinedload	joinedload()	多对一、一对一；集合关系需配合 unique()
raiseload	raiseload()	禁止懒加载，强制显式预加载

11.1 N+1 问题与 selectinload #

selectinload 先查主表，再用一条 WHERE id IN (...) 批量拉取关联集合，是列表场景最常用的写法。

# 从 sqlalchemy 导入 create_engine, String, Integer, ForeignKey, select
from sqlalchemy import create_engine, String, Integer, ForeignKey, select
# 从 sqlalchemy.orm 导入 (
from sqlalchemy.orm import (
# 字段定义：DeclarativeBase, Session, Mapped, mapped_column, relationship, selectinload,
    DeclarativeBase, Session, Mapped, mapped_column, relationship, selectinload,
)

# 定义 ORM 声明式基类
class Base(DeclarativeBase):
# 占位，无额外实现
    pass

# 定义 User 模型，继承 Base
class User(Base):
# 指定映射的数据库表名
    __tablename__ = "users"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 列字段定义
    name: Mapped[str] = mapped_column(String(50))
# 字段定义：addresses: Mapped[list["Address"]] = relationship(
    addresses: Mapped[list["Address"]] = relationship(
# 执行："Address", back_populates="user", cascade="all, delete-orphan"
        "Address", back_populates="user", cascade="all, delete-orphan"
    )

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<User(id={self.id}, name='{self.name}')>"

# 定义 Address 模型，继承 Base
class Address(Base):
# 指定映射的数据库表名
    __tablename__ = "addresses"
# 主键列
    id: Mapped[int] = mapped_column(primary_key=True)
# 外键列
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
# 列字段定义
    email: Mapped[str] = mapped_column(String(100))
# 字段定义：user: Mapped[User] = relationship("User", back_populates="addresses")
    user: Mapped[User] = relationship("User", back_populates="addresses")

# 定义对象的字符串表示，便于调试输出
    def __repr__(self):
# 返回格式化字符串
        return f"<Address(id={self.id}, email='{self.email}')>"

# 创建 MySQL 同步引擎
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,
    pool_pre_ping=True,
)
# 根据模型在数据库中创建表（先删后建，避免旧表结构残留导致插入失败）
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 开启事务块，正常退出时自动 commit
    with session.begin():
# 将对象加入会话（待提交）
        session.add(
# 创建模型实例
            User(
# 执行：name="赵六",
                name="赵六",
# 执行：addresses=[
                addresses=[
# 创建模型实例
                    Address(email="zhaoliu@example.com"),
# 创建模型实例
                    Address(email="zl@company.com"),
# 执行：],
                ],
            )
        )

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 懒加载：查 1 个用户 + 访问 addresses 时再查 1 次（列表场景会变成 N+1）
# 执行查询并取第一条结果
    user = session.scalars(select(User)).first()
# 执行：for addr in user.addresses:
    for addr in user.addresses:
# 打印输出
        print(addr.email)

# selectinload：预加载 addresses，循环时不再额外查询
# 执行 select 并返回标量结果流
    users = session.scalars(
# 为查询添加预加载选项
        select(User).options(selectinload(User.addresses))
# 执行：).all()
    ).all()
# 打印输出
    print("\nselectinload 预加载：")
# 执行：for user in users:
    for user in users:
# 执行：for addr in user.addresses:
        for addr in user.addresses:
# 打印输出
            print(f"  {user.name} -> {addr.email}")

11.2 joinedload #

joinedload 用 JOIN 一次 SQL 拉取主表与关联表。以下示例沿用 §11.1 的 User、Address 模型与 engine。访问 集合关系（一对多）时，JOIN 会产生重复行，结果需用 .unique() 去重：

# joinedload 示例：User ↔ Address 一对多，单条 SQL JOIN 预加载
# 导入 ForeignKey、String 类型、create_engine 和 select 查询方法
from sqlalchemy import ForeignKey, String, create_engine, select
# 导入 SQLAlchemy ORM 相关类和方法
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    joinedload,
    mapped_column,
    relationship,
)

# 定义基础类 Base，继承自 DeclarativeBase，用于声明模型
class Base(DeclarativeBase):
    # 空实现，用于继承
    pass

# 定义用户表 User，继承自 Base
class User(Base):
    # 指定表名为 users
    __tablename__ = "users"

    # 定义主键 id，类型为 int
    id: Mapped[int] = mapped_column(primary_key=True)
    # 定义 name 字段，类型为字符串，最大长度 50
    name: Mapped[str] = mapped_column(String(50))
    # 与 Address 建立一对多关系，反向引用 user，级联删除孤儿对象
    addresses: Mapped[list["Address"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )

# 定义地址表 Address，继承自 Base
class Address(Base):
    # 指定表名为 addresses
    __tablename__ = "addresses"

    # 定义主键 id，类型为 int
    id: Mapped[int] = mapped_column(primary_key=True)
    # 定义外键 user_id，关联 users.id
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    # 定义邮箱 email 字段，字符串类型，最大长度 100
    email: Mapped[str] = mapped_column(String(100))
    # 与 User 建立多对一关系，反向引用 addresses
    user: Mapped[User] = relationship(back_populates="addresses")

# 创建数据库引擎，连接本地 mysql，使用 pymysql 驱动
engine = create_engine(
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
    echo=True,           # 开启 SQL 输出
    pool_pre_ping=True,  # 预检测连接池有效性
)

# 删除所有已存在的表
Base.metadata.drop_all(engine)
# 创建所有表
Base.metadata.create_all(engine)

# 使用 Session 连接数据库，写入测试数据
with Session(engine) as session:
    # 开始一个数据库事务
    with session.begin():
        # 添加一个用户及其对应的两个地址
        session.add(
            User(
                name="赵六",
                addresses=[
                    Address(email="zhaoliu@example.com"),
                    Address(email="zl@company.com"),
                ],
            )
        )

# 再次打开 Session 进行查询操作
with Session(engine) as session:
    # joinedload：一条 SQL 把 users + addresses 一起查出，避免 N+1 查询问题
    users = (
        session.execute(
            select(User).options(joinedload(User.addresses)) # 使用 joinedload 联合预加载 addresses
        )
        .unique()         # 去除重复
        .scalars()        # 只取标量结果
        .all()            # 取所有结果
    )

    # 输出查询结果
    print("\njoinedload 预加载结果：")
    # 遍历每个用户
    for user in users:
        # 打印用户名及其所有地址的邮箱
        print(user.name, [a.email for a in user.addresses])

多对一方向（如 Address.user）通常不需要 unique()：

# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行 select 并返回标量结果流
    addresses = session.scalars(
# 为查询添加预加载选项
        select(Address).options(joinedload(Address.user))
# 执行：).all()
    ).all()
# 执行：for addr in addresses:
    for addr in addresses:
# 打印输出
        print(addr.email, addr.user.name)

11.3 选型建议 #

• 列表页批量展示 一对多 / 多对多 集合：优先 selectinload
• 详情页只需 多对一 父对象：joinedload 往往更直接

12. 异步 ORM #

同步 Session 会阻塞事件循环；在 FastAPI、aiohttp 等 asyncio 应用中，应使用 异步引擎 与 AsyncSession。

异步 ORM 需额外安装 aiomysql（§3.2 已说明）：

# 安装异步扩展与 MySQL 异步驱动
pip install "sqlalchemy[asyncio]" aiomysql

• MySQL 异步 URL：mysql+aiomysql://用户名:密码@主机:端口/数据库名?charset=utf8mb4

12.1 基本用法 #

与同步 API 对照：

同步	异步
create_engine()	create_async_engine()
Session(engine)	AsyncSession + async_sessionmaker
session.commit()	await session.commit() 或 async with session.begin()
session.scalars(select(...))	await session.scalars(select(...))
Base.metadata.create_all(engine)	await conn.run_sync(Base.metadata.create_all)

# // 示例标题：异步 SQLAlchemy + aiomysql
# 导入 asyncio 异步库
import asyncio

# 导入 SQLAlchemy 中常用类型和语句构建工具
from sqlalchemy import Integer, String, select, text
# 导入 SQLAlchemy 异步会话工厂和创建异步引擎的方法
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
# 导入声明基类、类型注解、字段定义方法等
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


# 定义 ORM 基类，所有模型继承自 Base
class Base(DeclarativeBase):
    # 空实现
    pass


# 定义 User 模型，继承自 Base
class User(Base):
    # 设置表名为 users
    __tablename__ = "users"

    # 定义主键 id，类型为 int，主键约束
    id: Mapped[int] = mapped_column(primary_key=True)
    # 定义 name 字段，字符串类型，最大 50 字符
    name: Mapped[str] = mapped_column(String(50))
    # 定义 age 字段，类型为 int
    age: Mapped[int]

    # 定义 __repr__ 方法，便于打印数据
    def __repr__(self) -> str:
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"


# 定义主异步方法
async def main() -> None:
    # 创建异步引擎，连接数据库，MySQL 使用 aiomysql 驱动
    engine = create_async_engine(
        "mysql+aiomysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
        echo=True,
        pool_pre_ping=True,
    )

    # 异步开启事务，执行数据库表结构初始化
    async with engine.begin() as conn:
        # 关闭外键检查（便于删除存在引用关系的表）
        await conn.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
        # 删除 addresses 表，如果存在
        await conn.execute(text("DROP TABLE IF EXISTS addresses"))
        # 删除 users 表，如果存在
        await conn.execute(text("DROP TABLE IF EXISTS users"))
        # 恢复外键检查
        await conn.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
        # 根据 Base 定义创建所有表
        await conn.run_sync(Base.metadata.create_all)

    # 创建异步Session工厂，设置提交后数据不失效
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    # 异步打开 session 会话，插入两条数据
    async with async_session() as session:
        # 会话级事务，自动处理提交/回滚
        async with session.begin():
            # 批量添加用户数据
            session.add_all([
                User(name="张三", age=25),
                User(name="李四", age=30),
            ])

    # 再次开启 Session，查询所有用户
    async with async_session() as session:
        # 执行 select 语句，按 id 升序查询 User，获取所有数据
        users = (await session.scalars(select(User).order_by(User.id))).all()
        # 打印 "所有用户：" 作为提示
        print("所有用户：")
        # 遍历并打印用户数据
        for user in users:
            print(user)

    # 释放数据库连接资源(销毁引擎)
    await engine.dispose()


# 检查代码入口，启动事件循环并运行主函数
if __name__ == "__main__":
    asyncio.run(main())

12.2 异步中的关系加载 #

异步环境下 禁止依赖懒加载（会触发隐式同步 IO）。查询带关联的数据时，务必使用 selectinload 等预加载（User 需已定义 addresses 关系，参见 §7）：

# 异步 selectinload 示例：User ↔ Address 一对多，两条 SQL 预加载集合
# 引入 asyncio 库，用于异步编程
import asyncio

# 从 sqlalchemy 导入常用的类型和方法
from sqlalchemy import ForeignKey, String, select, text
# 导入异步的 session 工厂和异步引擎创建方法
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
# 导入 ORM 相关类型与方法，包括基类、类型注解和关联配置
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
    selectinload,
)

# 定义基础类，所有 ORM 类都继承自该类
class Base(DeclarativeBase):
    # 占位，未添加其它属性或方法
    pass

# 定义用户表 ORM 映射类
class User(Base):
    # 设置数据库表名为 users
    __tablename__ = "users"

    # 定义主键 id，类型为 int
    id: Mapped[int] = mapped_column(primary_key=True)
    # 定义用户名，类型为字符串，最大长度 50
    name: Mapped[str] = mapped_column(String(50))
    # 关联 Address 实体，建立一对多关系，支持级联删除
    addresses: Mapped[list["Address"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )

# 定义地址表 ORM 映射类
class Address(Base):
    # 设置数据库表名为 addresses
    __tablename__ = "addresses"

    # 定义主键 id，类型为 int
    id: Mapped[int] = mapped_column(primary_key=True)
    # 外键 user_id，指向 users 表的 id 字段
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    # 邮箱字段，类型为字符串，最大长度 100
    email: Mapped[str] = mapped_column(String(100))
    # 关联 User 实体，建立多对一关系
    user: Mapped[User] = relationship(back_populates="addresses")

# 定义异步主函数
async def main() -> None:
    # 创建异步引擎，连接到本地 MySQL 数据库 aidb
    engine = create_async_engine(
        "mysql+aiomysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4",
        echo=True,
        pool_pre_ping=True,
    )

    # 使用异步上下文管理器开启数据库连接
    async with engine.begin() as conn:
        # 禁用外键约束，便于清空表
        await conn.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
        # 删除 addresses 表（如果存在）
        await conn.execute(text("DROP TABLE IF EXISTS addresses"))
        # 删除 users 表（如果存在）
        await conn.execute(text("DROP TABLE IF EXISTS users"))
        # 重新启用外键约束
        await conn.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
        # 根据 ORM 映射创建数据表结构
        await conn.run_sync(Base.metadata.create_all)

    # 创建异步 session 工厂，并设置事务提交后对象不会失效
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    # 创建异步 session，上下文管理数据库会话
    async with async_session() as session:
        # 在新的事务中进行数据写入
        async with session.begin():
            # 新增一个用户及其下属两个地址
            session.add(
                User(
                    name="赵六",
                    addresses=[
                        Address(email="zhaoliu@example.com"),
                        Address(email="zl@company.com"),
                    ],
                )
            )

    # 创建新的异步 session 查询数据
    async with async_session() as session:
        # selectinload：先查 users，再使用 IN 查询 addresses，避免 N+1 查询问题
        result = await session.scalars(
            select(User).options(selectinload(User.addresses))
        )
        # 取出所有的用户实体对象
        users = result.all()

        # 打印 selectinload 的预加载查询结果
        print("\nselectinload 预加载结果：")
        # 遍历每个用户对象
        for user in users:
            # 遍历当前用户下属的所有地址
            for addr in user.addresses:
                # 打印用户姓名和邮箱地址
                print(user.name, addr.email)

    # 关闭引擎，释放数据库连接资源
    await engine.dispose()

# 判断是否作为主模块运行
if __name__ == "__main__":
    # 使用 asyncio 运行异步主函数
    asyncio.run(main())

12.3 注意事项 #

• AsyncSession 中所有数据库操作都要 await
• 建表 / 反射等同步 API 用 await conn.run_sync(可调用对象)
• 在 async def 外不要混用同步 Session 与 AsyncSession
• 应用退出前调用 await engine.dispose() 释放连接池

13. 常见错误与最佳实践 #

13.1 错误 1：忘记提交事务 #

# 错误示例
# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行：user = User(name="张三", age=25)
    user = User(name="张三", age=25)
# 将对象加入会话（待提交）
    session.add(user)

# 正确示例（手动 commit）
# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行：user = User(name="张三", age=25)
    user = User(name="张三", age=25)
# 将对象加入会话（待提交）
    session.add(user)
# 提交事务，写入数据库
    session.commit()

# 也可使用 begin() 自动提交
# 使用上下文管理器开启 Session
with Session(engine) as session:
# 开启事务块，正常退出时自动 commit
    with session.begin():
# 将对象加入会话（待提交）
        session.add(User(name="张三", age=25))

13.2 错误 2：忘记关闭会话 #

# 手动创建会话（容易忘记关闭）
# 手动创建 Session（不推荐，易忘记关闭）
session = Session(engine)
# 执行：user = User(name="张三", age=25)
user = User(name="张三", age=25)
# 将对象加入会话（待提交）
session.add(user)
# 提交事务，写入数据库
session.commit()
# 忘记调用close()，会话未关闭

# 推荐写法：使用with语句自动管理会话的开启与关闭
# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行：user = User(name="张三", age=25)
    user = User(name="张三", age=25)
# 将对象加入会话（待提交）
    session.add(user)
# 提交事务，写入数据库
    session.commit()

13.3 错误 3：循环中频繁提交 #

# 错误：循环内反复 commit
# 使用上下文管理器开启 Session
with Session(engine) as session:
# 执行：for i in range(100):
    for i in range(100):
# 将对象加入会话（待提交）
        session.add(User(name=f"用户{i}", age=20 + i))
# 提交事务，写入数据库
        session.commit()

# 正确：add_all + 一次 commit
# 使用上下文管理器开启 Session
with Session(engine) as session:
# 批量将对象加入会话
    session.add_all([User(name=f"用户{i}", age=20 + i) for i in range(100)])
# 提交事务，写入数据库
    session.commit()

13.4 错误 4：N+1 查询 #

# 需 from sqlalchemy.orm import selectinload
# 错误：查出 N 个用户后，循环访问 addresses 会各触发一次 SQL
# 执行查询并取全部结果
users = session.scalars(select(User)).all()
# 执行：for user in users:
for user in users:
# 打印输出
    print(user.addresses)  # 每个 user 多 1 条 SELECT

# 正确：用 selectinload 预加载关联数据（详见 §11）
# 执行 select 并返回标量结果流
users = session.scalars(
# 为查询添加预加载选项
    select(User).options(selectinload(User.addresses))
# 执行：).all()
).all()
# 执行：for user in users:
for user in users:
# 打印输出
    print(user.addresses)  # 不再额外查询

13.5 最佳实践 #

1. 用 with Session(engine) 管理会话，避免忘记 close()
2. 需要自动提交时，用 session.begin() 或 sessionmaker.begin()，不要自己封装事务类
3. 查询用 select() + session.scalars()，不要用已废弃的 session.query()
4. 列表查询带关联数据时，用 selectinload / joinedload 避免 N+1
5. 异步场景用 AsyncSession，并显式预加载关联，不要依赖懒加载
6. 批量操作推荐 add_all()，避免循环中多次 commit()
7. 异常处理用 try-except，必要时 session.rollback()
8. 模型定义分离，生产代码将模型单独放文件
9. 类型提示（Mapped、mapped_column）让代码更可维护

14. Flask-SQLAlchemy #

• Flask-SQLAlchemy 是 Flask 官方生态的数据库扩展，在 SQLAlchemy 之上封装了常用默认值与 Web 场景辅助方法。
• 在 Flask 里通常改用 Flask-SQLAlchemy，少写 Engine / Session 样板代码。

14.1 与原生 SQLAlchemy 的对应关系 #

原生 SQLAlchemy	Flask-SQLAlchemy
create_engine()	app.config["SQLALCHEMY_DATABASE_URI"] + db.init_app(app)
DeclarativeBase / Base	db.Model（或 SQLAlchemy(model_class=Base)）
Session(engine)	db.session（绑定到 Flask 应用上下文）
select(User)	db.select(User)
Base.metadata.create_all(engine)	db.create_all()（需在 app.app_context() 内）
手动 session.get()	db.get_or_404(Model, pk)（查不到自动 404）

14.2 安装 #

# 安装 Flask、Flask-SQLAlchemy 与 MySQL 驱动
pip install flask flask-sqlalchemy pymysql

14.3 初始化与建表 #

# 从 flask 导入 Flask 应用类
from flask import Flask
# 从 flask_sqlalchemy 导入扩展入口 SQLAlchemy
from flask_sqlalchemy import SQLAlchemy
# 从 sqlalchemy.orm 导入 2.0 声明式基类（可选，用于类型提示）
from sqlalchemy.orm import DeclarativeBase

# 定义与教程一致的声明式基类
class Base(DeclarativeBase):
    pass

# 创建 Flask 应用实例
app = Flask(__name__)
# 配置 MySQL 连接 URL（与 §3.2 相同，按本地修改）
app.config["SQLALCHEMY_DATABASE_URI"] = (
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb?charset=utf8mb4"
)
# 可选：关闭对象修改追踪，节省内存（不需要旧版 Flask-SQLAlchemy 的信号时）
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

# 创建扩展对象，指定 model_class 以支持 Mapped 类型注解
db = SQLAlchemy(app, model_class=Base)

# 进入应用上下文（命令行脚本建表时必须）
with app.app_context():
    # 根据所有 db.Model 子类创建表
    db.create_all()

应用工厂写法（大型项目推荐）：

# 从 flask 导入 Flask 应用类
from flask import Flask
# 从 flask_sqlalchemy 导入扩展入口 SQLAlchemy
from flask_sqlalchemy import SQLAlchemy
# 从 sqlalchemy.orm 导入 2.0 声明式基类（可选，用于类型提示）
from sqlalchemy.orm import DeclarativeBase
# 定义与教程一致的声明式基类
class Base(DeclarativeBase):
    pass
# 先创建未绑定 app 的 db 对象
db = SQLAlchemy(model_class=Base)

# 应用工厂函数
def create_app():
    # 创建 Flask 实例
    app = Flask(__name__)
    # 设置数据库 URI
    app.config["SQLALCHEMY_DATABASE_URI"] = (
        "mysql+pymysql://root:root@127.0.0.1:3306/aidb2?charset=utf8mb4"
    )
    # 将 db 绑定到 app
    db.init_app(app)
    # 返回配置好的 app
    return app

14.4 定义模型 #

写法继承 db.Model 而不是自定义 Base：

# 从 flask 导入 Flask 应用类
from flask import Flask
# 从 flask_sqlalchemy 导入扩展入口 SQLAlchemy
from flask_sqlalchemy import SQLAlchemy
# 从 sqlalchemy.orm 导入 2.0 声明式基类（可选，用于类型提示）
from sqlalchemy.orm import DeclarativeBase
# 定义与教程一致的声明式基类
class Base(DeclarativeBase):
    pass
# 先创建未绑定 app 的 db 对象
db = SQLAlchemy(model_class=Base)
# 从 sqlalchemy 导入 String
from sqlalchemy import String
# 从 sqlalchemy.orm 导入 Mapped 与 mapped_column
from sqlalchemy.orm import Mapped, mapped_column

# 用户模型，继承 db.Model
class User(db.Model):
    # 映射到 users 表
    __tablename__ = "users"
    # 主键
    id: Mapped[int] = mapped_column(primary_key=True)
    # 用户名，唯一且非空
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    # 邮箱
    email: Mapped[str] = mapped_column(String(100), nullable=False)
    # 年龄
    age: Mapped[int]

    # 调试输出
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"
# 创建 Flask 应用实例
app = Flask(__name__)
# 配置 MySQL 连接 URL（与 §3.2 相同，按本地修改）
app.config["SQLALCHEMY_DATABASE_URI"] = (
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb2?charset=utf8mb4"
)
# 可选：关闭对象修改追踪，节省内存（不需要旧版 Flask-SQLAlchemy 的信号时）
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

# 创建扩展对象，指定 model_class 以支持 Mapped 类型注解
db = SQLAlchemy(app, model_class=Base)

# 进入应用上下文（命令行脚本建表时必须）
with app.app_context():
    # 根据所有 db.Model 子类创建表
    db.create_all()

14.5 在路由中 CRUD #

Flask 每个请求都在 应用上下文 中运行，可直接使用 db.session：

# 从 flask 导入请求、重定向、url_for
from flask import Flask,request, redirect, url_for
# 从 sqlalchemy 导入 select（也可只用 db.select）
from sqlalchemy import select
# 从 flask_sqlalchemy 导入扩展入口 SQLAlchemy
from flask_sqlalchemy import SQLAlchemy
# 从 sqlalchemy.orm 导入 2.0 声明式基类（可选，用于类型提示）
from sqlalchemy.orm import DeclarativeBase
# 定义与教程一致的声明式基类
class Base(DeclarativeBase):
    pass
# 先创建未绑定 app 的 db 对象
db = SQLAlchemy(model_class=Base)
# 从 sqlalchemy 导入 String
from sqlalchemy import String
# 从 sqlalchemy.orm 导入 Mapped 与 mapped_column
from sqlalchemy.orm import Mapped, mapped_column

# 用户模型，继承 db.Model
class User(db.Model):
    # 映射到 users 表
    __tablename__ = "users"
    # 主键
    id: Mapped[int] = mapped_column(primary_key=True)
    # 用户名，唯一且非空
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    # 邮箱
    email: Mapped[str] = mapped_column(String(100), nullable=False)
    # 年龄
    age: Mapped[int]

    # 调试输出
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"
# 创建 Flask 应用实例
app = Flask(__name__)
# 配置 MySQL 连接 URL（与 §3.2 相同，按本地修改）
app.config["SQLALCHEMY_DATABASE_URI"] = (
    "mysql+pymysql://root:root@127.0.0.1:3306/aidb2?charset=utf8mb4"
)
# 可选：关闭对象修改追踪，节省内存（不需要旧版 Flask-SQLAlchemy 的信号时）
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

# 创建扩展对象，指定 model_class 以支持 Mapped 类型注解
db = SQLAlchemy(app, model_class=Base)

# 用户列表：查询全部并按 id 排序
@app.route("/users")
def user_list():
    # 执行 select，scalars() 得到 User 对象迭代器
    users = db.session.scalars(
        db.select(User).order_by(User.id)
    ).all()
    # 渲染模板（此处省略模板文件）
    return {"users": [u.username for u in users]}

# 创建用户：POST 提交表单
@app.route("/users", methods=["POST"])
def user_create():
    # 从表单取字段
    user = User(
        username=request.form["username"],
        email=request.form["email"],
        age=int(request.form["age"]),
    )
    # 加入会话并提交
    db.session.add(user)
    db.session.commit()
    # 重定向到详情页
    return redirect(url_for("user_detail", id=user.id))

# 查看单个用户：按主键，不存在则 404
@app.route("/users/<int:id>")
def user_detail(id):
    # get_or_404 等价于 session.get + 404 处理
    user = db.get_or_404(User, id)
    return {"id": user.id, "username": user.username}

# 更新用户
@app.route("/users/<int:id>", methods=["PUT"])
def user_update(id):
    # 先查出对象
    user = db.get_or_404(User, id)
    # 修改属性
    user.email = request.form.get("email", user.email)
    # 提交（Flask-SQLAlchemy 也可依赖请求结束时的自动 commit，但显式更清晰）
    db.session.commit()
    return {"ok": True}

# 删除用户
@app.route("/users/<int:id>", methods=["DELETE"])
def user_delete(id):
    # 查出要删的对象
    user = db.get_or_404(User, id)
    # 标记删除
    db.session.delete(user)
    # 提交事务
    db.session.commit()
    # 返回列表页
    return redirect(url_for("user_list"))
# 进入应用上下文（命令行脚本建表时必须）
with app.app_context():
    # 根据所有 db.Model 子类创建表
    db.create_all()
#  
app.run(debug=True)    

14.6 查询辅助方法 #

# 按条件查唯一一条，找不到则返回 404
@app.route("/users/by-name/<name>")
def user_by_name(name):
    # one_or_404 接收 select 语句
    user = db.one_or_404(db.select(User).filter_by(username=name))
    return {"username": user.username}

# 分页：从 URL 参数 ?page=1&per_page=10 读取页码
@app.route("/users/page")
def user_page():
    # paginate 返回 Pagination 对象
    page = db.paginate(db.select(User).order_by(User.id))
    # page.items 为当前页数据，page.pages 为总页数
    return {
        "page": page.page,
        "pages": page.pages,
        "users": [u.username for u in page.items],
    }

14.7 注意事项 #

• 必须在应用上下文中操作数据库；脚本里用 with app.app_context():，路由里 Flask 自动提供。
• 请求结束时 Flask-SQLAlchemy 会自动 session.remove()，一般不必手动 close()；但 commit() 仍需显式调用（除非配置了自动提交钩子）。

14.8 API 速查（Flask-SQLAlchemy） #

操作	写法
配置数据库	app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://..."
初始化	db = SQLAlchemy(app) 或 db.init_app(app)
建表	with app.app_context(): db.create_all()
定义模型	class User(db.Model): ...
查询	db.session.scalars(db.select(User).where(...)).all()
按主键	db.session.get(User, 1) 或 db.get_or_404(User, 1)
插入	db.session.add(obj); db.session.commit()
删除	db.session.delete(obj); db.session.commit()
分页	db.paginate(db.select(User))

15. API 速查 #

操作	写法
建引擎	create_engine("mysql+pymysql://user:pass@host:3306/db?charset=utf8mb4")
建表	Base.metadata.create_all(engine)
开会话	with Session(engine) as session:
自动提交	with session.begin(): 或 SessionFactory.begin()
插入	session.add(obj) / add_all([...])
查询	session.scalars(select(Model).where(...)).all()
按主键	session.get(Model, pk)
更新	改属性 + commit()，或 update(Model).where(...).values(...)
删除	session.delete(obj)，或 delete(Model).where(...)
预加载	.options(selectinload(Model.relation))
异步引擎	create_async_engine("mysql+aiomysql://user:pass@host:3306/db?charset=utf8mb4")
异步查询	await session.scalars(select(Model))