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 [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
-~----------~----~----~----~------~----~------~--~---
