sorry, here the files
> hi.
> i have somewhat messy setup (~test case), about association with
> intermediate table/class, double pointing to one side and single
> pointing to another. i do set up both A-links in one item; and set up
> only first in another item, the other link (a2_link) is pre-set to None.
> And, i have the error below since r3695.
> The error seems to disappear if i do not explicitly initiate the a2_link
> to None - dont touch it or set to some object.
> Any idea what's wrong?
> ...
> File "/home/az/src/dbcook/sqlalchemy/orm/sync.py", line 91, in execute
> rule.execute(source, dest, obj, child, clearkeys)
> File "/home/az/src/dbcook/sqlalchemy/orm/sync.py", line 139, in execute
> raise exceptions.AssertionError("Dependency rule tried to blank-out
> primary key column '%s' on instance '%s'" % (str(self.dest_column),
> mapperutil.instance_str(dest)))
> sqlalchemy.exceptions.AssertionError: Dependency rule tried to blank-out
> primary key column 'IntermediateAB.a2_link_id' on instance
> '[EMAIL PROTECTED]'
>
>
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected]
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
from sa_gentestbase import *
setup()
Base.__repr__ = Base.__str__
#===============
#PYTHONPATH=`pwd`/..:/home/az/src/hor-trunk python mapper/relation.py -v generate
#config: db=, debug=, default_lazy=False, echo=False, force_lazy=False, generate=True, log_sa=, lower_pu=True
#========= generated SA set-up
t = Test_SA( 'setUp' )
t.setUp()
meta = t.meta
table_A = Table( 'A', meta,
Column( 'name', String, ),
Column( 'db_id', primary_key= True, type_= Integer, ),
)
table_B = Table( 'B', meta,
Column( 'name', String, ),
Column( 'db_id', primary_key= True, type_= Integer, ),
)
table_IntermediateAB = Table( 'IntermediateAB', meta,
Column( 'color', String, ),
Column( 'a_link_id', Integer, ForeignKey( 'A.db_id', ), nullable= False, primary_key= True, ),
Column( 'a2_link_id', Integer, ForeignKey( 'A.db_id', ), nullable= True, primary_key= True, ),
Column( 'b_boza_id', Integer, ForeignKey( 'B.db_id', ), nullable= False, primary_key= True, ),
)
meta.create_all()
class A( Base):
props = ['db_id', 'name']
class B( Base):
props = ['db_id', 'name']
class IntermediateAB( Base):
props = ['db_id', 'color', 'a_link', 'a2_link', 'b_boza']
mapper_A = mapper( A, table_A,)
mapper_A.add_property( 'all_ab', relation( IntermediateAB,
#collection_class= <bound method type._CollectionFactory of <class '__main__.IntermediateAB'>>,
lazy= True,
primaryjoin= table_IntermediateAB.c.a_link_id == table_A.c.db_id,
remote_side= table_IntermediateAB.c.a_link_id,
uselist= True,
) )
mapper_B = mapper( B, table_B,)
mapper_IntermediateAB = mapper( IntermediateAB, table_IntermediateAB, allow_null_pks=True)
mapper_IntermediateAB.add_property( 'a_link', relation( A,
foreign_keys= table_IntermediateAB.c.a_link_id,
lazy= False,
primaryjoin= table_IntermediateAB.c.a_link_id == table_A.c.db_id,
remote_side= table_A.c.db_id,
uselist= False,
) )
mapper_IntermediateAB.add_property( 'a2_link', relation( A,
foreign_keys= table_IntermediateAB.c.a2_link_id,
lazy= False,
primaryjoin= table_IntermediateAB.c.a2_link_id == table_A.c.db_id,
remote_side= table_A.c.db_id,
uselist= False,
) )
mapper_IntermediateAB.add_property( 'b_boza', relation( B,
foreign_keys= table_IntermediateAB.c.b_boza_id,
lazy= False,
primaryjoin= table_IntermediateAB.c.b_boza_id == table_B.c.db_id,
remote_side= table_B.c.db_id,
uselist= False,
) )
####################
#############
a = A( name= 'a1')
a3 = A( name= 'a3')
b1 = B( name= 'b1' )
b2 = B( name= 'b2' )
a.all_ab.append( IntermediateAB( b_boza= b1, color='green', a2_link = a3) )
a.all_ab.append( IntermediateAB( b_boza= b2, color='rrrr',
a2_link = a3 #XXX
) )
print '1111', [i.b_boza for i in a.all_ab]
s= create_session()
for z in locals().values():
if isinstance( z, Base): s.save(z)
s.flush()
s.clear()
for t in table_B, table_A, table_IntermediateAB:
print t,':', str( list( t.select().execute() ))
s= create_session()
aa = s.query( A ).filter_by(name='a1').first()
print aa
print 'xxxxxxx'
l = [i.b_boza for i in aa.all_ab]
print '2222', l
assert len(l) ==2
#expected output:
#1111 [B/id=None( name=b1 ), B/id=None( name=b2 )]
#=== whole database:
#<class '__main__.B'> : [(u'b1', 1), (u'b2', 2)]
#<class '__main__.A'> : [(None, u'a3', 1), (None, u'a1', 2)]
#<class '__main__.IntermediateAB'> : [(None, u'green', 2, 1, 1), (None, u'rrrr', 2, None, 2)]
#A/id=2( name=a1 )
#xxxxxxx
#2222 [B/id=1( name=b1 ), B/id=2( name=b2 )]
#$Id: sa_gentestbase.py 127 2007-10-26 09:46:54Z svilen_dobrev $
from sqlalchemy import *
from sqlalchemy.orm import *
#see baseobj.py
class Base( object):
'with __init__(kwargs) and nice/non-recursive str()'
def __init__( me, **kargs):
for k,v in kargs.iteritems(): setattr( me, k, v)
props = [ 'id' ] #[...]
props4ref = props + [ 'name' ]
def __str__( obj, props =None):
klas = obj.__class__
r = klas.__name__ + '('
for k in props or klas.props:
v = getattr( obj, k, '<notset>')
if isinstance( v, Base):
v = '>' + v.__str__( klas.props4ref)
r += ' '+k+'='+str(v)
return r+' )'
class config:
echo = False
dump = False
debug = False
log_sa = False
session_clear = True
reuse_db = False
leak = False
memory = False
db = 'sqlite:///:memory:'
repeat = 1
#_mem = ''
def memusage():
import os
pid = os.getpid()
m = ''
for l in file( '/proc/%(pid)s/status' % locals() ):
l = l.strip()
for k in 'VmPeak VmRSS VmData'.split():
if l.startswith(k):
m += '; '+l
if m: print m
# global _mem
# if _mem != m:
# _mem = m
# print m
import unittest
class Test_SA( unittest.TestCase):
_db = None
def setUp(me):
if config.debug or config.echo:
print '=====', me.id()
if config.reuse_db and me._db:
db = me._db
else:
db = create_engine( config.db)
if config.reuse_db:
me._db = db
format ='* SA: %(levelname)s %(message)s'
#plz no timestamps!
if config.log_sa:
import logging
logging.basicConfig( level=logging.DEBUG, format=format, stream =logging.sys.stdout)
logging.getLogger( 'sqlalchemy').setLevel( logging.DEBUG) #debug EVERYTHING!
me.db = db
db.echo = config.echo
me.meta = MetaData( db)
def tearDown(me):
me.meta.drop_all()
me.meta = None
#destroy ALL caches
clear_mappers()
if not config.reuse_db:
me.db.dispose()
me.db = None
if config.leak:
import gc
gc.set_debug( gc.DEBUG_UNCOLLECTABLE | gc.DEBUG_SAVEALL | gc.DEBUG_INSTANCES | gc.DEBUG_STATS ) #OBJECTS
gc.collect()
import sqlalchemy
print "MAPPER REG:", len( dict(sqlalchemy.orm.mapperlib.mapper_registry) )
print "SESION REG:", len( dict(sqlalchemy.orm.session._sessions) )
print "CLASSKEYS:", len( dict(sqlalchemy.util.ArgSingleton.instances) )
i = 0
for x in gc.get_objects():
if isinstance(x, sqlalchemy.orm.mapperlib.Mapper) or isinstance(x, MetaData):
i+=1
#print x
print 'gc/SA objects:', i
def query( me, session, expects, idname ='id'):
if config.debug:
print 'items:'
for item in expects:
print item['exp_single']
if config.dump:
print 'tables:'
for item in expects:
for x in item['table'].select().execute():
print item['table'], ':', x
for item in expects:
me.query1( session, idname=idname, **item)
def query1( me, session, idname, klas, table, oid, exp_single, exp_multi):
if config.session_clear: session.clear()
#single
q = session.query( klas).filter_by( **{idname: oid}).first()
me.assertEqual( exp_single, str(q),
klas.__name__+'.getby_'+idname+'():\n result= %(q)s\n expect= %(exp_single)s' % locals()
)
if config.session_clear: session.clear()
#multiple
q = session.query( klas)
x = [ str(z) for z in q ]
x.sort()
exp_multi.sort()
me.assertEqual( exp_multi, x,
klas.__name__+'.select():\n result= %(x)s\n expect= %(exp_multi)s' % locals()
)
def run( self, *a, **k):
for i in range( config.repeat):
unittest.TestCase.run( self, *a,**k)
if config.memory: memusage()
help = 'echo dump debug log_sa no_session_clear reuse_db leak memory'
def setup():
import sys
# sys.setrecursionlimit( 600)
for h in ['help', '-h', '--help']:
if h in sys.argv:
print 'options:', help
for k in help.split():
v = k in sys.argv
if v: sys.argv.remove(k)
if k.startswith('no_'):
k = k[3:]
v = not v
setattr( config, k, v)
for a in sys.argv[1:]:
kv = a.split('=')
if len(kv)==2:
k,v = kv
if k=='db':
config.db = v
elif k=='repeat':
config.repeat = int(v)
else: continue
sys.argv.remove(a)
print 'config:', ', '.join( '%s=%s' % (k,v) for k,v in config.__dict__.iteritems() if not k.startswith('__') )
# vim:ts=4:sw=4:expandtab
