CODEWITHSUNDEEP
pythonormsqlalchemy
EmpireJones
EmpireJones

Reputation: 3086

Directed Graph - SQLAlchemy ORM

I am trying to create a directed graph (with backward references) via SQLAlchemy; how can I get the desired output listed below? [see section titled 'Desired Output']

Example code:

example_db.py

from sqlalchemy import Column, Integer, ForeignKey, Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref

Base = declarative_base()

node_connector = Table('node_connector', Base.metadata, 
    Column('from_node_handle', Integer, ForeignKey('node.handle')),
    Column('to_node_handle', Integer, ForeignKey('node.handle')))


class Node(Base):
    __tablename__ = 'node'
    handle = Column(Integer, nullable=False, primary_key=True)
    next_nodes = relationship('Node',
        secondary=node_connector,
        foreign_keys=[node_connector.c.to_node_handle],
        backref=backref("previous_nodes",
            foreign_keys=[node_connector.c.from_node_handle]))

test/example_db_test.py

Module being executed

from sqlalchemy import create_engine, event
from sqlalchemy.engine import Engine
from sqlalchemy.orm import sessionmaker
import unittest

from ..example_db import Base, Node

class TestNodeDB(unittest.TestCase):

    def setUp(self):
        engine = create_engine('sqlite:///:memory:', echo=False)    
        SessionMaker = sessionmaker(bind=engine)
        self.__session = SessionMaker()
        Base.metadata.create_all(engine)

    def test_node_creation(self):
        node_a = Node()
        node_b = Node()
        node_c = Node()
        self.__session.add(node_a)
        self.__session.add(node_b)
        self.__session.add(node_c)
        node_a.next_nodes.append(node_b)
        node_b.next_nodes.append(node_c)
        self.__session.commit()

        nodes = self.__session.query(Node)

        for node in nodes:
            print "Node handle " + str(node.handle) + ":"
            for next_node in node.next_nodes:
                print "\t next node handle " + str(next_node.handle)
            for previous_node in node.previous_nodes:
                print "\t previous node handle " + str(previous_node.handle)



if __name__ == "__main__":
    unittest.main()

Output

Actual output

Node handle 1:
Node handle 2:
     next node handle 2
     previous node handle 2
Node handle 3:
     next node handle 3
     previous node handle 3

Desired Output

Node handle 1:
     next node handle 2
Node handle 2:
     next node handle 3
     previous node handle 1
Node handle 3:
     previous node handle 2

How can I get this desired output?

Thanks!

Upvotes: 2

Views: 923

Answers (1)

javex
javex

Reputation: 7544

SQLAlchemy even has documentation on this: Self-Referential Many-to-Many Relationship, which covers exactly what you want. Basically, you got it almost right, just instead of using the foreign_keys setting use primaryjoin and secondaryjoin:

next_nodes = relationship('Node',
    secondary=node_connector,
    primaryjoin=handle==node_connector.c.to_node_handle,
    secondaryjoin=handle==node_connector.c.from_node_handle,
    backref="previous_nodes",
    )

Now your tests work. One additional detail: You don't need to add node_b and node_c: They are added through the casacde: If node_a is in the session and you add something to a configured relationship, they will also be added. Also you don't need to commit: When you query the session figures it must first flush, then query.

And also pay attention that you need to watch over loading strategies here. Take the example on Configuring Self-Referential Eager Loading to figure out a way that works for you.

Upvotes: 3

Related Questions