使用SQLAlchemy 2.0调用.values().returning()时触发NotImplementedError

问题背景 #

刚接触Web开发几个月，当前基于FastAPI开发后端，使用SQLAlchemy 2.0操作数据库时，调用update_member函数更新数据时，在赋值result的代码行抛出NotImplementedError。

相关代码与信息 #

数据库表字段定义 #

models.Member表中first_name字段定义：

first_name = Column(String, nullable=False)

请求体patch_data #

已通过print确认键值对合法，其他字段值为None：

patch_data = {"first_name": "test" ...}

出错的update_member函数 #

async def update_member(db: AsyncSession,
                        short_uuid: str,
                        patch_data: MemberSelfPatch | MemberMasterPatch,
                        response_schema: Type[S]) -> S:
    async with db.begin():
        result = (await db.execute(
            update(models.Member)
            .where(models.Member.short_uuid == short_uuid)
            .values(patch_data.dict())
            .returning('*')
        )).fetchone()

可正常运行的对比函数 #

create_member函数

async def create_member(db: AsyncSession, data: MemberMasterCreate) -> MemberMasterView:
    uuid = uuid4()
    async with db.begin():
        db.add_all([
            Member(uuid=str(uuid), **data.dict()),
        ])
    return await get_member(db=db, uuid=uuid, response_schema=MemberMasterView)

get_member_by_short_uuid函数

async def get_member_by_short_uuid(db: AsyncSession, short_uuid: str, response_schema: Type[S]) -> S:    
    result = (await db.execute(
        select(models.Member)
        .where(models.Member.short_uuid == short_uuid)
    )).fetchone()

错误堆栈信息 #

INFO:     127.0.0.1:46434 - "PATCH /members/BqUoAGdN HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
Traceback (most recent call last):
File "/code/./tmeit_backend/deps.py", line 86, in __call__
  yield member
File "/code/.venv/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py", line 21, in __call__
  raise e
File "/code/.venv/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py", line 18, in __call__
  await self.app(scope, receive, send)
File "/code/.venv/lib/python3.10/site-packages/starlette/routing.py", line 680, in __call__
  await route.handle(scope, receive, send)
File "/code/.venv/lib/python3.10/site-packages/starlette/routing.py", line 275, in handle
  await self.app(scope, receive, send)
File "/code/.venv/lib/python3.10/site-packages/starlette/routing.py", line 65, in app
  response = await func(request)
File "/code/.venv/lib/python3.10/site-packages/fastapi/routing.py", line 231, in app
  raw_response = await run_endpoint_function(
File "/code/.venv/lib/python3.10/site-packages/fastapi/routing.py", line 160, in run_endpoint_function
  return await dependant.call(**values)
File "/code/./tmeit_backend/routers/members.py", line 123, in patch_member
  member = await update_member(db=db, short_uuid=short_uuid, patch_data=validated_data, response_schema=response_schema)
File "/code/./tmeit_backend/crud/members.py", line 104, in update_member
  result = (await db.execute(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/ext/asyncio/session.py", line 312, in execute
  result = await greenlet_spawn(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 154, in greenlet_spawn
  result = context.switch(*args, **kwargs)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 2081, in execute
  return self._execute_internal(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 1976, in _execute_internal
  result: Result[Any] = compile_state_cls.orm_execute_statement(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/bulk_persistence.py", line 1522, in orm_execute_statement
  return super().orm_execute_statement(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/context.py", line 250, in orm_execute_statement
  result = conn.execute(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1392, in execute
  return meth(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 487, in _execute_on_connection
  return connection._execute_clauseelement(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1608, in _execute_clauseelement
  compiled_sql, extracted_params, cache_hit = elem._compile_w_cache(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 652, in _compile_w_cache
  compiled_sql = self._compiler(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 288, in _compiler
  return dialect.statement_compiler(dialect, self, **kw)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/compiler.py", line 1075, in __init__
  Compiled.__init__(self, dialect, statement, **kwargs)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/compiler.py", line 639, in __init__
  self.string = self.process(self.statement, **compile_kwargs)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/compiler.py", line 671, in process
  return obj._compiler_dispatch(self, **kwargs)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/annotation.py", line 359, in _compiler_dispatch
  return self.__element.__class__._compiler_dispatch(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/visitors.py", line 143, in _compiler_dispatch
  return meth(self, **kw)  # type: ignore  # noqa: E501
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/compiler.py", line 5009, in visit_update
  compile_state = update_stmt._compile_state_factory(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/sql/base.py", line 638, in create_for_statement
  return klass.create_for_statement(statement, compiler, **kw)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/bulk_persistence.py", line 1336, in create_for_statement
  self._setup_for_orm_update(statement, compiler)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/bulk_persistence.py", line 1423, in _setup_for_orm_update
  new_stmt = self._setup_orm_returning(
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/bulk_persistence.py", line 493, in _setup_orm_returning
  fsc.setup_dml_returning_compile_state(dml_mapper)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/context.py", line 749, in setup_dml_returning_compile_state
  entity.setup_dml_returning_compile_state(self, adapter)
File "/code/.venv/lib/python3.10/site-packages/sqlalchemy/orm/context.py", line 2394, in setup_dml_returning_compile_state
  raise NotImplementedError()
     NotImplementedError

问题原因及解决方法 #

这个错误源于SQLAlchemy ORM模式下，批量更新构造器配合.returning('*')的用法存在限制：要么是数据库方言（如SQLite）对DML RETURNING语法的ORM映射支持不足，要么是ORM本身不支持直接返回完整实例。

方案1：改用先查询再更新的ORM常规流程（推荐） #

先查询到目标实例，修改属性后提交并刷新，确保获取最新数据：

async def update_member(db: AsyncSession,
                        short_uuid: str,
                        patch_data: MemberSelfPatch | MemberMasterPatch,
                        response_schema: Type[S]) -> S:
    async with db.begin():
        # 查询目标实例
        member = (await db.execute(
            select(models.Member).where(models.Member.short_uuid == short_uuid)
        )).scalar_one()
        # 仅更新请求中提供的非空字段
        update_data = patch_data.dict(exclude_unset=True)
        for key, value in update_data.items():
            if value is not None:
                setattr(member, key, value)
        # 提交并刷新实例
        await db.commit()
        await db.refresh(member)
    return response_schema.from_orm(member)

方案2：调整.returning用法，指定具体返回字段 #

如果坚持用批量更新构造器，不要用returning('*')，明确指定需要返回的字段后手动映射到响应模型：

async def update_member(db: AsyncSession,
                        short_uuid: str,
                        patch_data: MemberSelfPatch | MemberMasterPatch,
                        response_schema: Type[S]) -> S:
    async with db.begin():
        result = (await db.execute(
            update(models.Member)
            .where(models.Member.short_uuid == short_uuid)
            .values(patch_data.dict(exclude_unset=True))
            # 明确指定返回字段
            .returning(models.Member.id, models.Member.first_name, models.Member.short_uuid)
        )).fetchone()
        if not result:
            raise HTTPException(status_code=404, detail="Member not found")
        return response_schema(**result._asdict())

方案3：检查数据库方言支持 #

如果使用SQLite，需确保版本在3.35.0及以上（该版本才支持RETURNING语法），且SQLAlchemy对其RETURNING的ORM映射支持有限；建议切换到PostgreSQL等对RETURNING语法支持更完善的数据库。

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