SQLAlchemy一对多关系中，如何用ID数组更新User关联的Item？

我定义了一对多关系的SQLAlchemy模型User和Item：

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"), nullable=True)
    owner = relationship("User", back_populates="items")

数据库已有数据：

users:
id | email
1  | test@test.com

items:
id | owner_id | title
1  | 1        | aaa
2  | 1        | bbb
3  | NULL     | ccc

我编写了update_user函数用于更新用户信息，传入的data包含email和Item ID字符串数组：

def update_user(db: Session, user_id: int, data: dict):
    # data = {"email": "new@test.com", "items": ["1", "3"]}

    db_user = db.query(models.User).filter(models.User.id == user_id).first()

    db_user.email = data["email"]

    # 问题所在：直接赋值ID数组无效，需要Item对象而非ID
    db_user.items = data["items"]


    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

直接将ID数组赋值给db_user.items无法生效，且我不想先删除所有关联Item再重新添加（会导致已有Item的ID被不必要修改），请问SQLAlchemy有什么简便解决方案？

1. 批量查询目标Item对象 #

最直接的方式是根据传入的ID数组查询对应的Item对象，再赋值给db_user.items：

def update_user(db: Session, user_id: int, data: dict):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    db_user.email = data["email"]

    # 将字符串ID转为整数，查询对应的Item对象
    item_ids = [int(id_str) for id_str in data["items"]]
    db_user.items = db.query(models.Item).filter(models.Item.id.in_(item_ids)).all()

    db.commit()
    db.refresh(db_user)
    return db_user

SQLAlchemy会自动处理关联关系：把原来关联的Item（比如ID=2）的owner_id设为NULL，同时把目标Item（ID=1、3）的owner_id设为当前用户ID，无需手动删除再添加。

2. 使用Session.get批量获取对象 #

如果Item的ID是主键，用Session.get配合列表推导式更简洁，还能自动过滤不存在的ID：

item_ids = [int(id_str) for id_str in data["items"]]
# 过滤掉不存在的Item，避免关联无效对象
db_user.items = [item for item_id in item_ids if (item := db.get(models.Item, item_id))]

3. 增量更新（高效场景） #

如果用户关联的Item数量较多，全量替换效率偏低，可以对比现有关联和目标ID，只修改差异部分：

def update_user(db: Session, user_id: int, data: dict):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    db_user.email = data["email"]

    target_item_ids = set(int(id_str) for id_str in data["items"])
    current_item_ids = set(item.id for item in db_user.items)

    # 移除不在目标列表中的关联Item
    for item in db_user.items[:]:  # 遍历列表副本，避免迭代时修改原列表引发异常
        if item.id not in target_item_ids:
            db_user.items.remove(item)

    # 添加目标列表中未关联的Item
    missing_item_ids = target_item_ids - current_item_ids
    for item_id in missing_item_ids:
        if item := db.get(models.Item, item_id):
            db_user.items.append(item)

    db.commit()
    db.refresh(db_user)
    return db_user

这种方式仅对需要变更的Item操作，减少数据库交互，适合数据量大的场景。

内容的提问来源于stack exchange，提问作者weskyen