SqlAlchemy v2.0查询WHERE子句失效问题求助

环境 #

• Python 3.9
• SQLAlchemy v2.0.0
• SQLite

SQLAlchemy模型 #

Base: DeclarativeMeta = declarative_base()

class TableAModel(Base):
    __tablename__ = "TableA"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
    some_other_field: Mapped[str] = mapped_column(nullable=True)

    def __repr__(self):
        """Return a string representation of the object."""
        return (
            f"<TableA(id='{str(self.id)}', "
            f"some_other_field='{self.some_other_field}')>"
        )

查询语句 #

select_statement = select(TableAModel).where(
    TableAModel.id == some_existing_guid
)
results = self.sql_client.execute_select_statement(
    select_statement, select_first=True
)

DB层执行代码（db_layer.py） #

def execute(self, statement: Select, select_first: bool = False):
    """Execute SELECT statement."""
    with self.__get_session__() as session:
        session.begin()
        try:
            result = session.execute(statement)
            if select_first:
                return result.first()
            return result.fetchall()
        except SQLAlchemyError as err:
            get_logger_adapter().error(
                "Error occurred during select:", exc_info=True
            )
            raise err

问题描述 #

按SQLAlchemy文档实现按ID查询，但WHERE子句完全无效，总是返回None。已确认：

• 数据库中存在对应ID的行
• 不带WHERE子句的查询能正常返回该行
• 生成的原始SQL语句看起来正确：

SELECT "TableA".id, "TableA".some_other_field 
FROM "TableA" 
WHERE "TableA".id = :id_1

排查方向 #

1. UUID类型匹配问题 #

SQLite无原生UUID类型，SQLAlchemy会将其存储为字符串或二进制。检查some_existing_guid的类型匹配：

• 若数据库存的是字符串格式UUID，传入uuid.UUID对象可能隐式转换失败，尝试手动转字符串：

  select_statement = select(TableAModel).where(
      TableAModel.id == str(some_existing_guid)
  )
  

• 或者在模型定义时显式指定UUID存储规则，确保序列化/反序列化一致：

  id: Mapped[uuid.UUID] = mapped_column(
      UUID(as_uuid=True), primary_key=True
  )
  

2. 会话事务问题 #

DB层代码手动调用session.begin()，但with self.__get_session__()上下文已自动管理事务，手动开启可能导致查询在未提交的事务上下文执行，看不到最新数据。尝试去掉session.begin()：

def execute(self, statement: Select, select_first: bool = False):
    """Execute SELECT statement."""
    with self.__get_session__() as session:
        try:
            result = session.execute(statement)
            if select_first:
                return result.first()
            return result.fetchall()
        except SQLAlchemyError as err:
            get_logger_adapter().error(
                "Error occurred during select:", exc_info=True
            )
            raise err

3. 参数绑定验证 #

打印带实际参数的SQL，确认some_existing_guid是否正确传递：

print(select_statement.compile(compile_kwargs={"literal_binds": True}))

将输出的SQL直接在SQLite客户端执行，验证是否能返回结果。

4. 字符串大小写匹配问题 #

SQLite默认字符串比较区分大小写，若数据库中id列存的是小写UUID，而传入的UUID转字符串后是大写，会导致匹配失败。尝试统一大小写：

select_statement = select(TableAModel).where(
    func.lower(TableAModel.id) == str(some_existing_guid).lower()
)

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