Database.execute_many()
调用Connection.execute_many()
将您的查询分解为单独的单个查询(每个元素一个values
),这是方法(source):
async def execute_many(
self, query: typing.Union[ClauseElement, str], values: list
) -> None:
queries = [self._build_query(query, values_set) for values_set in values]
async with self._query_lock:
await self._connection.execute_many(queries)
请注意,它调用_build_query()
方法(source):
@staticmethod
def _build_query(
query: typing.Union[ClauseElement, str], values: dict = None
) -> ClauseElement:
if isinstance(query, str):
query = text(query)
return query.bindparams(**values) if values is not None else query
elif values:
return query.values(**values)
return query
由于您没有传递str
查询并且正在传递值,因此控制进入elif values:
条件处理,其中值的各个 dict 被解压缩到.values()
查询的方法中(即Update.values()
)。这实质上使得它试图编译这个查询:
query = (
User.update()
.where(User.c.id == bindparam("_id"))
.values(score=bindparam("score"))
.values(score=2, _id=1)
)
第二个 values 子句导致新的 Update 带有新的绑定参数,这些参数试图为score
和设置值_id
。这会导致查询编译失败,因为_id
表上没有列。
所以重现错误的MCVE真的是这样的:
from sqlalchemy.dialects import postgresql
User.update().values(score=2, _id=1).compile(dialect=postgresql.dialect())
这引发了:
Traceback (most recent call last):
File ".\main.py", line 31, in <module>
User.update().values(score=2, _id=1).compile(dialect=postgresql.dialect())
File "<string>", line 1, in <lambda>
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\elements.py", line 462, in compile
return self._compiler(dialect, bind=bind, **kw)
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\elements.py", line 468, in _compiler
return dialect.statement_compiler(dialect, self, **kw)
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\compiler.py", line 571, in __init__
Compiled.__init__(self, dialect, statement, **kwargs)
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\compiler.py", line 319, in __init__
self.string = self.process(self.statement, **compile_kwargs)
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\compiler.py", line 350, in process
return obj._compiler_dispatch(self, **kwargs)
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\visitors.py", line 92, in _compiler_dispatch
return meth(self, **kw)
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\compiler.py", line 2569, in visit_update
self, update_stmt, crud.ISUPDATE, **kw
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\crud.py", line 62, in _setup_crud_params
return _get_crud_params(compiler, stmt, **kw)
File "C:\Users\peter\Documents\git\stackoverflow\58668615-sqalchemy-update-bindparam-primary-key\.venv\lib\site-packages\sqlalchemy\sql\crud.py", line 177, in _get_crud_params
% (", ".join("%s" % c for c in check))
sqlalchemy.exc.CompileError: Unconsumed column names: _id
总结一下这个问题,您构建一个查询,并将绑定参数传递给Update.where()
和Update.values()
。然后,您将该查询和您的值传递到Database.execute_many()
他们将您的值列表的各个元素解压缩到您的查询的第二次调用中Update.values()
,该查询将您的查询替换为尝试为_id
不存在的列设置值的查询。
除了单独更新每一行之外,还有其他解决方案吗?
当使用 sqlalchemy 引擎和查询时,查询工作得很好:
# using a sqlalchemy engine
engine.execute(query, values)
否则,应该将查询作为字符串发送到,Database.execute_many()
因为这意味着查询将在方法的if isinstance(query, str):
一部分中得到处理,_build_query()
这将避免.values()
对查询进行第二次调用:
db.execute_many(str(query), values)