如何定义表示集合中最新对象的SQLAlchemy关系？

Question

我有一个SQLAlchemy模型，其中表x和表y之间有一对多的关系。在表y中，记录的id最大（如果有的话）且满足y.x_id = x.id的那条记录是特别的。类X和类Y分别对应表x和表y。

我知道怎么定义X.all_y（使用ORDER BY y.id排序）。那么，我该怎么定义X.latest_y，让它等同于X.all_y[-1]呢？

Answer 1

这是一个版本，来自@zzzeek的回答，它使用了声明式的方式来进行映射，而不是命令式的方式。这里用到了declared_attr，这个方法可以把关系插入到父类的__mapper_args__中。

import sqlalchemy as sa
from sqlalchemy import orm

Base = orm.declarative_base()


class Child(Base):
    __tablename__ = 'children'

    id = sa.Column(sa.Integer, primary_key=True)
    sortkey = sa.Column(sa.Integer, nullable=False)
    parent_id = sa.Column(sa.Integer, sa.ForeignKey('parents.id'))
    parent = orm.relationship('Parent', back_populates='children')


class Parent(Base):
    __tablename__ = 'parents'

    id = sa.Column(sa.Integer, primary_key=True)
    children = orm.relationship('Child', back_populates='parent')

    @orm.declared_attr
    def __mapper_args__(cls):
        children = Child.__table__
        most_recent_child = (
            sa.select(children.c.id)
            .where(children.c.parent_id == cls.id)
            .order_by(children.c.sortkey.desc())
            .limit(1)
            .correlate(cls.__table__)
            .scalar_subquery()
        )

        rel = orm.relation(
            Child,
            primaryjoin=sa.and_(
                Child.id == most_recent_child, Child.parent_id == cls.id
            ),
            uselist=False,
            viewonly=True,
        )
        return {'properties': {'latest_child': rel}}


# Build and test.
engine = sa.create_engine('sqlite://', echo=True, future=True)
Base.metadata.create_all(engine)
Session = orm.sessionmaker(engine, future=True)

with Session.begin() as s:
    children = [Child(sortkey=i) for i in range(1, 6)]
    parent = Parent(children=children)
    s.add(parent)

with Session() as s:
    w = s.scalars(sa.select(Parent)).first()
    assert w.latest_child.sortkey == 5, f'{w.latest_child.sortkey=}'
    assert len(w.children) == 5, f'{len(w.children)=}'

Answer 2

用纯粹的关系型方法来实现这个功能，需要使用一个子查询来获取“最新”或“最大”的值，这个值是和父级相关联的，然后把这个值和集合中的成员进行比较。这意味着如果你在决定“最新”的那一列上加个索引，你会得到更好的结果：

from sqlalchemy import *
from sqlalchemy.orm import *

engine = create_engine('sqlite:///:memory:', echo='debug')

m = MetaData()

parent = Table('parent', m, 
                Column('id', Integer, primary_key=True)
)

child = Table('child', m, 
                Column('id', Integer, primary_key=True),
                Column('parent_id', Integer, ForeignKey('parent.id')),
                Column('sortkey', Integer)
                )

m.create_all(engine)

class Parent(object):
    def __init__(self, children):
        self.all_c = children

class Child(object):
    def __init__(self, sortkey):
        self.sortkey = sortkey

latest_c = select([func.max(child.c.sortkey)]).\
                where(child.c.parent_id==parent.c.id).\
                correlate(parent).\
                as_scalar()

mapper(Parent, parent, properties={
    'all_c':relation(Child),
    'latest_c':relation(Child, 
                            primaryjoin=and_(
                                child.c.sortkey==latest_c, 
                                child.c.parent_id==parent.c.id
                            ),
                            uselist=False
    )
})

mapper(Child, child)

session = sessionmaker(engine)()

p1, p2, p3 = Parent([Child('a'), Child('b'), Child('c')]), \
                Parent([Child('b'), Child('c')]),\
                Parent([Child('f'), Child('g'), Child('c')])

session.add_all([p1, p2, p3])
session.commit()

assert p1.latest_c.sortkey == 'c'
assert p2.latest_c.sortkey == 'c'
assert p3.latest_c.sortkey == 'g'

另外，在某些平台上，你可以使用LIMIT，这样可以更快地得到结果，因为你可以避免聚合操作，并且可以通过主键来连接集合项：

latest_c = select([child.c.id]).\
                where(child.c.parent_id==parent.c.id).\
                order_by(child.c.sortkey.desc()).\
                limit(1).\
                correlate(parent).\
                as_scalar()

mapper(Parent, parent, properties={
    'all_c':relation(Child),
    'latest_c':relation(Child, 
                            primaryjoin=and_(
                                child.c.id==latest_c, 
                                child.c.parent_id==parent.c.id
                            ),
                            uselist=False
    )
})