Python中Sqlalchemy在同一transaction中的BinaryExpression比较问题

假设有一个 model 叫 StockModel, 有一个 Integer 型的 remaining 字段，然后有以下操作：

stock = session.query(StockModel).first()
with session.no_autoflush:
	try:
    	stock = StockModel.remaining - 1  # sql: remaining = remaining - 1
        # 此时 stock 是 <sqlalchemy.sql.elements.BinaryExpression>
        # 若要再对 stock 进行比较，如下：
        if stock.remaining > 10:
        	pass
        # 则会直接报错：
        # TypeError: Boolean value of this clause is not defined
        # 我现在的解决方式是，如果 stock 是 BinaryExpression 的话，就避免
        # 进行比较，直接做减法操作，然后在 commit 之后做二次校验。。。
        # 如下：
        if isinstance(stock.remaining, BinaryExpression):
        	stock.remaining = StockModel.remaining - (stock.remaining.right.value + decrease_amount)
     	session.commit()
        # 在这里做二次校验：
        if stock.remaining < 0:
        	raise Expression("Stock not enough")
    except BaseException:
    	session.rollback()
    finally:
    	session.close()

请问各位大佬，有没有更优雅的解决方案呢……在 sqlalchemy 的文档里徜徉了许久还是木有找到想要的解决办法……


Python中Sqlalchemy在同一transaction中的BinaryExpression比较问题

1 回复

这个问题我遇到过，确实挺坑的。在同一个SQLAlchemy事务里，直接比较两个BinaryExpression对象（比如Column == value这种表达式）会返回False，即使它们逻辑上应该相等。

根本原因是SQLAlchemy的BinaryExpression没有实现值相等的比较（__eq__方法）。当你写expr1 == expr2时，Python默认使用对象标识（is）进行比较，也就是比较它们是不是同一个内存对象，而不是比较它们代表的SQL逻辑是否相同。

看个具体例子：

from sqlalchemy import Column, Integer, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()
engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    age = Column(Integer)

Base.metadata.create_all(engine)

session = Session()
user = User(age=25)
session.add(user)

# 创建两个看起来一样的表达式
expr1 = User.age == 25
expr2 = User.age == 25

print(f"expr1 is expr2: {expr1 is expr2}")  # False - 不是同一个对象
print(f"expr1 == expr2: {expr1 == expr2}")  # False - 默认比较失败

# 正确的比较方式：比较表达式的组成部分
print(f"比较left: {expr1.left is expr2.left}")  # True - 都是User.age
print(f"比较right: {expr1.right.value == expr2.right.value}")  # True - 都是25

要正确比较两个BinaryExpression是否代表相同的SQL条件，你需要手动比较它们的各个组成部分：

def compare_expressions(expr1, expr2):
    """比较两个BinaryExpression是否逻辑等价"""
    if not isinstance(expr1, type(expr2)):
        return False
    
    # 比较操作符
    if expr1.operator.__name__ != expr2.operator.__name__:
        return False
    
    # 比较左侧（通常是Column）
    if expr1.left is not expr2.left:
        return False
    
    # 比较右侧值
    from sqlalchemy.sql.elements import BindParameter
    if isinstance(expr1.right, BindParameter) and isinstance(expr2.right, BindParameter):
        return expr1.right.value == expr2.right.value
    
    return expr1.right == expr2.right

# 使用比较函数
print(f"逻辑等价: {compare_expressions(expr1, expr2)}")  # True

如果你需要频繁进行这种比较，可以考虑封装一个工具函数，或者直接比较编译后的SQL字符串（但要注意参数绑定可能不同）。

简单说：别直接用==比较表达式对象，拆开比较各部分。

回到顶部