ugh can you attach a zipfile please, they came out inline
On Nov 6, 2007, at 5:46 PM, [EMAIL PROTECTED] wrote: > 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]' >> >> > > > > 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 --~--~---------~--~----~------------~-------~--~----~ You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To post to this group, send email to sqlalchemy@googlegroups.com 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 -~----------~----~----~----~------~----~------~--~---