使用 ORM、声明式样式和关联对象在 SQLAlchemy 中递归 select（深度有限）关系

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，PostgreSQL9.3.5

如何递归查询 select 退出到某个深度 n 给定起始房间 r？

示例（简单地图）：

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

什么查询会给我所有房间，从任意房间开始？
如何将上述查询限制为原始房间中一定数量的 'hops'？
- 例如，从 A 开始，select 所有房间，除了 E 和 I，将深度限制为 2。

当然可以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 跳。

我尝试按照 SQLAlchemy 文档中的 CTE 示例进行操作，但是很难理解如何使它像我一样为关联对象工作，尤其是因为我没有在纯 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 部分阅读更多相关信息。

现在 QUERY

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

# 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) 感兴趣，以便获得您可以访问的 Room 中的 id。为了找到通往该房间的最短路径，第四列 (level) 可能也很有趣。但是您可能仍然有多种方法可以到达同一个目标房间，所以请将这些方法过滤掉 post.

编辑： 使用 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

使用 ORM、声明式样式和关联对象在 SQLAlchemy 中递归 select（深度有限）关系

Recursively select (with limited depth) relationships in SQLAlchemy using ORM, declarative style, and Association objects

python

postgresql

recursion

sqlalchemy

flask-sqlalchemy

模特：

现在 QUERY