我的 SqlAlchemy 模型:
Base: DeclarativeMeta = declarative_base()
class TableAModel(Base):
__tablename__ = "TableA"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
last_updated: Mapped[str] = mapped_column(nullable=True)
def __repr__(self):
"""Return a string representation of the object."""
return (
f"<TableA(id='{str(self.id)}', "
f"last_updated='{self.last_updated}')>"
)
选择语句:
select_statement = select(TableAModel).where(
TableAModel.id == some_existing_guid
)
results = self.sql_client.execute_select_statement(
select_statement, select_first=True
)
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 子句的常规选择确实会返回现有行。我将原始查询转储到控制台,查询看起来是正确的。
SELECT "TableA".id, "TableA".last_updated
FROM "TableA"
WHERE "TableA".id = :id_1
必须更换我的主键数据类型。我注意到在 SQLite 中,我的 uuid 映射到
varchar
。一旦我将映射交换为字符串,我的 where 子句就开始工作。
id: Mapped[str] = mapped_column(primary_key=True)