如何用SQLAlchemy的column_property判断字段是否在集合中？

你的问题核心是混淆了Python层面的表达式和SQLAlchemy需要的SQL列表达式。直接用pipeline not in NON_RE_RUNNABLE_PIPELINES会在Python代码执行时判断pipeline这个Column对象是否在集合中，得到固定布尔值，而不是让数据库动态判断字段值，所以会触发AttributeError。

正确实现方式 #

使用SQLAlchemy的in_操作符生成SQL的NOT IN表达式，配合column_property实现动态计算：

NON_RE_RUNNABLE_PIPELINES = {'pipe2', 'pipe1'}
class PipelineStatus(Base):
    __tablename__ = "pipeline_status"

    id = Column(UUID(as_uuid=True), primary_key=True, unique=True, nullable=False)
    pipeline = Column(String)
    # 通过SQL表达式实现动态判断
    re_runnable = column_property(~pipeline.in_(NON_RE_RUNNABLE_PIPELINES))

原理说明 #

• pipeline.in_(NON_RE_RUNNABLE_PIPELINES)会生成SQL语句：pipeline IN ('pipe1', 'pipe2')
• 前缀~是SQLAlchemy的取反操作，对应SQL的NOT，最终生成pipeline NOT IN ('pipe1', 'pipe2')
• column_property会把这个SQL表达式作为虚拟列，查询时由数据库动态计算结果，返回布尔值

为什么之前的尝试无效 #

• 直接写pipeline not in NON_RE_RUNNABLE_PIPELINES：Python会判断pipeline这个Column实例是否在字符串集合中，结果是固定的False，传给column_property时因为不是合法的SQL表达式而报错
• 使用literal(pipeline not in NON_RE_RUNNABLE_PIPELINES)：同样是把Python层面得到的固定False作为静态值，不会动态关联数据库字段的实际值

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