在使用 SQLAlchemy 执行选择时需要一些帮助来访问 ForeignKey 行的数据

Question

后台架构：

class Checkpoint(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    creator = db.Column(db.Integer, db.ForeignKey('user.id'))
    name = db.Column(db.String(255))
    description = db.Column(db.String(255), nullable=True)
    price = db.Column(db.Float, nullable=True)
    expiry = db.Column(db.DateTime, nullable=True)
    date_created = db.Column(db.DateTime)
    type = db.Column(db.String(255))
    image = db.Column(db.String(255))
    longitude = db.Column(db.Float)
    latitude = db.Column(db.Float)

class UserCheckpoint(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    user = db.relationship("User")
    checkpoint_id = db.Column(db.Integer, db.ForeignKey('checkpoint.id'))
    checkpoint = db.relationship("Checkpoint")

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(255))
    facebook_info = db.Column(db.String(255), db.ForeignKey('facebook_user.id'))
    facebook_user = db.relationship("FacebookUser")

class FriendConnection(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    fb_user_from = db.Column(db.String(255), db.ForeignKey('facebook_user.id'))
    fb_user_to = db.Column(db.String(255), db.ForeignKey('facebook_user.id'))

class FacebookUser(db.Model):
    id = db.Column(db.String(255), primary_key=True)
    name = db.Column(db.String(255))
    first_name = db.Column(db.String(255), nullable=True)
    middle_name = db.Column(db.String(255), nullable=True)
    last_name = db.Column(db.String(255), nullable=True)
    gender = db.Column(db.String(255), nullable=True)
    username = db.Column(db.String(255), nullable=True)
    link = db.Column(db.String(255), nullable=True)

我有一个用户，如您所见，每个用户都有一个 Facebook 个人资料，以及一个描述 Facebook 个人资料间友谊的表格。因此，给定用户，用户将拥有 Facebook 好友列表。我想在给定的检查点条件下获取属于用户或其朋友的所有用户检查点：

coord_conditions = and_(Checkpoint.longitude <= longitude + exp_no,
                                Checkpoint.longitude >= longitude - exp_no,
                                Checkpoint.latitude <= latitude + exp_no,
                                Checkpoint.latitude >= latitude - exp_no,
                                )

如何使用 SQLAlchemy 中的 ORM 执行此操作？谢谢！

总结：鉴于 user_id 属于朋友/自己的列表，如何选择 UserCheckpoints；而 UserCheckpoint.checkpoint 需要满足一组条件。

Answer 1

每个关系都有两种方法来定义相关对象的条件：.has() 用于单个引用对象，.any() 用于集合。这些方法允许将您的任务直接转换为 SQLAlchemy 表达式。让我们将缺失的关系添加到FacebookUser：

class FacebookUser(Model):
    # Definitions from question are here
    user = relationship(User, uselist=False)
    friends = relationship('FacebookUser',
                           secondary=FriendConnection.__table__,
                           primaryjoin=(id==FriendConnection.fb_user_from),
                           secondaryjoin=(FriendConnection.fb_user_to==id))

我已经定义 FacebookUser.user 假设一对一关系（通常在外键列上补充唯一约束）。如果您允许多个用户连接到一个 Facebook 帐户，只需删除 uselist=False 并调整名称即可。

对坐标条件的简短定义：

coord_conditions = Checkpoint.longitude.between(longitude - exp_no,
                                                longitude + exp_no) & \
                   Checkpoint.latitude.between(latitude - exp_no,
                                               latitude + exp_no)

即使是近似值，这个条件也肯定是错误的（-179.9° 和 179.9° 非常接近，而差异很大），但这不是问题的主要主题。

感兴趣的用户的条件（id等于user_id的用户和他的朋友）：

user_cond = (User.id==user_id) | \
            User.facebook_user.has(
                FacebookUser.friends.any(FacebookUser.user.has(id=user_id)))

现在查询很简单：

session.query(UserCheckpoint).filter(
        UserCheckpoint.checkpoint.has(coord_conditions) & \
        UserCheckpoint.user.has(user_cond))

除非您有（或预期）性能问题，否则我建议避免以牺牲可读性为代价进行优化。

Answer 2

基本上您的查询可以分为两部分：

1. 给定user_id，创建一个用户列表，其中包含用户本人以及所有直接朋友
2. 给定来自 1. 的用户列表，获取所有 UserCheckpoint，其 Checkpoint 将满足条件。

未测试代码：

# get direct user for given user_id
u1 = (session.query(User.id.label("user_1_id"), User.id.label("user_id"))
     )

# get friends of the user in one direction (from other user to this one)
User2 = aliased(User)
FacebookUser2 = aliased(FacebookUser)
u2 = (session.query(User2.id.label("user_1_id"), User.id.label("user_id")).
        join(FacebookUser2, User2.facebook_info == FacebookUser2.id).
        join(FriendConnection, FacebookUser2.id == FriendConnection.fb_user_from).
        join(FacebookUser, FacebookUser.id == FriendConnection.fb_user_to).
        join(User, User.facebook_info == FacebookUser.id)
     )

# get friends of the user in other direction (from this user to the other)
User2 = aliased(User)
FacebookUser2 = aliased(FacebookUser)
u3 = (session.query(User2.id.label("user_1_id"), User.id.label("user_id")).
        join(FacebookUser2, User2.facebook_info == FacebookUser2.id).
        join(FriendConnection, FacebookUser2.id == FriendConnection.fb_user_to).
        join(FacebookUser, FacebookUser.id == FriendConnection.fb_user_from).
        join(User, User.facebook_info == FacebookUser.id)
     )

# create a union to have all pairs (me_or_friend_id, user_id)
u_all = union_all(u1, u2, u3)
# **edit-1: added alias **
u_all = u_all.alias("user_list_view")
# final query which adds filters requirested (by user_id and the checkpoint condition)
q = (session.query(UserCheckpoint).
        join(Checkpoint).filter(coord_conditions).
        join(u_all, UserCheckpoint.user_id == u_all.c.user_1_id).
        filter(u_all.c.user_id == user_id)
    )
for u_cp in q.all():
    print u_cp

---

注意，如果您在模型中定义了更多relationships，则可以稍微简化查询，然后可以从join 子句中删除一些primaryjoin 条件。