- count() function on selectables now uses table primary key or
first column instead of "1" for criterion, also uses label "rowcount"
instead of "count".
+- got rudimental "mapping to multiple tables" functionality cleaned up,
+more correctly documented
0.2.1
- "pool" argument to create_engine() properly propigates
# map to it - the identity of an AddressUser object will be
# based on (user_id, address_id) since those are the primary keys involved
- m = mapper(AddressUser, j)
+ m = mapper(AddressUser, j, properties={
+ 'user_id':[users_table.c.user_id, addresses_table.c.user_id]
+ })
+
+A second example:
- A second example:
{python}
# many-to-many join on an association table
j = join(users_table, userkeywords,
# map to it - the identity of a KeywordUser object will be
# (user_id, keyword_id) since those are the primary keys involved
- m = mapper(KeywordUser, j)
+ m = mapper(KeywordUser, j, properties={
+ 'user_id':[users_table.c.user_id, userkeywords.c.user_id],
+ 'keyword_id':[userkeywords.c.keyword_id, keywords.c.keyword_id]
+ })
+
+In both examples above, "composite" columns were added as properties to the mappers; these are aggregations of multiple columns into one mapper property, which instructs the mapper to keep both of those columns set at the same value.
### Mapping a Class against Arbitary Selects {@name=selects}
l = self.pks_by_table[t]
except KeyError:
l = self.pks_by_table.setdefault(t, util.HashSet(ordered=True))
- if not len(t.primary_key):
- raise exceptions.ArgumentError("Table " + t.name + " has no primary key columns. Specify primary_key argument to mapper.")
for k in t.primary_key:
l.append(k)
-
+
+ if len(self.pks_by_table[self.mapped_table]) == 0:
+ raise exceptions.ArgumentError("Could not assemble any primary key columsn from given tables for table '%s'" % (self.mapped_table.name))
+
# make table columns addressable via the mapper
self.columns = util.OrderedProperties()
self.c = self.columns
list."""
#print "SAVE_OBJ MAPPER", self.class_.__name__, objects
connection = uow.transaction.connection(self)
- for table in self.tables:
+ for table in self.tables.sort(reverse=False):
#print "SAVE_OBJ table ", self.class_.__name__, table.name
# looping through our set of tables, which are all "real" tables, as opposed
# to our main table which might be a select statement or something non-writeable
connection = uow.transaction.connection(self)
#print "DELETE_OBJ MAPPER", self.class_.__name__, objects
- for table in util.reversed(self.tables):
+ for table in self.tables.sort(reverse=True):
if not self._has_pks(table):
continue
delete = []
self.tables = []
def add(self, table):
self.tables.append(table)
- def sort(self, reverse=False ):
+ if hasattr(self, '_sorted'):
+ del self._sorted
+ def sort(self, reverse=False):
+ try:
+ sorted = self._sorted
+ except AttributeError, e:
+ self._sorted = self._do_sort()
+ return self.sort(reverse=reverse)
+ if reverse:
+ x = sorted[:]
+ x.reverse()
+ return x
+ else:
+ return sorted
+
+ def _do_sort(self):
import sqlalchemy.orm.topological
tuples = []
class TVisitor(schema.SchemaVisitor):
to_sequence( child )
if head is not None:
to_sequence( head )
- if reverse:
- sequence.reverse()
return sequence
def testunicodeget(self):
"""tests that Query.get properly sets up the type for the bind parameter. using unicode would normally fail
on postgres, mysql and oracle unless it is converted to an encoded string"""
- table = Table('foo', db,
+ metadata = BoundMetaData(db)
+ table = Table('foo', metadata,
Column('id', Unicode(10), primary_key=True),
Column('data', Unicode(40)))
try:
"""tests a save of an object where each instance spans two tables. also tests
redefinition of the keynames for the column properties."""
usersaddresses = sql.join(users, addresses, users.c.user_id == addresses.c.user_id)
- print usersaddresses.corresponding_column(users.c.user_id)
- print repr(usersaddresses._orig_cols)
m = mapper(User, usersaddresses,
properties = dict(
email = addresses.c.email_address,
u.email = 'multi@test.org'
ctx.current.flush()
+ id = m.identity(u)
+ print id
+ ctx.current.clear()
+
+ u = m.get(id)
+ assert u.user_name == 'multitester'
+
usertable = users.select(users.c.user_id.in_(u.foo_id)).execute().fetchall()
self.assertEqual(usertable[0].values(), [u.foo_id, 'multitester'])
addresstable = addresses.select(addresses.c.address_id.in_(u.address_id)).execute().fetchall()
addresstable = addresses.select(addresses.c.address_id.in_(u.address_id)).execute().fetchall()
self.assertEqual(addresstable[0].values(), [u.address_id, u.foo_id, 'lala@hey.com'])
- u = m.select(users.c.user_id==u.foo_id)[0]
- self.echo( repr(u.__dict__))
+ ctx.current.clear()
+ u = m.get(id)
+ assert u.user_name == 'imnew'
+
+ def testm2mmultitable(self):
+ # many-to-many join on an association table
+ j = join(users, userkeywords,
+ users.c.user_id==userkeywords.c.user_id).join(keywords,
+ userkeywords.c.keyword_id==keywords.c.keyword_id)
+
+ # a class
+ class KeywordUser(object):
+ pass
+ # map to it - the identity of a KeywordUser object will be
+ # (user_id, keyword_id) since those are the primary keys involved
+ m = mapper(KeywordUser, j, properties={
+ 'user_id':[users.c.user_id, userkeywords.c.user_id],
+ 'keyword_id':[userkeywords.c.keyword_id, keywords.c.keyword_id],
+ 'keyword_name':keywords.c.name
+
+ })
+
+ k = KeywordUser()
+ k.user_name = 'keyworduser'
+ k.keyword_name = 'a keyword'
+ ctx.current.flush()
+ print m.instance_key(k)
+ id = (k.user_id, k.keyword_id)
+ ctx.current.clear()
+ k = ctx.current.query(KeywordUser).get(id)
+ assert k.user_name == 'keyworduser'
+ assert k.keyword_name == 'a keyword'
+
def testonetoone(self):
m = mapper(User, users, properties = dict(
address = relation(mapper(Address, addresses), lazy = True, uselist = False)
import os
import testbase
-__all__ = ['db', 'users', 'addresses', 'orders', 'orderitems', 'keywords', 'itemkeywords',
+__all__ = ['db', 'users', 'addresses', 'orders', 'orderitems', 'keywords', 'itemkeywords', 'userkeywords',
'User', 'Address', 'Order', 'Item', 'Keyword'
]
)
+userkeywords = Table('userkeywords', metadata,
+ Column('user_id', INT, ForeignKey("users")),
+ Column('keyword_id', INT, ForeignKey("keywords")),
+)
+
itemkeywords = Table('itemkeywords', metadata,
Column('item_id', INT, ForeignKey("items")),
Column('keyword_id', INT, ForeignKey("keywords")),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
