创建一个hybrid_property返回上一条记录的值

Question

在尝试创建一个hybrid_property以返回前一个记录的值时，我已经做到了这一点:

from datetime import date
from sqlalchemy import Column, Integer, Date, select, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import sessionmaker
Base = declarative_base()

class A(Base):
__tablename__ = "a"
id_ = Column(Integer, primary_key=True)
record_date = Column(Date)
example_value = Column(Integer)
@hybrid_property
def prev_value(self):
return

@prev_value.expression
def prev_value(cls):
stmt = select(A.example_value)
stmt = stmt.order_by(A.record_date.desc())
stmt = stmt.limit(1)
stmt = stmt.label("prev_value")
return stmt

engine = create_engine("sqlite:///:memory:")
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add(A(record_date=date(2022, 5 ,1), example_value=1))
session.add(A(record_date=date(2022, 5 ,2), example_value=2))
session.add(A(record_date=date(2022, 5 ,3), example_value=3))
session.commit()
prev_value = session.execute(select(A.prev_value).where(A.id_ == 3)).scalar()
print(prev_value)

当前返回的是None，而应该返回2。

我应该在python方法和表达式变体中放入什么?

Answer 1

解决方案:

@prev_value.expression
def prev_value(cls):
A1 = aliased(A, name="a_prev")
stmt = select(A1.example_value)
stmt = stmt.filter(A1.record_date < cls.record_date)
stmt = stmt.order_by(A1.record_date.desc())
stmt = stmt.limit(1)
stmt = stmt.label("prev_value")
return stmt

解释

---

当我从你的问题逐字运行代码(使用sqlite)，我得到的结果实际上是3，而不是None如您所示。实际上，对于查询session.execute(select(A, A.prev_value)) # (1):

所请求的所有行，返回的结果都是相同的3值。

(<A [1] (example_value = 1, id_ = 1, record_date = datetime.date(2022, 5, 1))>, 3)
(<A [2] (example_value = 2, id_ = 2, record_date = datetime.date(2022, 5, 2))>, 3)
(<A [3] (example_value = 3, id_ = 3, record_date = datetime.date(2022, 5, 3))>, 3)

为什么我得到3为您的示例代码?

我认为这是因为子查询没有任何条件将其链接到所请求的行。假设之前的值应该是record_date之前的值，那么要添加到查询的链接应该是:

stmt = stmt.filter(A.record_date < cls.record_date)

但是，运行它，现在将为所有结果生成None。让我们看一下生成的SQL以及没有找到任何行的原因:

SELECT a.id_,
a.record_date,
a.example_value,
(SELECT a.example_value
FROM a
WHERE a.record_date < a.record_date  # >>> the ISSUE is here: always FALSE
ORDER BY a.record_date DESC
LIMIT 1) AS prev_value
FROM a

问题是主查询和子查询都指向同一个表/视图。

解决子查询:为了解决它，我们只需要显式地创建一个子查询，问题就解决了:

@prev_value.expression
def prev_value(cls):
A1 = aliased(A, name="a_prev")
stmt = select(A1.example_value)
stmt = stmt.filter(A1.record_date < cls.record_date)
stmt = stmt.order_by(A1.record_date.desc())
stmt = stmt.limit(1)
stmt = stmt.label("prev_value")
return stmt

和相同的查询(1)产生以下结果:

(<A [1] (example_value = 1, id_ = 1, record_date = datetime.date(2022, 5, 1))>, None)
(<A [2] (example_value = 2, id_ = 2, record_date = datetime.date(2022, 5, 2))>, 1)
(<A [3] (example_value = 3, id_ = 3, record_date = datetime.date(2022, 5, 3))>, 2)

基于以下生成的SQL:

SELECT a.id_,
a.record_date,
a.example_value,
(SELECT a_prev.example_value
FROM a AS a_prev
WHERE a_prev.record_date < a.record_date
ORDER BY a_prev.record_date DESC
LIMIT 1) AS prev_value
FROM a