我有一个名为ExtendedUser
的表,该表与User
表具有一对一的关系,而extended_user
表上的反向引用名为User
。 User
表与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
具有User
的UserPost
? 最佳答案
发生错误是因为.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
,并测试结果中是否没有ExtendedUser
的user_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)]