Two*_*ist 3 python postgresql recursion sqlalchemy flask-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
在n给定起始室的情况下,如何以递归方式选择退出到某个深度r？
示例(简单地图):
E
/
B C -- D
\ /
A
|
F
/ \
G H
\
I
假设关系可以由上面的地图表示,
A,选择除了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]).我该怎么办？
van*_*van 10
我希望我没有使您的模型过度复杂化,但为了测试查询(下面的内容),我使用了以下模型定义:
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的Rooms.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
| 归档时间： |
|
| 查看次数： |
1426 次 |
| 最近记录： |