| actually, heres both tests passing. however, i am sure this mapping is going to have more problems, its really complicated. one for the test suite for sure if it ever works fully. |
from sqlalchemy import *
import time
import os
metadata = DynamicMetaData()
tables = {}
class Project(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
setattr(self, key, value)
class Version(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
setattr(self, key, value)
self.date = self.date or time.strftime("%Y-%m-%d %H:%M:%S")
class Node(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
setattr(self, key, value)
class Algorithm(Node):
pass
class Parameter(Node):
pass
class VersionKeyMapper(MapperExtension):
def before_insert(self, mapper, connection, instance):
versions_t = mapper.select_table
s = select(
[func.max(versions_t.c.id)+1],
versions_t.c.project_id==instance.project_id
)
r = s.execute()
new_id = r.fetchone()[0]
instance.id = new_id or 1
class Edge(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
setattr(self, key, value)
# Define the tables
tables['projects'] = \
Table('projects', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(40)))
tables['nodes'] = \
Table('nodes', metadata,
Column('project_id', Integer, ForeignKey('projects.id')),
Column('id', Integer, primary_key=True),
Column('type', Integer),
Column('name', String(30), nullable=False),
Column('description', String(80)),
Column('comments', String),
Column('start_version_id', Integer),
Column('end_version_id', Integer),
Column('sub_project_id', Integer, ForeignKey('projects.id')),
Column('sub_project_version_id', Integer),
Column('value', String),
Column('previous_id', Integer, ForeignKey('nodes.id')),
ForeignKeyConstraint(
['project_id','start_version_id'],
['versions.project_id','versions.id']
),
ForeignKeyConstraint(
['project_id','end_version_id'],
['versions.project_id','versions.id']
),
ForeignKeyConstraint(
['sub_project_id','sub_project_version_id'],
['versions.project_id','versions.id'])
)
tables['edges'] = \
Table('edges', metadata,
Column('project_id', Integer, ForeignKey('projects.id')),
Column('id', Integer, primary_key=True),
Column('name', String(30)),
Column('from_node_id', Integer, ForeignKey('nodes.id')),
Column('to_node_id', Integer, ForeignKey('nodes.id')),
Column('start_version_id', Integer, ForeignKey('versions.id')),
Column('end_version_id', Integer, ForeignKey('versions.id')),
Column('previous_id', Integer, ForeignKey('edges.id')))
tables['versions'] = \
Table('versions', metadata,
Column('project_id', Integer,
ForeignKey('projects.id'), primary_key=True),
Column('id', Integer, primary_key=True),
Column('date', DateTime),
Column('comment', String(300)))
Version.mapper = mapper(Version, tables['versions'],
extension=VersionKeyMapper(), batch=False)
Project.mapper = mapper(Project, tables['projects'],
properties = {
'versions' : relation(Version, backref = 'project')#,
})
node_types = {'Node':0, 'Algorithm':1,'Parameter':2}
Node.mapper = mapper(Node, tables['nodes'],
polymorphic_on=tables['nodes'].c.type,
polymorphic_identity=node_types['Node'],
properties = {
'project' : relation(Project,
primaryjoin=
tables['nodes'].c.project_id == Project.c.id,
uselist=False,
foreignkey=Project.c.id,
backref=backref('nodes', primaryjoin=
tables['nodes'].c.project_id
== tables['projects'].c.id,
foreignkey=tables['nodes'].c.project_id)),
'start_version' : relation(Version,
primaryjoin=and_(
tables['nodes'].c.project_id==Version.c.project_id,
tables['nodes'].c.start_version_id==Version.c.id
),backref='start_nodes', viewonly=True),
'end_version' : relation(Version,
primaryjoin=and_(
tables['nodes'].c.project_id==Version.c.project_id,
tables['nodes'].c.end_version_id==Version.c.id
),backref='end_nodes', viewonly=True),
'previous' : relation(Node,
primaryjoin=
tables['nodes'].c.previous_id
== tables['nodes'].c.id,
backref='next',uselist=False)
})
Edge.mapper = mapper(Edge,tables['edges'],
properties = {
'project' : relation(Project,
primaryjoin=
tables['edges'].c.project_id
== Project.c.id,
uselist=False,
foreignkey=Project.c.id),
'start_version' : relation(Version,
primaryjoin=and_(
tables['edges'].c.project_id
== Version.c.project_id,
tables['edges'].c.start_version_id
== Version.c.id
),backref='start_edges', viewonly=True),
'end_version' : relation(Version,
primaryjoin=and_(
tables['edges'].c.project_id
== Version.c.project_id,
tables['edges'].c.end_version_id
== Version.c.id
),backref='end_edges', viewonly=True),
'previous' : relation(Edge,
primaryjoin=tables['edges'].c.previous_id
== tables['edges'].c.id,
backref='next',uselist=False)
})
Node.mapper.add_property('inputs', relation(Edge,
primaryjoin=tables['nodes'].c.id == tables['edges'].c.to_node_id,
backref='to_node'))
Node.mapper.add_property('outputs', relation(Edge,
primaryjoin=tables['nodes'].c.id == tables['edges'].c.from_node_id,
backref='from_node'))
Algorithm.mapper = mapper(Algorithm, inherits=Node.mapper,
polymorphic_identity=node_types['Algorithm'],
properties = {
'sub_project' : relation(Project,
primaryjoin=(tables['nodes'].c.sub_project_id == Project.c.id),
foreignkey=Project.c.id,
uselist=False),
'sub_project_version' : relation(Version,
primaryjoin=and_(
Node.c.sub_project_id == Version.c.project_id,
Node.c.sub_project_version_id == Version.c.id
), backref = 'alg_nodes_using', viewonly=True)
})
Project.mapper.add_property('alg_nodes_using', relation(Algorithm,
primaryjoin=(tables['nodes'].c.sub_project_id == Project.c.id),
foreignkey=tables['nodes'].c.sub_project_id))
Parameter.mapper = mapper(Parameter, inherits=Node.mapper,
polymorphic_identity=node_types['Parameter'])
class RelateDB:
def __init__(self, filename='relate.db', meta=metadata):
self.filename = filename
self.engine = create_engine('sqlite:///%s' % self.filename)
self.metadata = meta
self.metadata.connect(self.engine)
self.session = create_session(bind_to = self.engine)
def create_file(self):
self.metadata.create_all()
def delete_file(self):
try:
os.remove(self.filename)
except OSError: # File didn't exist
pass
def recreate_file(self):
self.delete_file()
self.create_file()
import logging
#logging.basicConfig()
#logging.getLogger('sqlalchemy.orm').setLevel(logging.DEBUG)
#logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
from db import *
import unittest
class TestRelateDB(unittest.TestCase):
def setUp(self):
self.db = RelateDB('test.db')
self.db.recreate_file()
def tearDown(self):
self.db.delete_file()
def test_algorithm(self):
a = Algorithm(name='a1')
self.db.session.save(a)
self.db.session.flush()
print "ITS", self.db.session.query(Node).get_by_name('a1')
assert type(self.db.session.query(Node).get_by_name('a1')) is Algorithm
def test_nodes(self):
algs = [Algorithm(name='a'+str(num)) for num in range(0,7)]
params = [Parameter(name='p'+str(num)) for num in range(0,20)]
for p in params:
print p.name
for a in algs:
self.db.session.save(a)
Edge(from_node=params[0],to_node=algs[0])
Edge(from_node=params[1],to_node=algs[0])
Edge(from_node=params[2],to_node=algs[0])
Edge(from_node=params[1],to_node=algs[1])
Edge(from_node=params[2],to_node=algs[1])
Edge(from_node=params[3],to_node=algs[1])
Edge(from_node=params[4],to_node=algs[1])
Edge(from_node=params[4],to_node=algs[2])
Edge(from_node=params[5],to_node=algs[2])
Edge(from_node=params[6],to_node=algs[2])
Edge(from_node=algs[0],to_node=params[7])
Edge(from_node=algs[0],to_node=params[8])
Edge(from_node=algs[1],to_node=params[9])
Edge(from_node=algs[1],to_node=params[10])
Edge(from_node=algs[2],to_node=params[11])
Edge(from_node=algs[2],to_node=params[12])
Edge(from_node=algs[2],to_node=params[13])
Edge(from_node=params[7],to_node=algs[3])
Edge(from_node=params[8],to_node=algs[3])
Edge(from_node=params[9],to_node=algs[3])
Edge(from_node=params[10],to_node=algs[3])
Edge(from_node=params[4],to_node=algs[4])
Edge(from_node=params[10],to_node=algs[4])
Edge(from_node=params[11],to_node=algs[4])
Edge(from_node=params[12],to_node=algs[4])
Edge(from_node=params[13],to_node=algs[4])
Edge(from_node=algs[3],to_node=params[14])
Edge(from_node=algs[3],to_node=params[15])
Edge(from_node=algs[4],to_node=params[16])
Edge(from_node=params[7],to_node=algs[5])
Edge(from_node=params[14],to_node=algs[5])
Edge(from_node=params[15],to_node=algs[5])
Edge(from_node=params[14],to_node=algs[6])
Edge(from_node=params[15],to_node=algs[6])
Edge(from_node=params[16],to_node=algs[6])
Edge(from_node=params[13],to_node=algs[6])
Edge(from_node=algs[5],to_node=params[17])
Edge(from_node=algs[6],to_node=params[18])
Edge(from_node=algs[6],to_node=params[19])
self.db.session.flush()
a = self.db.session.query(Algorithm).get_by_name('a4')
print [p.__class__ for p in a.inputs]
print [p.from_node.name for p in a.inputs]
print [p.id for p in a.inputs]
assert set([p.from_node.name for p in a.inputs]) \
== set(['p10','p4','p11','p12','p13'])
if __name__ == '__main__':
unittest.main()
On Sep 26, 2006, at 4:29 PM, Hogarty, David A. wrote: I'm getting a strange error when flushing the session after the following sequence of creations: def test_nodes(self):
algs = [Algorithm(name='a'+str(num)) for num in range(0,7)]
1) params = [Parameter(name='p'+str(num)) for num in range(0,20)]
for a in algs:
self.db.session.save(a)
for p in params:
self.db.session.save(p)
Edge(from_node=params[0],to_node=algs[0])
Edge(from_node=params[1],to_node=algs[0])
Edge(from_node=params[2],to_node=algs[0])
Edge(from_node=params[1],to_node=algs[1])
Edge(from_node=params[2],to_node=algs[1])
Edge(from_node=params[3],to_node=algs[1])
Edge(from_node=params[4],to_node=algs[1])
Edge(from_node=params[4],to_node=algs[2])
Edge(from_node=params[5],to_node=algs[2])
Edge(from_node=params[6],to_node=algs[2])
Edge(from_node=algs[0],to_node=params[7])
Edge(from_node=algs[0],to_node=params[8])
Edge(from_node=algs[1],to_node=params[9])
Edge(from_node=algs[1],to_node=params[10])
Edge(from_node=algs[2],to_node=params[11])
Edge(from_node=algs[2],to_node=params[12])
Edge(from_node=algs[2],to_node=params[13])
Edge(from_node=params[7],to_node=algs[3])
Edge(from_node=params[8],to_node=algs[3])
Edge(from_node=params[9],to_node=algs[3])
Edge(from_node=params[10],to_node=algs[3])
Edge(from_node=params[4],to_node=algs[4])
Edge(from_node=params[10],to_node=algs[4])
Edge(from_node=params[11],to_node=algs[4])
Edge(from_node=params[12],to_node=algs[4])
Edge(from_node=params[13],to_node=algs[4])
Edge(from_node=algs[3],to_node=params[14])
Edge(from_node=algs[3],to_node=params[15])
Edge(from_node=algs[4],to_node=params[16])
Edge(from_node=params[7],to_node=algs[5])
Edge(from_node=params[14],to_node=algs[5])
Edge(from_node=params[15],to_node=algs[5])
Edge(from_node=params[14],to_node=algs[6])
Edge(from_node=params[15],to_node=algs[6])
Edge(from_node=params[16],to_node=algs[6])
Edge(from_node=params[13],to_node=algs[6])
Edge(from_node=algs[5],to_node=params[17])
Edge(from_node=algs[6],to_node=params[18])
Edge(from_node=algs[6],to_node=params[19])
self.db.session.flush()
a = self.db.session.query(Algorithm).get_by_name('a4')
assert set([p.name for p in a.inputs]) \
== set(['p10','p4','p11','p12','p13']) At some point, it tries to re-insert one of the Parameter nodes created at 1) above, initially created with 2006-09-26 14:53:59,192 INFO sqlalchemy.engine.base.ComposedSQLEngine.0x190cc66c
INSERT INTO nodes (project_id, type, name, description, comments, start_version
_id, end_version_id, sub_project_id, sub_project_version_id, value, previous_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
2006-09-26 14:53:59,193 INFO sqlalchemy.engine.base.ComposedSQLEngine.0x190cc66c
[None, 2, 'p14', None, None, None, None, None, None, None, None] with the following (now is inserting over the existing primary key, causing an integrity error) SQLError: (IntegrityError) PRIMARY KEY must be unique 'INSERT INTO nodes (projec
t_id, id, type, name, description, comments, start_version_id, end_version_id, s
ub_project_id, sub_project_version_id, value, previous_id) VALUES (?, ?, ?, ?, ?
, ?, ?, ?, ?, ?, ?, ?)' [None, 2, 2, 'p14', None, None, None, None, None, None,
None, None] Attached are two files that are a test case. You need nose to run the test: put them in the same directory and run 'nosetests' (Or manually run the setUp, test_nodes, and tearDown methods of test_db.py:TestRelateDB). On re-running the test case before sending, I found that it has several different possible results. I included four output files from recent runs to demonstrate. All of these runs are with the attached code. Any help would be much appreciated, -Dave H <test.tar.gz> ------------------------------------------------------------------------- Take Surveys. Earn Cash. Influence the Future of IT Join SourceForge.net's Techsay panel and you'll get the chance to share your opinions on IT & business topics through brief surveys -- and earn cash Sqlalchemy-users mailing list |
-------------------------------------------------------------------------
Take Surveys. Earn Cash. Influence the Future of IT
Join SourceForge.net's Techsay panel and you'll get the chance to share your
opinions on IT & business topics through brief surveys -- and earn cash
http://www.techsay.com/default.php?page=join.php&p=sourceforge&CID=DEVDEV
_______________________________________________
Sqlalchemy-users mailing list
Sqlalchemy-users@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/sqlalchemy-users
