Source code for examples.graphs.directed_graph

  1. """a directed graph example."""
  3. from sqlalchemy import Column
  4. from sqlalchemy import create_engine
  5. from sqlalchemy import ForeignKey
  6. from sqlalchemy import Integer
  7. from sqlalchemy.ext.declarative import declarative_base
  8. from sqlalchemy.orm import relationship
  9. from sqlalchemy.orm import sessionmaker
  12. Base = declarative_base()
  15. class Node(Base):
  16.     __tablename__ = "node"
  18.     node_id = Column(Integer, primary_key=True)
  20.     def higher_neighbors(self):
  21.         return [x.higher_node for x in self.lower_edges]
  23.     def lower_neighbors(self):
  24.         return [x.lower_node for x in self.higher_edges]
  27. class Edge(Base):
  28.     __tablename__ = "edge"
  30.     lower_id = Column(Integer, ForeignKey("node.node_id"), primary_key=True)
  32.     higher_id = Column(Integer, ForeignKey("node.node_id"), primary_key=True)
  34.     lower_node = relationship(
  35.         Node, primaryjoin=lower_id == Node.node_id, backref="lower_edges"
  36.     )
  38.     higher_node = relationship(
  39.         Node, primaryjoin=higher_id == Node.node_id, backref="higher_edges"
  40.     )
  42.     def __init__(self, n1, n2):
  43.         self.lower_node = n1
  44.         self.higher_node = n2
  47. engine = create_engine("sqlite://", echo=True)
  48. Base.metadata.create_all(engine)
  50. session = sessionmaker(engine)()
  52. # create a directed graph like this:
  53. #       n1 -> n2 -> n1
  54. #                -> n5
  55. #                -> n7
  56. #          -> n3 -> n6
  58. n1 = Node()
  59. n2 = Node()
  60. n3 = Node()
  61. n4 = Node()
  62. n5 = Node()
  63. n6 = Node()
  64. n7 = Node()
  66. Edge(n1, n2)
  67. Edge(n1, n3)
  68. Edge(n2, n1)
  69. Edge(n2, n5)
  70. Edge(n2, n7)
  71. Edge(n3, n6)
  73. session.add_all([n1, n2, n3, n4, n5, n6, n7])
  74. session.commit()
  76. assert [x for x in n3.higher_neighbors()] == [n6]
  77. assert [x for x in n3.lower_neighbors()] == [n1]
  78. assert [x for x in n2.lower_neighbors()] == [n1]
  79. assert [x for x in n2.higher_neighbors()] == [n1, n5, n7]