graph1.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

"""a directed graph example."""

from sqlalchemy import *
from sqlalchemy.orm import *

import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

meta = MetaData('sqlite://')

nodes = Table('nodes', meta,
    Column("nodeid", Integer, primary_key=True)
)
 
# here we have lower.nodeid <= higher.nodeid
edges = Table('edges', meta,
    Column("lower_id", Integer, ForeignKey('nodes.nodeid'), primary_key=True),
    Column("higher_id", Integer, ForeignKey('nodes.nodeid'), primary_key=True)
)

meta.create_all()

class Node(object):
    def __init__(self, id):
        self.nodeid = id
    def add_neighbor(self, othernode):
        Edge(self, othernode)
    def higher_neighbors(self):
        return [x.higher_node for x in self.lower_edges]
    def lower_neighbors(self):
        return [x.lower_node for x in self.higher_edges]

class Edge(object):
    def __init__(self, n1, n2):
        if n1.nodeid < n2.nodeid:
            self.lower_node = n1
            self.higher_node = n2
        else:
            self.lower_node = n2
            self.higher_node = n1

mapper(Node, nodes)
mapper(Edge, edges, properties={
    'lower_node':relation(Node,
primaryjoin=edges.c.lower_id==nodes.c.nodeid, backref='lower_edges'),
    'higher_node':relation(Node,
primaryjoin=edges.c.higher_id==nodes.c.nodeid, backref='higher_edges')
    }
)

session = create_session()

# create a directed graph like this:
#       n1 -> n2 -> n5
#                -> n7
#          -> n3 -> n6
                
n1 = Node(1)
n2 = Node(2)
n3 = Node(3)
n4 = Node(4)
n5 = Node(5)
n6 = Node(6)
n7 = Node(7)

n2.add_neighbor(n5)
n3.add_neighbor(n6)
n7.add_neighbor(n2)
n1.add_neighbor(n3)
n2.add_neighbor(n1)

[session.save(x) for x in [n1, n2, n3, n4, n5, n6, n7]]
session.flush()

session.clear()

n2 = session.query(Node).get(2)
n3 = session.query(Node).get(3)

assert [x.nodeid for x in n3.higher_neighbors()] == [6]
assert [x.nodeid for x in n3.lower_neighbors()] == [1]
assert [x.nodeid for x in n2.lower_neighbors()] == [1]
assert [x.nodeid for x in n2.higher_neighbors()] == [5,7]