SQLAlchemy查询无主键表报row_id列无效错误如何解决

安全阿岳

2026-8-30

问题背景

需要从无主键的数据表中获取数据，为让表可被SQLAlchemy识别并完成ORM映射编写了对应代码，但始终无法正常查询表中数据，初始实现代码如下：

table = 'my_table'
db_tables = automap_base()
metadata = MetaData()
my_table = Table(table, db_tables.metadata, Column('row_id', Integer, primary_key=True), autoload=True, autoload_with=db.engine)
db_tables.prepare(db.engine, reflect=True)

data = db.session.query(db_tables.classes.my_table).filter(
    db_tables.classes.my_table.device_name.like('%uni%'),
)

执行以下任意触发实际SQL执行的操作（调用.all()或遍历查询对象）时，代码都会崩溃：

第一种：查询语句后直接调用.all()

db.session.query(db_tables.classes.my_table).filter(
        db_tables.classes.my_table.device_name.like('%uni%'),
    ).all()

第二种：对已生成的查询对象调用.all()

data.all()

第三种：直接遍历查询对象取数

for row in data:
  row.name

执行上述操作返回的报错信息如下：

(pyodbc.ProgrammingError) ('42S22', "[42S22] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]Invalid column name 'row_id'. (207) (SQLExecDirectW)")

问题原因

报错核心来自两处代码逻辑错误：

手动定义的row_id主键字段在目标物理表中不存在，SQLAlchemy生成ORM查询时会默认把主键字段加入查询列，最终发给SQL Server的语句会查询不存在的字段，直接触发SQL语法错误。
db_tables.prepare()阶段传入了reflect=True参数，这一步会重新从数据库加载全量表结构，直接覆盖你之前手动给my_table配置的主键规则，就算你把虚拟主键名改成表中已有字段，这一步的反射覆盖也会导致主键配置失效。

另外要注意：SQLAlchemy ORM本身强制要求映射的表必须存在主键，没有主键的表无法完成正常的ORM映射，不能靠凭空添加不存在的字段绕过这个限制。

修复方案

按以下步骤调整代码即可正常查询：

先完成表结构反射，不要在反射阶段凭空加不存在的字段
选择表中实际存在、可以唯一标识单行数据的一个或多个字段设置为主键；如果表中完全没有可作为唯一标识的字段，可以用SQL Server自带的%%physloc%%行定位虚拟列作为只读场景下的临时主键
调用prepare()时不要开启reflect=True，避免覆盖手动配置的主键规则

修正后的参考代码：

from sqlalchemy import MetaData, Table, Column, Integer, String, text
from sqlalchemy.ext.automap import automap_base

table_name = 'my_table'
db_tables = automap_base()

# 第一步：先反射加载目标表的实际结构
my_table = Table(
    table_name,
    db_tables.metadata,
    autoload=True,
    autoload_with=db.engine
)

# 第二步：给表配置合法主键
# 方案1：用表中实际存在的唯一字段/字段组合，比如用device_id + collect_time两个字段确定唯一行
# my_table.primary_key = [my_table.c.device_id, my_table.c.collect_time]

# 方案2：无合适唯一字段时，用SQL Server自带的physloc虚拟列做只读场景主键
my_table.append_column(
    Column("physloc", String, primary_key=True, server_default=text("%%physloc%%"))
)

# 第三步：prepare时不要重复反射，避免覆盖主键配置
db_tables.prepare()

# 后续查询可正常执行
data = db.session.query(db_tables.classes.my_table).filter(
    db_tables.classes.my_table.device_name.like('%uni%')
).all()

注意：用%%physloc%%作为主键的方案仅适用于只读查询，不要基于该字段做更新、删除操作，避免出现数据操作错误。

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