SQLAlchemy bulk_update_mappings更新user_info表失败求助

表结构 #

UserInfo表的定义如下：

class UserInfo(Base):
    __tablename__ = 'user_info'

    id = Column(INTEGER, primary_key=True)
    user_code = Column(CHAR(16), nullable=False, comment='工号')
    goal_1 = Column(CHAR(16))
    goal_2 = Column(CHAR(16))
    prem_1 = Column(CHAR(16))
    prem_2 = Column(CHAR(16))

该表约有10000条数据，所有goal_*和prem_*字段均为NULL，需要用随机数据更新这些字段，测试阶段仅更新50行。

尝试的代码及问题 #

第一种代码（报错） #

def insert_prem_date(self, count=50):
    users = self.db.session.query(UserInfo).limit(count)

    user_codes = [i.user_code for i in users]
    print(user_codes)

    prem_list = []
    for _code in user_codes:
        prem_list.append(
            {"user_code": _code,
             "prem_1": str(random.randint(500, 500000)),
             "prem_2": str(random.randint(500, 500000)),
             "goal_1": str(random.randint(500, 500000)),
             "goal_2": str(random.randint(500, 500000))
             }
        )
    self.db.session.bulk_update_mappings(
        UserInfo,
        prem_list
    )

    self.db.session.commit()

执行后报错：找不到主键列用于标识要更新的行。已确认user_code在表中存在，但无法完成更新。

第二种代码（无报错但不更新） #

for _code in user_codes:
    prem_list.append(
        {UserInfo.user_code: _code,
         UserInfo.prem_1: str(random.randint(500, 500000)),
         UserInfo.prem_2: str(random.randint(500, 500000)),
         UserInfo.goal_1: str(random.randint(500, 500000)),
         UserInfo.goal_2: str(random.randint(500, 500000))
         }
    )

无报错，但数据库中的数据没有被更新。

bulk_update_mappings方法要求每个字典中必须包含主键字段来定位要更新的行，当前表主键是id，而非user_code，这是核心问题。下面提供两种可行的实现方式：

方式一：基于主键id批量更新 #

def insert_prem_date(self, count=50):
    # 查询时同时获取主键id和user_code
    users = self.db.session.query(UserInfo.id, UserInfo.user_code).limit(count).all()

    prem_list = []
    for user_id, _code in users:
        prem_list.append(
            {
                "id": user_id,  # 必须包含主键用于定位
                "prem_1": str(random.randint(500, 500000)),
                "prem_2": str(random.randint(500, 500000)),
                "goal_1": str(random.randint(500, 500000)),
                "goal_2": str(random.randint(500, 500000))
            }
        )
    self.db.session.bulk_update_mappings(UserInfo, prem_list)
    self.db.session.commit()

方式二：直接用update语句批量更新（更高效） #

如果不需要先查询数据，可直接通过SQLAlchemy的update语法批量更新随机50行：

from sqlalchemy import func

def insert_prem_date(self, count=50):
    # 获取随机的50个用户主键id
    random_user_ids = self.db.session.query(UserInfo.id)\
                        .order_by(func.random())\
                        .limit(count)\
                        .subquery()
    
    # 执行批量更新
    self.db.session.query(UserInfo)\
        .filter(UserInfo.id.in_(random_user_ids))\
        .update(
            {
                "prem_1": func.cast(func.floor(func.random() * 499501 + 500), CHAR),
                "prem_2": func.cast(func.floor(func.random() * 499501 + 500), CHAR),
                "goal_1": func.cast(func.floor(func.random() * 499501 + 500), CHAR),
                "goal_2": func.cast(func.floor(func.random() * 499501 + 500), CHAR)
            },
            synchronize_session=False  # 关闭会话同步，提升更新效率
        )
    self.db.session.commit()

失败原因说明 #

1. 第一种代码用user_code作为更新标识，但bulk_update_mappings默认仅通过主键定位行，因此报错找不到主键列。
2. 第二种代码用ORM字段对象作为字典键，不符合bulk_update_mappings的要求——该方法需要字典键为字符串形式的字段名，导致无法识别更新条件，最终没有数据被修改。

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