SQLAlchemy Delete语句where子句同表列比较报错解决

我有一张名为Stats的表，共包含3个字段：

• created：记录创建时间
• data：JSON blob类型字段
• max_age_ms：记录最大存活时长，单位毫秒。例如取值60000时，若created时间早于当前时间60秒，该记录就满足清理条件。

我最初尝试编写如下where子句实现过期记录清理：

utc_now = datetime.utcnow()
purge_ = Stats.__table__.delete().where(Stats.created>=(utc_now - timedelta(seconds=Stats.max_age_ms/1000)))
session.execute(purge_)
session.flush()

执行后抛出异常：

Exception has occurred: TypeError
unsupported type for timedelta seconds component: BinaryExpression

问题出在Stats.max_age_ms字段上，我不确定比较运算符的右侧是否不支持直接传入表字段名，所以尝试参考文档使用hybrid_property，分别搭配filter_by和filter方法查询，仍然抛出同类错误：

# 第一次尝试
session.query(Stats).filter_by(age_seconds>100)
# 报错：NameError: name 'age_seconds' is not defined

# 第二次尝试
session.query(Stats).filter(Stats.age_seconds>100)
# 报错：...AttributeError: 'Comparator' object has no attribute 'seconds'
# 完整报错栈：
# The above exception was the direct cause of the following exception:
# Traceback (most recent call last):
#   File "<string>", line 1, in <module>
#   File "C:\ProgramData\Anaconda3\envs\py39\lib\site-packages\sqlalchemy\ext\hybrid.py", line 925, in __get__
#     return self._expr_comparator(owner)
#   File "C:\ProgramData\Anaconda3\envs\py39\lib\site-packages\sqlalchemy\ext\hybrid.py", line 1143, in expr_comparator
#     ...
# AttributeError: Neither 'BinaryExpression' object nor 'Comparator' object has an attribute 'seconds'

后续尝试使用purge_dt混合属性时也出现类似报错：

utc_now = datetime.utcnow()
old_entries = session.query(Stats).filter(Stats.purge_dt>Stats.created).all()
# 报错：Exception has occurred: TypeError ... unsupported type for timedelta days component: InstrumentedAttribute

目前暂时通过逐行遍历判断的方案实现需求，小数据量下可正常运行，但数据量增长后存在明显性能隐患：

stats = session.query(Stats).all()
ready_for_purge = [x for x in stats if x.created<=x.purge_dt]
for x in ready_for_purge:
    session.delete(x)
session.flush()

附相关模型定义代码：

import sqlalchemy as sa
import sqlalchemy.dialects.postgresql as psql
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.ext.hybrid import hybrid_property

class Stats(DbModel):
    __tablename__ = "stats"

    id = sa.Column(sa.Integer, primary_key=True)
    data = sa.Column(MutableDict.as_mutable(psql.JSONB))
    max_age_ms = sa.Column(sa.Integer(), index=False, nullable=False)
    created = sa.Column(sa.DateTime, nullable=False, default=datetime_gizmo.utc_now)

    @hybrid_property
    def age_seconds(self):
        utc_now = datetime.utcnow()
        return (utc_now - self.created).seconds

    @hybrid_property
    def purge_dt(self):
        utc_now = datetime.utcnow()
        return utc_now - timedelta(seconds=self.max_age_ms/1000)

    @purge_dt.expression
    def purge_dt(cls):
        utc_now = datetime.utcnow()
        return sa.func.dateadd(sa.func.now(), sa.bindparam('timedelta', timedelta(seconds=cls.max_age_ms/1000), sa.Interval()))

报错的核心原因是：Python原生的timedelta计算只能处理Python侧的数值/时间对象，不能直接处理SQLAlchemy的字段表达式（也就是BinaryExpression/InstrumentedAttribute对象），这类字段运算必须翻译成对应数据库可执行的SQL函数表达式，不能在Python侧直接计算。

针对PostgreSQL方言，正确的实现方式如下：

1. 修正混合属性的SQL表达式，不要在表达式里用Python的timedelta处理表字段，直接用数据库侧的时间运算逻辑
2. 时间差计算统一放到SQL执行层完成，避免Python侧和SQL侧的运算逻辑不兼容

修正后的模型代码：

import sqlalchemy as sa
import sqlalchemy.dialects.postgresql as psql
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.ext.hybrid import hybrid_property
from datetime import datetime

class Stats(DbModel):
    __tablename__ = "stats"

    id = sa.Column(sa.Integer, primary_key=True)
    data = sa.Column(MutableDict.as_mutable(psql.JSONB))
    max_age_ms = sa.Column(sa.Integer(), index=False, nullable=False)
    created = sa.Column(sa.DateTime, nullable=False, default=datetime.utcnow)

    # Python侧实例属性计算逻辑（单条记录取值时用）
    @hybrid_property
    def is_expired(self):
        return (datetime.utcnow() - self.created).total_seconds() * 1000 >= self.max_age_ms

    # SQL侧表达式逻辑（生成查询/删除语句时用）
    @is_expired.expression
    def is_expired(cls):
        # PostgreSQL支持直接用时间戳差值做运算，先把created转成epoch时间戳（单位秒）
        # 当前时间epoch减去created的epoch，得到已存活秒数，乘以1000转成毫秒，和max_age_ms比较
        return (sa.extract('epoch', sa.func.now()) - sa.extract('epoch', cls.created)) * 1000 >= cls.max_age_ms

直接执行批量删除的代码，不需要先查询所有记录到内存：

# 直接过滤过期记录执行删除，全程在数据库侧完成，性能远高于全表遍历
purge_stmt = Stats.__table__.delete().where(Stats.is_expired)
session.execute(purge_stmt)
session.commit()

如果不想用混合属性，也可以直接在删除语句里写过滤条件：

utc_now = datetime.utcnow()
# 注意：这里不要用Python的timedelta算字段差值，直接写SQL表达式
purge_stmt = Stats.__table__.delete().where(
    (sa.extract('epoch', utc_now) - sa.extract('epoch', Stats.created)) * 1000 >= Stats.max_age_ms
)
session.execute(purge_stmt)
session.commit()

注意：如果用的不是PostgreSQL，只需要把时间运算的函数换成对应数据库支持的语法即可，核心逻辑是字段参与的时间运算必须生成SQL表达式，不能用Python原生的时间计算函数处理表字段对象。

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