将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])中复制statistics.mean([st.page_views_per_million for st in self.traffic_summary])或复杂的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逻辑。合并返回第一个非空值。如果没有相关的流量摘要，则第一个子查询的结果为null。如果结果是，则nullif将其转换为null。在这种情况下，如果第二个子查询不是null，则返回第二个子查询的结果，即存在相关的通信量数据点。最后在默认情况下返回。