All pastes #925046 Raw Edit

Untitled

public python v1 · immutable
·published 2008-03-01 22:56 UTC
rendered paste body
import sqlalchemyimport sqlalchemy.ormimport md5metadata = sqlalchemy.MetaData()room_table = sqlalchemy.Table("rooms", metadata,    sqlalchemy.Column("id", sqlalchemy.String(32), primary_key=True),    sqlalchemy.Column("title", sqlalchemy.String(200)),    sqlalchemy.Column("roomdesc", sqlalchemy.Text()),                              )edge_table = sqlalchemy.Table("edges", metadata,    sqlalchemy.Column("from_room", sqlalchemy.String(32), sqlalchemy.ForeignKey("rooms.id"), index=True, primary_key=True),    sqlalchemy.Column("to_room", sqlalchemy.String(32), sqlalchemy.ForeignKey("rooms.id"), index=True, primary_key=True),    sqlalchemy.Column("direction", sqlalchemy.Integer()),                              )class Edge(object):    def __init__(self, from_node, to_node, direction):        self.from_node = from_node        self.to_node = to_node        self.direction = direction    def __repr__(self):        return "<Edge(%s, %s, %s)>" % (self.from_room, self.to_room, self.direction)class Room(object):    def __init__(self, title, roomdesc):        self.id = md5.new(title + roomdesc).hexdigest()        self.title = title        self.roomdesc = roomdesc    def __repr__(self):        return "<Room(%s, %s)>" % (self.id, self.title)sqlalchemy.orm.mapper(Room, room_table)sqlalchemy.orm.mapper(Edge, edge_table, properties={    "from_node": sqlalchemy.orm.relation(Room, primaryjoin=edge_table.c.from_room==room_table.c.id, backref="to_directions"),    "to_node": sqlalchemy.orm.relation(Room, primaryjoin=edge_table.c.to_room==room_table.c.id, backref="from_directions"),                                                   })engine = sqlalchemy.create_engine("sqlite:///:memory:", echo=True)metadata.create_all(engine)Session = sqlalchemy.orm.sessionmaker(bind=engine, autoflush=True, transactional=True)session = Session()r1 = Room("room1", "someroom")r2 = Room("room2", "someotherroom")Edge(r1, r2, 1)session.save(r1)session.commit()session.clear()print session.query(Room).filter(Room.from_directions.any() | Room.to_directions.any()).all()