python - 使用 ORM、声明式样式和关联对象在 SQLAlchemy 中递归选择(深度有限)关系

Question

给定:

DIRECTIONS = db.Enum('N', 'NE', 'E', 'SE', 'S', 'SW', 'W', 'NW',
                     name='directions')


class Exit(BaseModel):
    __tablename__ = 'exits'
    src = db.Column(db.Integer, db.ForeignKey('room.id'), primary_key=True)
    dst = db.Column(db.Integer, db.ForeignKey('room.id'), primary_key=True)
    direction = db.Column(DIRECTIONS, primary_key=True)
    from_room = db.relationship('Room', foreign_keys=[dst],
                                backref=db.backref('exits',
                                                   lazy='dynamic'))
    to_room = db.relationship('Room', foreign_keys=[src]))

使用:

SqlAlchemy 0.9.8，PostgreSQL 9.3.5

在给定起始房间 r 的情况下，如何查询以递归方式选择某个深度 n 的导出？

示例(一个简单的 map ):

                           E                                               
                          /                                                
                B   C -- D                                                 
                 \ /                                                       
                  A                                                         
                  |                                                         
                  F                                                         
                 / \                                                        
                G   H                                                       
                     \                                                      
                      I

假设关系可以用上面的 map 表示，

• 从某个任意房间开始，什么查询会给我所有房间？
• 我如何才能将上述查询限制为从原始房间开始的特定“跳数”？
  • 例如，从 A 开始，通过将深度限制为 2，选择除 E 和 I 之外的所有房间。

当然，我可以用 Python 做到这一点:

rooms = [starting_room]                                                     
for exit in starting_room.exits.all():                                      
    if exit.to_room not in rooms:                                           
        add_to_map(exit)                                                    
        rooms.append(exit.to_room)                                          
for room in rooms[1:]:                                                      
    for exit in room.exits.all():                                           
        if exit.to_room not in rooms:                                                       add_to_map(exit)                                                
            rooms.append(exit.to_room)

但这很昂贵，不适合超过 2 跳。

我试过遵循CTE SQLAlchemy 文档中的示例，但很难理解如何使它像我一样适用于关联对象，尤其是因为我没有在纯 SQL 中使用 CTE 的经验。

我的尝试:

>>> starting_exits = db.session.query(Exit).filter_by(from_room=starting_room).came='starting_exits', recursive=True)
>>> start = orm.aliased(starting_exits, name='st')
>>> exit = orm.aliased(Exit, name='e')
>>> cte = starting_exits.union_all(db.session.query(exit).filter(exit.src == start.c.dst))
>>> db.session.query(cte).all()

无限期挂起，即使我对其进行切片 (.all()[:5])。我应该做什么？

Answer 1

我希望我没有使您的模型过于复杂，但为了测试查询(如下所示)，我使用了以下模型定义:

模型:

class Room(Base):
    __tablename__ = 'room'
    id = Column(Integer, primary_key=True)
    name = Column(String)

    exits = association_proxy(
        'lnk_exits', 'to_room',
        # creator=lambda v: Exit(to_room=v),
        creator=lambda k, v: Exit(direction=k, to_room=v),
    )
    entries = association_proxy(
        'lnk_entries', 'from_room',
        # creator=lambda v: Exit(from_room=v),
        creator=lambda k, v: Exit(direction=k, from_room=v),
    )


class Exit(Base):
    __tablename__ = 'exits'
    src = Column(Integer, ForeignKey('room.id'), primary_key=True)
    dst = Column(Integer, ForeignKey('room.id'), primary_key=True)
    direction = Column(DIRECTIONS, primary_key=True)

    from_room = relationship(
        Room, foreign_keys=[dst],
        # backref='lnk_exits',
        backref=backref(
            "lnk_exits",
            collection_class=attribute_mapped_collection("direction"),
            cascade="all, delete-orphan",
        )
    )
    to_room = relationship(
        Room,
        foreign_keys=[src],
        # backref='lnk_entries',
        backref=backref(
            "lnk_entries",
            collection_class=attribute_mapped_collection("direction"),
            cascade="all, delete-orphan",
        )
    )

您确实不需要像我那样使用关系，但我喜欢我这样做的方式，因为它允许我处理房间之间的关系，如下所示:

# Insert test data
rooms = [Room(name=name) for name in 'ABCDEFGHI']
session.add_all(rooms)

A, B, C, D, E, F, G, H, I = rooms

A.entries = {'NW': B, 'NE': C, 'S': F}
B.entries = {'SE': A}
C.entries = {'E': D, 'SW': A}
D.entries = {'W': C, 'NE': E}
E.entries = {'SW': D}
F.entries = {'N': A, 'SW': G, 'SE': H}
G.entries = {'NE': F}
H.entries = {'NW': F, 'SE': I}

if True:  # add cycle, in which case we get duplicates in the results
    B.entries['E'] = C
    C.entries['W'] = B

session.commit()

您可以在 Association Proxy 中阅读更多相关信息文档的一部分。

现在查询

请注意，为了使用下面的查询，您不需要上面的任何关联代理和相关内容。即使关系简单，当前查询也会挂起 A <--> B因为你CTE将无限期地来回导航。所以诀窍是添加 level信息到CTE这样您就可以在级别上限制搜索。下面的查询应该让你开始:

# parameters
start_id = session.query(Room).filter(Room.name == 'A').first().id
max_level = 2

# CTE definition
starting_exits = (session.query(Exit, literal(0).label("level"))
    .filter(Exit.src == start_id)
    .cte(name="starting_exits", recursive=True)
)
start = aliased(starting_exits, name="st")

exit = aliased(Exit, name="e")
joined = (session.query(exit, (start.c.level + 1).label("level"))
    .filter(exit.src == start.c.dst)
    # @note: below line will avoid simple cycles of 2, which does not always help, but should reduce the result-set significantly already
    .filter(exit.dst != start.c.src)
    .filter(start.c.level < max_level)
)

cte = start.union_all(joined)
for x in session.query(cte).order_by(cte.c.src, cte.c.dst, cte.c.level):
    print(x)

我假设您只对结果查询的第二列 ( dst ) 感兴趣，以便获得 id的 Room你可以到达。可能还有第四列 ( level ) 会很有趣，以便找到到那个房间的最短路径。但是您可能仍然有多种方法可以到达同一个目标房间，因此请事后过滤掉这些方法。

编辑: 使用 cte 的简单方法为了获得房间(模型实例)将是:

# get all Rooms (no duplicates)
s = session.query(cte.c.dst.distinct().label("room_id")).subquery(name="rooms")
q = session.query(Room).join(s, Room.id == s.c.room_id)
for r in q:
    print(r)

编辑 2:要获取导出和房间(模型实例)以重建图形，只需对上面的查询进行另一个连接:

exits = (session.query(Exit, Room)
         .join(s, Exit.dst == s.c.room_id)
         .join(Room, Room.id == s.c.room_id))
for exit, room in exits:
    print exit, room