将 SQLAlchemy @hybrid_property 转换为 SQL .expression

Question

我正在尝试使用@hybrid_property 对我的父表进行排序，并且我了解到为了做到这一点，@hybrid_property 必须附加到一个有效的表达式。

阅读 - SQLAlchemy order by function result

    @hybrid_property
    def traffic(self):
        # this getter is used when accessing the property of an instance
        if self.traffic_summary and self.traffic_summary != []:
            traffic = statistics.mean(
                [st.page_views_per_million for st in self.traffic_summary]
            )
            if traffic == 0:
                if self.traffic_history and self.traffic_history != []:
                    traffic = statistics.mean(
                    [st.page_views_per_million for st in self.traffic_history[:30]]
                    )
        else:
            if self.traffic_history and self.traffic_history != []:
                traffic = statistics.mean(
                [st.page_views_per_million for st in self.traffic_history[:30]]
                )

            else:
                traffic = 0

        return int(traffic)

    @traffic.expression
    def traffic(cls):
        # this expression is used when querying the model
        return case(
            [(cls.traffic_summary != None), cls.traffic_history)],
            else_=cls.traffic_summary
        )

@traffic.expression 是具体代码，我要修改，问题是，我完全不知道如何复制statistics.mean([st.page_views_per_million for st in self.traffic_summary]) 或SQL 中复杂的Python 逻辑。

我的问题是双重的。

1. 如何继续将上述内容转换为 SQL？
2. 这样复杂的逻辑是否可以转成SQL？

更新了关系模型和在 parent_table 上设置关系的方式：

    traffic_summary = db.relationship(
        "traffic_summary", backref="traffic", passive_deletes=True, lazy="subquery"
    )
    traffic_by_country = db.relationship(
        "traffic_by_country",
        backref="store",
        passive_deletes=True,
        lazy="select",
        order_by="desc(traffic_by_country.contribution_of_users)",
    )
    traffic_history = db.relationship(
        "traffic_datapoint",
        backref="traffic",
        passive_deletes=True,
        lazy="select",
        order_by="desc(traffic_datapoint.date)",
    )

class traffic_datapoint(ResourceMixin, db.Model):
    id = db.Column(db.BigInteger, primary_key=True)
    date = db.Column(db.DateTime)
    page_views_per_million = db.Column(db.BigInteger)
    page_views_per_user = db.Column(db.Float)
    alexa_rank = db.Column(db.BigInteger)
    reach_per_million = db.Column(db.BigInteger)
    store_id = db.Column(
        db.BigInteger,
        db.ForeignKey(top_store.id, onupdate="CASCADE", ondelete="CASCADE"),
        index=True,
        nullable=True,
    )


class traffic_by_country(ResourceMixin, db.Model):
    id = db.Column(db.BigInteger, primary_key=True)
    country_code = db.Column(db.String(30))
    country_name = db.Column(db.String(100))
    contribution_of_pageviews = db.Column(db.Float)
    contribution_of_users = db.Column(db.Float)
    store_id = db.Column(
        db.BigInteger,
        db.ForeignKey(top_store.id, onupdate="CASCADE", ondelete="CASCADE"),
        index=True,
        nullable=True,
    )


class traffic_summary(ResourceMixin, db.Model):
    id = db.Column(db.BigInteger, primary_key=True)
    summary_type = db.Column(db.String(100))
    alexa_rank = db.Column(db.BigInteger)
    alexa_rank_delta = db.Column(db.BigInteger)
    page_views_per_million = db.Column(db.BigInteger)
    page_views_per_user = db.Column(db.Float)
    reach_per_million = db.Column(db.BigInteger)
    store_id = db.Column(
        db.BigInteger,
        db.ForeignKey(top_store.id, onupdate="CASCADE", ondelete="CASCADE"),
        index=True,
        nullable=True,
    )

Answer 1

将列表理解转换为 SQL 有点容易。毕竟它们类似于 SQL 查询。您可以从中选择关系变量或表，可以选择匹配谓词的元组，并且可以将结果限制为属性的子集。总结（聚合）有点不同，但不会太多。

# Assume S is a set of named tuples
[(T.X, T.Y) for T in S if T.Z == 'foo']
# is like
# SELECT T.X, T.Y FROM S AS T WHERE T.Z = 'foo'

从上面我们得到

@traffic.expression
def traffic(cls):
    traffic_history = select([traffic_datapoint.page_views_per_million]).\
        where(traffic_datapoint.store_id == cls.id).\
        order_by(traffic_datapoint.date.desc()).\
        limit(30).\
        correlate_except(traffic_datapoint).\
        alias()

    return func.coalesce(
        func.nullif(
            select([func.avg(traffic_summary.page_views_per_million)]).
                where(traffic_summary.store_id == cls.id).
                correlate_except(traffic_summary).
                as_scalar(),
            0),
        select([func.avg(traffic_history.c.page_views_per_million)]).
            as_scalar(),
        0).cast(Integer)

coalesce(nullif(..., 0), ..., 0) 复制了 Python 中的 if-else 逻辑。合并返回第一个非空值。如果没有相关的流量摘要，则第一个子查询结果为空。如果结果为 0，则 nullif 将其转换为 null。在这种情况下，返回第二个子查询的结果，如果它不为空，即存在相关的流量数据点。最后默认返回0。