RonObvious

many-to-many relation through association select statement

Jul 13th, 2018
197
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 2.10 KB | None | 0 0
  1. from sqlalchemy import Integer, String, Column, create_engine, or_, select
  2. from sqlalchemy.ext.declarative import declarative_base
  3. from sqlalchemy.orm import (relationship, sessionmaker, aliased, join, foreign,
  4.                             remote)
  5.  
  6. engine = create_engine('sqlite:///:memory:', echo=True)
  7. Base = declarative_base()
  8.  
  9. _next_two_nodes = None
  10. def next_two_nodes_select():
  11.     global _next_two_nodes
  12.     if _next_two_nodes is None:
  13.         _leftside = aliased(Node, name="leftside")
  14.         _rightside = aliased(Node, name="rightside")
  15.         _next_two_nodes = select(
  16.             [_leftside.id.label("left_node_id"),
  17.              _rightside.id.label("right_node_id")]
  18.         ).select_from(
  19.             join(
  20.                 _leftside, _rightside,
  21.                 or_(
  22.                     _leftside.id + 1 == _rightside.id,
  23.                     _leftside.id + 2 == _rightside.id
  24.                 )
  25.             )
  26.         ).alias()
  27.     return _next_two_nodes
  28.  
  29.  
  30. class Node(Base):
  31.     __tablename__ = 'node'
  32.     id = Column(Integer, primary_key=True)
  33.     label = Column(String, unique=True)
  34.     next_two_nodes = relationship(
  35.         "Node", secondary=next_two_nodes_select,
  36.         primaryjoin=(lambda: foreign(Node.id)
  37.                      == remote(next_two_nodes_select().c.left_node_id)),
  38.         secondaryjoin=(lambda: foreign(next_two_nodes_select().c.right_node_id)
  39.                        == remote(Node.id)),
  40.         backref="previous_two_nodes",
  41.         viewonly=True
  42.     )
  43.  
  44.     def __repr__(self):
  45.         return "Node(id={}, Label={})".format(self.id, self.label)
  46.  
  47.  
  48. Base.metadata.create_all(engine)
  49. Session = sessionmaker(bind=engine)
  50. session = Session()
  51.  
  52. nodes = [
  53.     Node(id=1, label="Node1"),
  54.     Node(id=2, label="Node2"),
  55.     Node(id=3, label="Node3"),
  56.     Node(id=4, label="Node4")
  57. ]
  58. session.add_all(nodes)
  59. session.commit()
  60.  
  61. node = session.query(Node).filter_by(id=2).one()
  62. print(node.next_two_nodes)
  63. print(node.previous_two_nodes)
  64.  
  65. NodeAlias = aliased(Node)
  66. print(session.query(Node).join(NodeAlias, Node.next_two_nodes).filter(NodeAlias.id == 3).all())
Add Comment
Please, Sign In to add comment