def fetchone(self):
row = self.cursor.fetchone()
if row is not None:
+ print repr(row)
return RowProxy(self, row)
else:
return None
instance = self._create(row, identitykey, localmap)
result.append(instance)
else:
+ instance = localmap[identitykey]
for key, prop in self.props.iteritems():
prop.execute(instance, key, row, identitykey, localmap, True)
statement._outerjoin = sql.outerjoin(primarytable, targettable, self.whereclause)
statement.append_from(statement._outerjoin)
statement.append_column(targettable)
+ for key, value in self.mapper.props.iteritems():
+ value.setup(key, self.mapper.selectable, statement)
def execute(self, instance, key, row, identitykey, localmap, isduplicate):
try:
identitykey = self.mapper._identity_key(row)
if not localmap.has_key(identitykey):
subinstance = self.mapper._create(row, identitykey, localmap)
- list.append(subinstance)
+ if subinstance is not None:
+ list.append(subinstance)
+ else:
+ subinstance = localmap[identitykey]
+ for key, prop in self.mapper.props.iteritems():
+ prop.execute(subinstance, key, row, identitykey, localmap, True)
class IdentityMap(dict):
def get_key(self, row, class_, table, selectable):
return (class_, table, tuple([row[column.label] for column in selectable.primary_keys]))
-
_global_identitymap = IdentityMap()
+
+def clear_identity():
+ _global_identitymap.clear()
for c in self.table.columns:
values.append((c, bindparam(c.name)))
else:
+ d = {}
for key, value in parameters.iteritems():
if isinstance(key, schema.Column):
- column = key
+ d[key] = value
else:
- column = self.table.columns[str(key)]
-
- if not isinstance(value, BindParamClause):
- value = bindparam(column.name, value)
-
- values.append((column, value))
+ d[self.table.columns[str(key)]] = value
+ for c in self.table.columns:
+ if d.has_key(c):
+ value = d[c]
+ if not isinstance(value, BindParamClause):
+ value = bindparam(c.name, value)
+ values.append((c, value))
return values
def _engine(self):
def __init__(self, table, parameters = None, **params):
self.table = table
self.select = None
-
self.parameters = parameters
self.engine = self.table._engine()
class Order:
def __repr__(self):
- return "Order: " + repr(self.description) + repr(self.isopen)
+ return "Order: " + repr(self.description) + " " + repr(self.isopen) + " " + repr(getattr(self, 'items', None))
+
+class Item:
+ def __repr__(self):
+ return "Item: " + repr(self.item_name)
class MapperTest(PersistTest):
def setUp(self):
+ mapper.clear_identity()
+
self.users = Table('users', db,
Column('user_id', INT, primary_key = True),
Column('user_name', VARCHAR(20)),
Column('isopen', INT)
)
+ self.orderitems = Table('items', db,
+ Column('item_id', INT, primary_key = True),
+ Column('order_id', INT),
+ Column('item_name', VARCHAR(50))
+ )
+
self.users.build()
self.users.insert().execute(user_id = 7, user_name = 'jack')
self.users.insert().execute(user_id = 8, user_name = 'ed')
self.orders.build()
self.orders.insert().execute(order_id = 1, user_id = 7, description = 'order 1', isopen=0)
- self.orders.insert().execute(order_id = 2, user_id = 7, description = 'order 2', isopen=0)
+ self.orders.insert().execute(order_id = 2, user_id = 9, description = 'order 2', isopen=0)
self.orders.insert().execute(order_id = 3, user_id = 7, description = 'order 3', isopen=1)
- self.orders.insert().execute(order_id = 4, user_id = 7, description = 'order 4', isopen=1)
+ self.orders.insert().execute(order_id = 4, user_id = 9, description = 'order 4', isopen=1)
self.orders.insert().execute(order_id = 5, user_id = 7, description = 'order 5', isopen=0)
+ self.orderitems.build()
+ self.orderitems.insert().execute(item_id=1, order_id=2, item_name='item 1')
+ self.orderitems.insert().execute(item_id=3, order_id=3, item_name='item 3')
+ self.orderitems.insert().execute(item_id=2, order_id=2, item_name='item 2')
+ self.orderitems.insert().execute(item_id=5, order_id=3, item_name='item 5')
+ self.orderitems.insert().execute(item_id=4, order_id=3, item_name='item 4')
def testmapper(self):
m = mapper.Mapper(User, self.users)
def testmultieager(self):
m = mapper.Mapper(User, self.users, properties = dict(
addresses = mapper.EagerLoader(mapper.Mapper(Address, self.addresses), self.users.c.user_id==self.addresses.c.user_id),
- orders = mapper.EagerLoader(mapper.Mapper(Order, self.orders), and_(self.orders.c.isopen == 1, self.users.c.user_id==self.orders.c.user_id)),
+ orders = mapper.EagerLoader(mapper.Mapper(Order, self.orders), self.users.c.user_id==self.orders.c.user_id),
), identitymap = mapper.IdentityMap())
l = m.select()
print repr(l)
-# return
+
+ def testdoubleeager(self):
openorders = alias(self.orders, 'openorders')
closedorders = alias(self.orders, 'closedorders')
m = mapper.Mapper(User, self.users, properties = dict(
l = m.select()
print repr(l)
+ def testnestedeager(self):
+ ordermapper = mapper.Mapper(Order, self.orders, properties = dict(
+ items = mapper.EagerLoader(mapper.Mapper(Item, self.orderitems), self.orders.c.order_id == self.orderitems.c.order_id)
+ ))
+
+ m = mapper.Mapper(User, self.users, properties = dict(
+ addresses = mapper.EagerLoader(mapper.Mapper(Address, self.addresses), self.users.c.user_id==self.addresses.c.user_id),
+ orders = mapper.EagerLoader(ordermapper, self.users.c.user_id==self.orders.c.user_id),
+ ))
+ l = m.select()
+ print repr(l)
def tearDown(self):
self.users.drop()
self.addresses.drop()
self.orders.drop()
+ self.orderitems.drop()
pass
if __name__ == "__main__":
# provided as strings
self.runtest(
insert(self.table, dict(id = 3, name = 'jack')),
- ""
+ "INSERT INTO mytable (myid, name) VALUES (:myid, :name)"
)
# insert with a subselect provided
- self.runtest(
- insert(self.table, select([self.table2])),
- ""
- )
+ #self.runtest(
+ # insert(self.table, select([self.table2])),
+ # ""
+ #)
def testupdate(self):
self.runtest(update(self.table, self.table.c.id == 7), "UPDATE mytable SET name=:name WHERE mytable.myid = :mytable_myid", params = {self.table.c.name:'fred'})
def testdelete(self):
self.runtest(delete(self.table, self.table.c.id == 7), "DELETE FROM mytable WHERE mytable.myid = :mytable_myid")
- def footestupdate(self):
- self.runtest(
- update(self.table, {self.table.c.id : select([self.table2.c.id], self.table2.c.name == 'jack')})
- )
-
- def footestonetomany(self):
- return
-
- table1 = Table('users',
- Column('user_id',3, key='id'),
- Column('user_name', 4, key='name'),
- Column('user_email', 4, key='desc'),
- )
-
- table2 = Table('orders',
- Column('order_id',3, key='id'),
- Column('user_id', 4),
- Column('information', 4),
- )
-
- Relation(table1, table2, table2.c.user_id == table1.c.id, lazy = False)
-
- self.runtest(
- table1.select(includerelations = True, use_labels = True),
- "SELECT users.user_id, users.user_name, users.user_email, orders.order_id, orders.user_id, orders.information FROM users LEFT OUTER JOIN orders ON orders.user_id = users.user_id"
- )
-
- def footestmanytomany(self):
- return
-
- table2 = Table('clubs',
- Column('club_id',3, key='id'),
- Column('club_name', 4, key='name'),
- Column('club_description', 4, key='description'),
- )
-
- table1 = Table('users',
- Column('user_id',3, key='id'),
- Column('user_name', 4, key='name'),
- Column('user_email', 4, key='desc'),
- )
-
- table3 = Table('user_clubs',
- Column('user_id', 3),
- Column('club_id', 4)
- )
-
- sq = join(table2, table3, table2.c.id==table3.c.user_id)
-
- Relation(table1, table2, table1.c.id==table3.c.user_id,
- association = Relation(table3, table2, table3.c.club_id==table2.c.id),
- lazy = False)
-
- self.runtest(
- table1.select(includerelations = True),
- ""
- )
-
-
- def donttesttableselect(self):
-
- print select(
- [table2, sq],
- sq.c.name == table2.c.name
- )
-
-
-
-
- print table.select(table.c.id == 'hithere').dump()
-
- print table.select(and_(table2.c.id == table.c.id, table2.c.id == 'hilar')).dump()
-
- print table.select(and_(table.c.id < 5, table.c.name == 'hilar')).dump()
-
-
-
- print repr(table.impl.params)
-
- print table.select(None).dump()
-
- u2 = alias(table, 'u2')
-
def runtest(self, clause, result, engine = None, params = None):
c = clause.compile(engine, params)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
