multiple times, with a common column shared between the
joins.
+ - removed ancient assertion that mapped selectables require
+ "alias names" - the mapper creates its own alias now if
+ none is present. Though in this case you need to use
+ the class, not the mapped selectable, as the source of
+ column attributes - so a warning is still issued.
+
- sql
- Fixed bug with union() when applied to non-Table connected
select statements
raise exceptions.ArgumentError("Invalid setting for with_polymorphic")
else:
self.with_polymorphic = None
-
- check_tables = [self.local_table]
- if self.with_polymorphic:
- check_tables.append(self.with_polymorphic[1])
- for table in check_tables:
- if table and isinstance(table, expression._SelectBaseMixin):
- raise exceptions.ArgumentError("Mapping against a Select object requires that it has a name. Use an alias to give it a name, i.e. s = select(...).alias('myselect')")
+
+ if isinstance(self.local_table, expression._SelectBaseMixin):
+ util.warn("mapper %s creating an alias for the given selectable - use Class attributes for queries." % self)
+ self.local_table = self.local_table.alias()
+
+ if self.with_polymorphic and isinstance(self.with_polymorphic[1], expression._SelectBaseMixin):
+ self.with_polymorphic[1] = self.with_polymorphic[1].alias()
# our 'polymorphic identity', a string name that when located in a result set row
# indicates this Mapper should be used to construct the object instance for that row.
self._pks_by_table = {}
self._cols_by_table = {}
- all_cols = util.Set(chain(*[c2 for c2 in [col.proxy_set for col in [c for c in self._columntoproperty]]]))
+ all_cols = util.Set(chain(*[col.proxy_set for col in self._columntoproperty]))
pk_cols = util.Set([c for c in all_cols if c.primary_key])
# identify primary key columns which are also mapped by this mapper.
by primary key.
"""
- params = dict([(primary_key, sql.bindparam(None, type_=primary_key.type)) for primary_key in self.primary_key])
- return sql.and_(*[k==v for (k, v) in params.iteritems()]), params
+ params = [(primary_key, sql.bindparam(None, type_=primary_key.type)) for primary_key in self.primary_key]
+ return sql.and_(*[k==v for (k, v) in params]), dict(params)
_get_clause = property(util.cache_decorator(_get_clause))
def _equivalent_columns(self):
properties={
'addresses':relation(Address, backref='email_address')
})
- try:
- class_mapper(Address)
- class_mapper(User)
- assert False
- except exceptions.ArgumentError:
- pass
+ self.assertRaises(exceptions.ArgumentError, compile_mappers)
def test_prop_accessor(self):
mapper(User, users)
- try:
- class_mapper(User).properties
- assert False
- except NotImplementedError, uoe:
- assert str(uoe) == "Public collection of MapperProperty objects is provided by the get_property() and iterate_properties accessors."
+ self.assertRaises(NotImplementedError, getattr, class_mapper(User), 'properties')
def test_badcascade(self):
mapper(Address, addresses)
- try:
- mapper(User, users, properties={'addresses':relation(Address, cascade="fake, all, delete-orphan")})
- assert False
- except exceptions.ArgumentError, e:
- assert str(e) == "Invalid cascade option 'fake'"
+ self.assertRaises(exceptions.ArgumentError, relation, Address, cascade="fake, all, delete-orphan")
def test_columnprefix(self):
mapper(User, users, column_prefix='_', properties={
def test_no_pks(self):
s = select([users.c.user_name]).alias('foo')
- try:
- mapper(User, s)
- assert False
- except exceptions.ArgumentError, e:
- assert "could not assemble any primary key columns for mapped table 'foo'" in str(e)
+ self.assertRaises(exceptions.ArgumentError, mapper, User, s)
def test_recompile_on_othermapper(self):
"""test the global '__new_mappers' flag such that a compile
pass
mapper(Foo, users)
sess = create_session()
- try:
- Foo('one', _sa_session=sess)
- assert False
- except:
- assert len(list(sess)) == 0
- try:
- Foo('one')
- assert False
- except TypeError, e:
- pass
+ self.assertRaises(TypeError, Foo, 'one', _sa_session=sess)
+ assert len(list(sess)) == 0
+ self.assertRaises(TypeError, Foo, 'one')
@testing.uses_deprecated('SessionContext', 'SessionContextExt')
def test_constructorexceptions(self):
def test_no_tables(self):
class Subset(object):
pass
- selectable = select(["x", "y", "z"]).alias('foo')
- try:
- mapper(Subset, selectable)
- compile_mappers()
- assert False
- except exceptions.InvalidRequestError, e:
- assert str(e) == "Could not find any Table objects in mapped table 'SELECT x, y, z'", str(e)
+ selectable = select(["x", "y", "z"])
+ self.assertRaisesMessage(exceptions.InvalidRequestError, "Could not find any Table objects", mapper, Subset, selectable)
+ @testing.emits_warning('.*creating an Alias.*')
def test_basic(self):
class Subset(Base):
pass
- subset_select = select([common_table.c.id, common_table.c.data]).alias('subset')
+ subset_select = select([common_table.c.id, common_table.c.data])
subset_mapper = mapper(Subset, subset_select)
sess = create_session(bind=testing.db)
sess.flush()
sess.clear()
- assert [Subset(data=1)] == sess.query(Subset).all()
+ self.assertEquals(sess.query(Subset).all(), [Subset(data=1)])
+ self.assertEquals(sess.query(Subset).filter(Subset.data==1).one(), Subset(data=1))
+ self.assertEquals(sess.query(Subset).filter(Subset.data!=1).first(), None)
+
+ subset_select = class_mapper(Subset).mapped_table
+ self.assertEquals(sess.query(Subset).filter(subset_select.c.data==1).one(), Subset(data=1))
+
# TODO: more tests mapping to selects
if __name__ == '__main__':
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
