sqlalchemy hybrid_attribute表达式

Question

假设以下模型：

class Worker(Model):
    __tablename__ = 'workers'
    ...
    jobs = relationship('Job',
                        back_populates='worker',
                        order_by='desc(Job.started)',
                        lazy='dynamic')

    @hybrid_property
    def latest_job(self):
        return self.jobs.first()  # jobs already ordered descending

    @latest_job.expression
    def latest_job(cls):
        Job = db.Model._decl_class_registry.get('Job')
        return select([func.max(Job.started)]).where(cls.id == Job.worker_id).as_scalar()

class Job(Model):
    ...
    started = db.Column(db.DateTime, default=datetime.utcnow)
    worker_id = db.Column(db.Integer, db.ForeignKey('workers.id'))
    worker = db.relationship('Worker', back_populates='jobs')

虽然此查询提供了正确的结果：

db.session.query(Worker).join(Job.started).filter(Job.started >= datetime.datetime(2017, 5, 10, 0, 2, 45, 932983)).distinct().count()

我假设我可以直接查询该字段，但此查询失败：

db.session.query(Worker).join(Job).filter(Worker.latest_job.started >= datetime.datetime(2017, 5, 10, 0, 2, 45, 932983)).count()

出现以下错误：

AttributeError: Neither 'hybrid_property' object nor 'ExprComparator' object associated with Worker.latest_job has an attribute 'started'

如何直接查询该属性？这里我漏掉了什么？

编辑1:遵循@Ilja的建议，我尝试了：

db.session.query(Worker).\
    join(Job).\
    filter(Worker.latest_job >= datetime.datetime(2017, 5, 10, 0, 2, 45, 932983)).\
    count()

但是会出现这个错误：

TypeError: '>=' not supported between instances of 'Select' and 'datetime.datetime'

Answer 1

当在SQL (类)上下文中使用时，您将从混合属性返回一个标量子查询，因此只需像使用值表达式一样使用它：

db.session.query(Worker).\
    filter(Worker.latest_job >= datetime.datetime(2017, 5, 10, 0, 2, 45, 932983)).\
    count()

在这种情况下，混合属性本身需要显式地处理相关性：

@latest_job.expression
def latest_job(cls):
    Job = db.Model._decl_class_registry.get('Job')
    return select([func.max(Job.started)]).\
        where(cls.id == Job.worker_id).\
        correlate(cls).\
        as_scalar()

请注意，混合属性的Python端和SQL端之间存在一些不对称。与在SQL中生成max(started)的相关标量子查询相比，它在实例上访问时会生成最新的Job对象。如果您希望它在SQL语言中也返回一个Job行，那么可以这样做

@latest_job.expression
def latest_job(cls):
    Job = db.Model._decl_class_registry.get('Job')
    return Job.query.\
        filter(cls.id == Job.worker_id).\
        order_by(Job.started.desc()).\
        limit(1).\
        correlate(cls).\
        subquery()

但这在很大程度上实际上用处不大，因为通常-但并不总是-这种相关子查询将比对子查询的连接慢。例如，为了获取具有满足原始标准的最新职务的工人：

job_alias = db.aliased(Job)
# This reads as: find worker_id and started of jobs that have no matching
# jobs with the same worker_id and greater started, or in other words the
# worker_id, started of the latest jobs.
latest_jobs = db.session.query(Job.worker_id, Job.started).\
    outerjoin(job_alias, and_(Job.worker_id == job_alias.worker_id,
                              Job.started < job_alias.started)).\
    filter(job_alias.id == None).\
    subquery()

db.session.query(Worker).\
    join(latest_jobs, Worker.id == latest_jobs.c.worker_id).\
    filter(latest_jobs.c.started >= datetime.datetime(2017, 5, 10, 0, 2, 45, 932983)).\
    count()

当然，如果您只想要计数，那么您根本不需要连接：

job_alias = db.aliased(Job)
db.session.query(func.count()).\
    outerjoin(job_alias, and_(Job.worker_id == job_alias.worker_id,
                              Job.started < job_alias.started)).\
    filter(job_alias.id == None,
           Job.started >= datetime.datetime(2017, 5, 10, 0, 2, 45, 932983)).\
    scalar()

请注意，对Query.scalar()的调用与对Query.as_scalar()的调用不同，它只返回第一行的第一个值。