我有一个名为ExtendedUser的表,该表与User表具有一对一的关系,而extended_user表上的反向引用名为UserUser表与UserPosts表具有一对多的关系,在posts表上具有名为User的后向引用。

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)

class UserPost(Base):
    __tablename__ = "user_posts"

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", backref="posts")

class ExtendedUser(Base):
    __tablename__ = "extended_users"

    user_id = Column(Integer, ForeignKey("users.id"), primary_key=True)
    user = relationship("User", backref="extended_user", uselist=False, lazy="joined")


ExtendedUser开始,我想选择所有ExtendedUser没有帖子的User
所以我尝试失败的是

sess.query(ExtendedUser).filter(not_(ExtendedUser.user.posts.any()))


但这不起作用,我得到一个错误:

AttributeError: Neither 'InstrumentedAttribute' object nor 'Comparator' object associated with ExtendedUser.user has an attribute 'posts'


如何为查询建模,以便仅返回ExtendedUser具有UserUserPost

最佳答案

发生错误是因为.user属性是不同的东西,具体取决于您是从类还是从类的实例访问它。例如,这引发了您得到的异常:

ExtendedUser.user.posts


这是因为访问类.user上的ExtendedUser返回一个InstrumentedAttribute对象,并且InstrumentedAttribute的实例没有名为posts的属性。

这有效:

inst = ExtendedUser()
inst.user.posts


上面的方法起作用是因为我们已经访问了.user实例上的ExtendedUser,该实例返回了User实例,该实例具有名为posts的属性。

类和实例属性访问之间的这种不同行为是由Python的descriptor protocol控制的。

实现目标的一种方法是使用子查询在user_id表中查询唯一的user_posts,并测试结果中是否没有ExtendedUseruser_id

q = sess.query(ExtendedUser).\
    filter(
        not_(
            ExtendedUser.user_id.in_(
                sess.query(UserPost.user_id).distinct()
            )
        )
    )


以下是一个有效的示例,但首先我必须稍微更改ExtendedUser.user的定义:

class ExtendedUser(Base):
    ...
    user = relationship(
        "User", backref=backref("extended_user", uselist=False), lazy="joined")


注意使用backref功能,该功能允许我在uselist=False上设置User.extended_user。您的示例在uselist=False上具有ExtendedUser.user,但由于在ExtendedUser上定义了外键,因此此处并不需要,因此ExtendedUser.user只能指向一个用户,并且sqlalchemy会自动知道该集合不应该成为清单。没有该更改,我将得到一个TypeError: Incompatible collection type: ExtendedUser is not list-like异常。

好的,这是示例:

sess = Session()
user_1 = User(extended_user=ExtendedUser())
user_2 = User(extended_user=ExtendedUser())
user_3 = User(extended_user=ExtendedUser())
user_1.posts = [UserPost()]
sess.add_all([user_1, user_2, user_3])
sess.commit()
q = sess.query(ExtendedUser).\
    filter(
        not_(
            ExtendedUser.user_id.in_(
                sess.query(UserPost.user_id).distinct()
            )
        )
    )

print(q.all())  # [ExtendedUser(user_id=2), ExtendedUser(user_id=3)]

10-07 15:14