23. FastAPI 使用 SQLAlchemy 2 Async 更新与删除关系数据（单条与百万条）
更新删除首先要确定范围、关系加载和事务
沿用第 20～22 篇配置和模型。单条修改适合读取 ORM 对象后赋值，批量修改适合直接执行 UPDATE/DELETE。异步代码还要额外确认：访问的关系已经加载，所有数据库 I/O 都使用了 await。
所有删除示例只应对自己的练习批次执行。以下代码通过 ID 或 batch_tag 明确限定范围；真实接口还要校验当前用户的操作权限。
情况一：修改一个用户的三类关系
保存为 change_one.py：
import asyncio
from sqlalchemy import select
from sqlalchemy.orm import joinedload, selectinload
from database import SessionLocal, master_engine
from models import Post, Profile, Role, User
async def update_one(user_id: int, role_id: int) -> None:
async with SessionLocal.begin() as db:
# profile 和 roles 后面要直接访问，必须提前加载
user = await db.scalar(
select(User)
.options(
joinedload(User.profile),
selectinload(User.roles),
)
.where(User.id == user_id)
)
role = await db.get(Role, role_id)
if user is None or role is None:
raise ValueError("用户或角色不存在")
user.name = "修改后的名字"
if user.profile is None:
user.profile = Profile(bio="新资料")
else:
user.profile.bio = "更新现有资料"
post = await db.scalar(
select(Post)
.where(Post.user_id == user_id)
.order_by(Post.id)
.limit(1)
)
if post is not None:
post.title = "修改后的第一篇文章"
# 修改多对多关联行，不会删除旧的共享 Role 对象
user.roles = [role]
async def delete_one(user_id: int) -> bool:
async with SessionLocal.begin() as db:
user = await db.get(User, user_id)
if user is None:
return False
# AsyncSession.delete() 是异步方法
await db.delete(user)
return True
async def main() -> None:
try:
# 替换为第 22 篇返回的真实 ID，不要假设一定是 1
await update_one(user_id=1, role_id=1)
# 确认该用户可以删除后，再单独启用下一行
# deleted = await delete_one(user_id=1)
# print("是否删除：", deleted)
finally:
await master_engine.dispose()
if __name__ == "__main__":
asyncio.run(main())替换 ID 后执行：
uv run python change_one.py更新一对一时优先修改已有资料；一对多通过文章的 user_id 限定归属；多对多赋值改变关联行，不会删除旧角色。如果还要直接访问 user.posts，必须再添加 selectinload(User.posts)，或使用 await db.refresh(user, attribute_names=["posts"])，不能依赖同步 ORM 的隐式 lazy load。
单条 delete_one() 使用 await db.delete(user)。当前模型配置了 passive_deletes=True，示例依赖真实数据库外键的 ON DELETE CASCADE 清理资料、文章和关联行；共享角色本身保留。仅修改 Python 模型但没有同步迁移数据库约束，不能保证级联生效。
情况二：异步修改一百万用户及关联记录
下面是 change_many.py。目标应是第 22 篇导入的百万用户批次，role_id 是预先存在的目标角色。
from sqlalchemy import delete, select, update
from database import SessionLocal
from models import Post, Profile, Role, User, user_roles
async def update_batch(tag: str, role_id: int, batch_size: int = 2000) -> int:
if not tag or batch_size < 1:
raise ValueError("需要明确批次和正的批大小")
async with SessionLocal() as db:
if await db.get(Role, role_id) is None:
raise ValueError("目标角色不存在")
last_id, done = 0, 0
while True:
async with SessionLocal.begin() as db:
ids = list(
(
await db.scalars(
select(User.id)
.where(
User.batch_tag == tag,
User.id > last_id,
)
.order_by(User.id)
.limit(batch_size)
)
).all()
)
if not ids:
break
await db.execute(
update(User).where(User.id.in_(ids)).values(active=False),
execution_options={"synchronize_session": False},
)
await db.execute(
update(Profile)
.where(Profile.user_id.in_(ids))
.values(bio="批量更新后的资料"),
execution_options={"synchronize_session": False},
)
await db.execute(
update(Post)
.where(Post.user_id.in_(ids))
.values(title="批量更新后的文章"),
execution_options={"synchronize_session": False},
)
# 本例要求每人最终只保留指定角色，因此先清关联再添加
await db.execute(
delete(user_roles).where(user_roles.c.user_id.in_(ids))
)
await db.execute(
user_roles.insert(),
[{"user_id": user_id, "role_id": role_id} for user_id in ids],
)
last_id = ids[-1]
done += len(ids)
print(f"更新用户 {done}，游标 {last_id}", flush=True)
return done
async def delete_batch(tag: str, batch_size: int = 2000) -> int:
if not tag or batch_size < 1:
raise ValueError("需要明确批次和正的批大小")
last_id, done = 0, 0
while True:
async with SessionLocal.begin() as db:
ids = list(
(
await db.scalars(
select(User.id)
.where(
User.batch_tag == tag,
User.id > last_id,
)
.order_by(User.id)
.limit(batch_size)
)
).all()
)
if not ids:
break
# 批量 DELETE 依靠数据库 ON DELETE CASCADE 清理关联行
await db.execute(
delete(User).where(User.id.in_(ids)),
execution_options={"synchronize_session": False},
)
last_id = ids[-1]
done += len(ids)
print(f"删除用户 {done}，游标 {last_id}", flush=True)
return done另存 run_change.py，替换为真实批次和角色 ID：
import asyncio
from change_many import update_batch
from database import master_engine
async def main() -> None:
try:
done = await update_batch(
"替换为第22篇打印的batch_tag",
role_id=1,
)
print("实际更新用户数：", done)
finally:
await master_engine.dispose()
if __name__ == "__main__":
asyncio.run(main())执行：
uv run python run_change.py删除时在同类脚本中导入 delete_batch，核对批次后使用 await delete_batch(...)。不要同时执行更新和删除练习。若批次确有 1000000 用户，最终函数返回 1000000；若批次只有 1000，就只操作 1000。
为什么选主键游标，不选 OFFSET
删除第一批后，剩余行前移，再用 OFFSET 2000 会跳过部分未处理数据。主键游标通过 id > last_id 继续，不受前面删除影响。每批只加载少量整数 ID，避免把百万个 ORM 对象长期放在内存中。
批量 UPDATE/DELETE 不会自动加载每个 ORM 对象，也不会逐个触发 ORM 删除事件。synchronize_session=False 避免同步大量内存对象，适合本例每批新建且没有加载实体的 AsyncSession。如果同一个 Session 中已有相关对象，属性可能是旧值，应使用新 Session，或显式过期后 await db.refresh(...)。
并发、恢复与校验
批次执行期间应冻结目标数据或固定任务范围，避免并发写入不断改变处理集合。需要断点续跑时，把游标和本批操作一起持久化到数据库；只在打印日志中记录游标不等于可靠检查点。
大量修改可能带来长锁等待、死锁、redo/binlog 增长及复制延迟。适当缩小批次，遇到可重试的事务错误时回滚整批并有限重试。异步批次仍然按顺序执行，不要在同一个 AsyncSession 上并发 gather() 多个 UPDATE。
rowcount 的匹配数与实际变化数还受 MySQL 和驱动语义影响，不能单靠它证明业务完成。第 22 篇导入的百万任务被多个请求同时执行时，还应增加任务级锁或幂等控制。
练习：先用 1000 用户批次验证更新覆盖三种关系；删除后确认 User、Profile、Post、user_roles 中无该批关联残留，而 Role 仍存在。确认结果后再放大到百万规模。