Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from sqlalchemy import Integer, String, Column, create_engine, or_, select
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.orm import (relationship, sessionmaker, aliased, join, foreign,
- remote)
- engine = create_engine('sqlite:///:memory:', echo=True)
- Base = declarative_base()
- _next_two_nodes = None
- def next_two_nodes_select():
- global _next_two_nodes
- if _next_two_nodes is None:
- _leftside = aliased(Node, name="leftside")
- _rightside = aliased(Node, name="rightside")
- _next_two_nodes = select(
- [_leftside.id.label("left_node_id"),
- _rightside.id.label("right_node_id")]
- ).select_from(
- join(
- _leftside, _rightside,
- or_(
- _leftside.id + 1 == _rightside.id,
- _leftside.id + 2 == _rightside.id
- )
- )
- ).alias()
- return _next_two_nodes
- class Node(Base):
- __tablename__ = 'node'
- id = Column(Integer, primary_key=True)
- label = Column(String, unique=True)
- next_two_nodes = relationship(
- "Node", secondary=next_two_nodes_select,
- primaryjoin=(lambda: foreign(Node.id)
- == remote(next_two_nodes_select().c.left_node_id)),
- secondaryjoin=(lambda: foreign(next_two_nodes_select().c.right_node_id)
- == remote(Node.id)),
- backref="previous_two_nodes",
- viewonly=True
- )
- def __repr__(self):
- return "Node(id={}, Label={})".format(self.id, self.label)
- Base.metadata.create_all(engine)
- Session = sessionmaker(bind=engine)
- session = Session()
- nodes = [
- Node(id=1, label="Node1"),
- Node(id=2, label="Node2"),
- Node(id=3, label="Node3"),
- Node(id=4, label="Node4")
- ]
- session.add_all(nodes)
- session.commit()
- node = session.query(Node).filter_by(id=2).one()
- print(node.next_two_nodes)
- print(node.previous_two_nodes)
- NodeAlias = aliased(Node)
- print(session.query(Node).join(NodeAlias, Node.next_two_nodes).filter(NodeAlias.id == 3).all())
Add Comment
Please, Sign In to add comment