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

鉴于：

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的查询？

示例(简单地图)：

假设关系可以由上面的地图表示,

>什么查询会给我所有的房间,从一些任意的房间开始？
>我怎样才能将上述查询限制在原始房间的一定数量的“跃点”中？

>例如,从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跳.

我已经尝试过遵循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]).我该怎么办？

解决方法:

我希望我没有使您的模型过度复杂化,但为了测试查询(下面的内容),我使用了以下模型定义：

该模型：

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,以便您可以限制在级别上的搜索.下面的查询可以帮助您入门：

# 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.为了找到到那个房间的最短路径,可能还会感兴趣的是第四列(级别).但是你仍然可以通过多种方式进入同一个目标房间,所以请事后过滤掉.

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

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

相关推荐