Source code for fornax.model

from sqlalchemy import Column, Integer, Float, CheckConstraint, String
from sqlalchemy import PrimaryKeyConstraint, Index, UniqueConstraint
from sqlalchemy import ForeignKey, ForeignKeyConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.types import BigInteger
Base = declarative_base()


[docs]class Graph(Base): """ A graph containing nodes and edges """ __tablename__ = 'graph' graph_id = Column(Integer, primary_key=True) Index('graph_id')
[docs]class Query(Base): __tablename__ = 'query' __table_args__ = ( UniqueConstraint('query_id', 'start_graph_id', 'end_graph_id'), ) query_id = Column(Integer, primary_key=True) start_graph_id = Column( Integer, ForeignKey("graph.graph_id"), nullable=False, index=True) end_graph_id = Column( Integer, ForeignKey("graph.graph_id"), nullable=False, index=True) Index( 'query_idx', 'query_id', 'start_graph_id', 'end_graph_id', unique=True) Index('query_id')
[docs]class Match(Base): """Joins Query Nodes to Candidate Target Nodes""" __tablename__ = 'match' __table_args__ = ( PrimaryKeyConstraint( 'query_id', 'start_graph_id', 'end_graph_id', 'start', 'end'), ForeignKeyConstraint( ['start_graph_id', 'start'], ['node.graph_id', 'node.node_id'], name="fk_match_start"), ForeignKeyConstraint( ['end_graph_id', 'end'], ['node.graph_id', 'node.node_id'], name="fk_match_end"), ForeignKeyConstraint( ['query_id', 'start_graph_id', 'end_graph_id'], ['query.query_id', 'query.start_graph_id', 'query.end_graph_id'], name="fk_query" ) ) start = Column(BigInteger) end = Column(BigInteger) start_graph_id = Column(Integer) end_graph_id = Column(Integer) query_id = Column(Integer) meta = Column(String, nullable=True) Index('query_id') weight = Column( Float, CheckConstraint("weight>0", name="min_check"), CheckConstraint("weight<=1", name="max_check"), nullable=False ) start_node = relationship( 'Node', primaryjoin="and_(Match.start == Node.node_id, \ Match.start_graph_id == Node.node_id)", backref="start_matches" ) end_node = relationship( 'Node', primaryjoin="and_(Match.end == Node.node_id, \ Match.end_graph_id == Node.graph_id)", backref="end_matches" ) def __repr__(self): return "<Match(start={}, end={}, weight={}, start_graph_id={}, \ end_graph_id={}, query={})>".format( self.start, self.end, self.weight, self.start_graph_id, self.end_graph_id, self.query_id )
[docs]class Node(Base): """Node in a Graph""" __tablename__ = 'node' __table_args__ = ( PrimaryKeyConstraint('graph_id', 'node_id'), ) node_id = Column( BigInteger, CheckConstraint("node_id>=0", name="q_min_id_check") ) graph_id = Column(Integer, ForeignKey("graph.graph_id")) meta = Column(String, nullable=True) Index('graph_id')
[docs] def neighbours(self): return [x.end_node for x in self.start_edges]
def __repr__(self): return "<Node(node_id={}, graph_id={})>".format( self.node_id, self.graph_id)
[docs]class Edge(Base): """Joins Nodes in a Graph""" __tablename__ = 'edge' __table_args__ = ( PrimaryKeyConstraint('graph_id', 'start', 'end'), ForeignKeyConstraint( ['graph_id', 'start'], ['node.graph_id', 'node.node_id'] ), ForeignKeyConstraint( ['graph_id', 'end'], ['node.graph_id', 'node.node_id'] ) ) start = Column(BigInteger) end = Column(BigInteger) graph_id = Column(Integer) meta = Column(String, nullable=True) start_node = relationship( 'Node', primaryjoin='and_(Node.node_id == Edge.start, \ Node.graph_id == Edge.graph_id)', backref='start_edges' ) end_node = relationship( 'Node', primaryjoin='and_(Node.node_id == Edge.end, \ Node.graph_id == Edge.graph_id)', backref='end_edges' ) def __repr__(self): return "<Edge(start={}, end={}, graph_id={})>".format( self.start, self.end, self.graph_id)