"""General mapper operations with an emphasis on selecting/loading.""" from test.lib.testing import assert_raises, assert_raises_message import sqlalchemy as sa from test.lib import testing from sqlalchemy import MetaData, Integer, String, ForeignKey, func, util from test.lib.schema import Table, Column from sqlalchemy.engine import default from sqlalchemy.orm import mapper, relationship, backref, \ create_session, class_mapper, configure_mappers, reconstructor, \ validates, aliased, defer, deferred, synonym, attributes, \ column_property, composite, dynamic_loader, \ comparable_property, Session from sqlalchemy.orm.persistence import _sort_states from test.lib.testing import eq_, AssertsCompiledSQL from test.lib import fixtures from test.orm import _fixtures from test.lib.assertsql import CompiledSQL import logging class MapperTest(_fixtures.FixtureTest, AssertsCompiledSQL): __dialect__ = 'default' def test_prop_shadow(self): """A backref name may not shadow an existing property name.""" Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(Address, addresses) mapper(User, users, properties={ 'addresses':relationship(Address, backref='email_address') }) assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers) def test_update_attr_keys(self): """test that update()/insert() use the correct key when given InstrumentedAttributes.""" User, users = self.classes.User, self.tables.users mapper(User, users, properties={ 'foobar':users.c.name }) users.insert().values({User.foobar:'name1'}).execute() eq_(sa.select([User.foobar]).where(User.foobar=='name1').execute().fetchall(), [('name1',)]) users.update().values({User.foobar:User.foobar + 'foo'}).execute() eq_(sa.select([User.foobar]).where(User.foobar=='name1foo').execute().fetchall(), [('name1foo',)]) def test_utils(self): users = self.tables.users from sqlalchemy.orm.util import _is_mapped_class, _is_aliased_class class Foo(object): x = "something" @property def y(self): return "somethign else" m = mapper(Foo, users) a1 = aliased(Foo) f = Foo() for fn, arg, ret in [ (_is_mapped_class, Foo.x, False), (_is_mapped_class, Foo.y, False), (_is_mapped_class, Foo, True), (_is_mapped_class, f, False), (_is_mapped_class, a1, True), (_is_mapped_class, m, True), (_is_aliased_class, a1, True), (_is_aliased_class, Foo.x, False), (_is_aliased_class, Foo.y, False), (_is_aliased_class, Foo, False), (_is_aliased_class, f, False), (_is_aliased_class, a1, True), (_is_aliased_class, m, False), ]: assert fn(arg) == ret def test_prop_accessor(self): users, User = self.tables.users, self.classes.User mapper(User, users) assert_raises(NotImplementedError, getattr, sa.orm.class_mapper(User), 'properties') def test_friendly_attribute_str_on_uncompiled_boom(self): User, users = self.classes.User, self.tables.users def boom(): raise Exception("it broke") mapper(User, users, properties={ 'addresses':relationship(boom) }) # test that QueryableAttribute.__str__() doesn't # cause a compile. eq_(str(User.addresses), "User.addresses") def test_exceptions_sticky(self): """test preservation of mapper compile errors raised during hasattr(), as well as for redundant mapper compile calls. Test that repeated calls don't stack up error messages. """ Address, addresses, User = (self.classes.Address, self.tables.addresses, self.classes.User) mapper(Address, addresses, properties={ 'user':relationship(User) }) try: hasattr(Address.user, 'property') except sa.orm.exc.UnmappedClassError: assert util.compat.py32 for i in range(3): assert_raises_message(sa.exc.InvalidRequestError, "^One or more mappers failed to " "initialize - can't proceed with " "initialization of other mappers. " "Original exception was: Class " "'test.orm._fixtures.User' is not mapped$" , configure_mappers) def test_column_prefix(self): users, User = self.tables.users, self.classes.User mapper(User, users, column_prefix='_', properties={ 'user_name': synonym('_name') }) s = create_session() u = s.query(User).get(7) eq_(u._name, 'jack') eq_(u._id,7) u2 = s.query(User).filter_by(user_name='jack').one() assert u is u2 def test_no_pks_1(self): User, users = self.classes.User, self.tables.users s = sa.select([users.c.name]).alias('foo') assert_raises(sa.exc.ArgumentError, mapper, User, s) def test_no_pks_2(self): User, users = self.classes.User, self.tables.users s = sa.select([users.c.name]).alias() assert_raises(sa.exc.ArgumentError, mapper, User, s) def test_reconfigure_on_other_mapper(self): """A configure trigger on an already-configured mapper still triggers a check against all mappers.""" users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) mapper(User, users) sa.orm.configure_mappers() assert sa.orm.mapperlib._new_mappers is False m = mapper(Address, addresses, properties={ 'user': relationship(User, backref="addresses")}) assert m.configured is False assert sa.orm.mapperlib._new_mappers is True u = User() assert User.addresses assert sa.orm.mapperlib._new_mappers is False def test_configure_on_session(self): User, users = self.classes.User, self.tables.users m = mapper(User, users) session = create_session() session.connection(m) def test_incomplete_columns(self): """Loading from a select which does not contain all columns""" addresses, Address = self.tables.addresses, self.classes.Address mapper(Address, addresses) s = create_session() a = s.query(Address).from_statement( sa.select([addresses.c.id, addresses.c.user_id])).first() eq_(a.user_id, 7) eq_(a.id, 1) # email address auto-defers assert 'email_addres' not in a.__dict__ eq_(a.email_address, 'jack@bean.com') def test_column_not_present(self): users, addresses, User = (self.tables.users, self.tables.addresses, self.classes.User) assert_raises_message(sa.exc.ArgumentError, "not represented in the mapper's table", mapper, User, users, properties={'foo' : addresses.c.user_id}) def test_constructor_exc(self): """TypeError is raised for illegal constructor args, whether or not explicit __init__ is present [ticket:908].""" users, addresses = self.tables.users, self.tables.addresses class Foo(object): def __init__(self): pass class Bar(object): pass mapper(Foo, users) mapper(Bar, addresses) assert_raises(TypeError, Foo, x=5) assert_raises(TypeError, Bar, x=5) def test_sort_states_comparisons(self): """test that _sort_states() doesn't compare insert_order to state.key, for set of mixed persistent/pending. In particular Python 3 disallows this. """ class Foo(object): def __init__(self, id): self.id = id m = MetaData() foo_t = Table('foo', m, Column('id', String, primary_key=True) ) m = mapper(Foo, foo_t) class DontCompareMeToString(int): # Py3K # pass # Py2K def __lt__(self, other): assert not isinstance(other, basestring) return int(self) < other # end Py2K foos = [Foo(id='f%d' % i) for i in range(5)] states = [attributes.instance_state(f) for f in foos] for s in states[0:3]: s.key = m._identity_key_from_state(s) states[3].insert_order = DontCompareMeToString(5) states[4].insert_order = DontCompareMeToString(1) states[2].insert_order = DontCompareMeToString(3) eq_( _sort_states(states), [states[4], states[3], states[0], states[1], states[2]] ) def test_props(self): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) m = mapper(User, users, properties = { 'addresses' : relationship(mapper(Address, addresses)) }) assert User.addresses.property is m.get_property('addresses') def test_configure_on_prop_1(self): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) mapper(User, users, properties = { 'addresses' : relationship(mapper(Address, addresses)) }) User.addresses.any(Address.email_address=='foo@bar.com') def test_configure_on_prop_2(self): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) mapper(User, users, properties = { 'addresses' : relationship(mapper(Address, addresses)) }) eq_(str(User.id == 3), str(users.c.id==3)) def test_configure_on_prop_3(self): users, addresses, User = (self.tables.users, self.tables.addresses, self.classes.User) class Foo(User):pass mapper(User, users) mapper(Foo, addresses, inherits=User) assert getattr(Foo().__class__, 'name').impl is not None def test_deferred_subclass_attribute_instrument(self): users, addresses, User = (self.tables.users, self.tables.addresses, self.classes.User) class Foo(User):pass mapper(User, users) configure_mappers() mapper(Foo, addresses, inherits=User) assert getattr(Foo().__class__, 'name').impl is not None def test_configure_on_get_props_1(self): User, users = self.classes.User, self.tables.users m =mapper(User, users) assert not m.configured assert list(m.iterate_properties) assert m.configured def test_configure_on_get_props_2(self): User, users = self.classes.User, self.tables.users m= mapper(User, users) assert not m.configured assert m.get_property('name') assert m.configured def test_configure_on_get_props_3(self): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) m= mapper(User, users) assert not m.configured configure_mappers() m2 = mapper(Address, addresses, properties={ 'user':relationship(User, backref='addresses') }) assert m.get_property('addresses') def test_add_property(self): users, addresses, Address = (self.tables.users, self.tables.addresses, self.classes.Address) assert_col = [] class User(fixtures.ComparableEntity): def _get_name(self): assert_col.append(('get', self._name)) return self._name def _set_name(self, name): assert_col.append(('set', name)) self._name = name name = property(_get_name, _set_name) def _uc_name(self): if self._name is None: return None return self._name.upper() uc_name = property(_uc_name) uc_name2 = property(_uc_name) m = mapper(User, users) mapper(Address, addresses) class UCComparator(sa.orm.PropComparator): __hash__ = None def __eq__(self, other): cls = self.prop.parent.class_ col = getattr(cls, 'name') if other is None: return col == None else: return sa.func.upper(col) == sa.func.upper(other) m.add_property('_name', deferred(users.c.name)) m.add_property('name', synonym('_name')) m.add_property('addresses', relationship(Address)) m.add_property('uc_name', sa.orm.comparable_property(UCComparator)) m.add_property('uc_name2', sa.orm.comparable_property( UCComparator, User.uc_name2)) sess = create_session(autocommit=False) assert sess.query(User).get(7) u = sess.query(User).filter_by(name='jack').one() def go(): eq_(len(u.addresses), len(self.static.user_address_result[0].addresses)) eq_(u.name, 'jack') eq_(u.uc_name, 'JACK') eq_(u.uc_name2, 'JACK') eq_(assert_col, [('get', 'jack')], str(assert_col)) self.sql_count_(2, go) u.name = 'ed' u3 = User() u3.name = 'some user' sess.add(u3) sess.flush() sess.rollback() def test_add_prop_via_backref_resets_memoizations_reconfigures(self): users, User = self.tables.users, self.classes.User addresses, Address = self.tables.addresses, self.classes.Address m1 = mapper(User, users) User() m2 = mapper(Address, addresses, properties={ 'user':relationship(User, backref="addresses") }) # configure mappers takes place when User is generated User() assert hasattr(User, 'addresses') assert "addresses" in [p.key for p in m1._polymorphic_properties] def test_replace_property(self): users, User = self.tables.users, self.classes.User m = mapper(User, users) m.add_property('_name',users.c.name) m.add_property('name', synonym('_name')) sess = create_session() u = sess.query(User).filter_by(name='jack').one() eq_(u._name, 'jack') eq_(u.name, 'jack') u.name = 'jacko' assert m._columntoproperty[users.c.name] is m.get_property('_name') sa.orm.clear_mappers() m = mapper(User, users) m.add_property('name', synonym('_name', map_column=True)) sess.expunge_all() u = sess.query(User).filter_by(name='jack').one() eq_(u._name, 'jack') eq_(u.name, 'jack') u.name = 'jacko' assert m._columntoproperty[users.c.name] is m.get_property('_name') def test_add_column_prop_deannotate(self): User, users = self.classes.User, self.tables.users Address, addresses = self.classes.Address, self.tables.addresses class SubUser(User): pass m = mapper(User, users) m2 = mapper(SubUser, addresses, inherits=User) m3 = mapper(Address, addresses, properties={ 'foo':relationship(m2) }) # add property using annotated User.name, # needs to be deannotated m.add_property("x", column_property(User.name + "name")) s = create_session() q = s.query(m2).select_from(Address).join(Address.foo) self.assert_compile( q, "SELECT " "anon_1.addresses_id AS anon_1_addresses_id, " "anon_1.users_id AS anon_1_users_id, " "anon_1.users_name AS anon_1_users_name, " "anon_1.addresses_user_id AS anon_1_addresses_user_id, " "anon_1.addresses_email_address AS " "anon_1_addresses_email_address, " "anon_1.users_name || :name_1 AS anon_2 " "FROM addresses JOIN (SELECT users.id AS users_id, " "users.name AS users_name, addresses.id AS addresses_id, " "addresses.user_id AS addresses_user_id, " "addresses.email_address AS addresses_email_address " "FROM users JOIN addresses ON users.id = " "addresses.user_id) AS anon_1 ON " "anon_1.users_id = addresses.user_id" ) def test_column_prop_deannotate(self): """test that column property deannotates, bringing expressions down to the original mapped columns. """ User, users = self.classes.User, self.tables.users m = mapper(User, users) assert User.id.property.columns[0] is users.c.id assert User.name.property.columns[0] is users.c.name expr = User.name + "name" expr2 = sa.select([User.name, users.c.id]) m.add_property("x", column_property(expr)) m.add_property("y", column_property(expr2)) assert User.x.property.columns[0] is not expr assert User.x.property.columns[0].element.left is users.c.name assert User.x.property.columns[0].element.right is not expr.right assert User.y.property.columns[0] is not expr2 assert User.y.property.columns[0].element.\ _raw_columns[0] is users.c.name assert User.y.property.columns[0].element.\ _raw_columns[1] is users.c.id def test_synonym_replaces_backref(self): addresses, users, User = (self.tables.addresses, self.tables.users, self.classes.User) assert_calls = [] class Address(object): def _get_user(self): assert_calls.append("get") return self._user def _set_user(self, user): assert_calls.append("set") self._user = user user = property(_get_user, _set_user) # synonym is created against nonexistent prop mapper(Address, addresses, properties={ 'user':synonym('_user') }) sa.orm.configure_mappers() # later, backref sets up the prop mapper(User, users, properties={ 'addresses':relationship(Address, backref='_user') }) sess = create_session() u1 = sess.query(User).get(7) u2 = sess.query(User).get(8) # comparaison ops need to work a1 = sess.query(Address).filter(Address.user==u1).one() eq_(a1.id, 1) a1.user = u2 assert a1.user is u2 eq_(assert_calls, ["set", "get"]) def test_self_ref_synonym(self): t = Table('nodes', MetaData(), Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('parent_id', Integer, ForeignKey('nodes.id'))) class Node(object): pass mapper(Node, t, properties={ '_children':relationship(Node, backref=backref('_parent', remote_side=t.c.id)), 'children':synonym('_children'), 'parent':synonym('_parent') }) n1 = Node() n2 = Node() n1.children.append(n2) assert n2.parent is n2._parent is n1 assert n1.children[0] is n1._children[0] is n2 eq_(str(Node.parent == n2), ":param_1 = nodes.parent_id") def test_non_primary_identity_class(self): User = self.classes.User users, addresses = self.tables.users, self.tables.addresses class AddressUser(User): pass m1 = mapper(User, users, polymorphic_identity='user') m2 = mapper(AddressUser, addresses, inherits=User, polymorphic_identity='address') m3 = mapper(AddressUser, addresses, non_primary=True) assert m3._identity_class is m2._identity_class eq_( m2.identity_key_from_instance(AddressUser()), m3.identity_key_from_instance(AddressUser()) ) def test_illegal_non_primary(self): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) mapper(User, users) mapper(Address, addresses) mapper(User, users, non_primary=True, properties={ 'addresses':relationship(Address) }) assert_raises_message( sa.exc.ArgumentError, "Attempting to assign a new relationship 'addresses' " "to a non-primary mapper on class 'User'", configure_mappers ) def test_illegal_non_primary_2(self): User, users = self.classes.User, self.tables.users assert_raises_message( sa.exc.InvalidRequestError, "Configure a primary mapper first", mapper, User, users, non_primary=True) def test_illegal_non_primary_3(self): users, addresses = self.tables.users, self.tables.addresses class Base(object): pass class Sub(Base): pass mapper(Base, users) assert_raises_message(sa.exc.InvalidRequestError, "Configure a primary mapper first", mapper, Sub, addresses, non_primary=True ) def test_prop_filters(self): t = Table('person', MetaData(), Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('type', String(128)), Column('name', String(128)), Column('employee_number', Integer), Column('boss_id', Integer, ForeignKey('person.id')), Column('vendor_id', Integer)) class Person(object): pass class Vendor(Person): pass class Employee(Person): pass class Manager(Employee): pass class Hoho(object): pass class Lala(object): pass class Fub(object):pass class Frob(object):pass class HasDef(object): def name(self): pass class Empty(object):pass empty = mapper(Empty, t, properties={'empty_id' : t.c.id}, include_properties=[]) p_m = mapper(Person, t, polymorphic_on=t.c.type, include_properties=('id', 'type', 'name')) e_m = mapper(Employee, inherits=p_m, polymorphic_identity='employee', properties={'boss' : relationship(Manager, backref=backref('peon'), remote_side=t.c.id)}, exclude_properties=('vendor_id', )) m_m = mapper(Manager, inherits=e_m, polymorphic_identity='manager', include_properties=('id', 'type')) v_m = mapper(Vendor, inherits=p_m, polymorphic_identity='vendor', exclude_properties=('boss_id', 'employee_number')) h_m = mapper(Hoho, t, include_properties=('id', 'type', 'name')) l_m = mapper(Lala, t, exclude_properties=('vendor_id', 'boss_id'), column_prefix="p_") hd_m = mapper(HasDef, t, column_prefix="h_") fb_m = mapper(Fub, t, include_properties=(t.c.id, t.c.type)) frb_m = mapper(Frob, t, column_prefix='f_', exclude_properties=(t.c.boss_id, 'employee_number', t.c.vendor_id)) configure_mappers() def assert_props(cls, want): have = set([n for n in dir(cls) if not n.startswith('_')]) want = set(want) eq_(have, want) def assert_instrumented(cls, want): have = set([p.key for p in class_mapper(cls).iterate_properties]) want = set(want) eq_(have, want) assert_props(HasDef, ['h_boss_id', 'h_employee_number', 'h_id', 'name', 'h_name', 'h_vendor_id', 'h_type']) assert_props(Person, ['id', 'name', 'type']) assert_instrumented(Person, ['id', 'name', 'type']) assert_props(Employee, ['boss', 'boss_id', 'employee_number', 'id', 'name', 'type']) assert_instrumented(Employee,['boss', 'boss_id', 'employee_number', 'id', 'name', 'type']) assert_props(Manager, ['boss', 'boss_id', 'employee_number', 'peon', 'id', 'name', 'type']) # 'peon' and 'type' are both explicitly stated properties assert_instrumented(Manager, ['peon', 'type', 'id']) assert_props(Vendor, ['vendor_id', 'id', 'name', 'type']) assert_props(Hoho, ['id', 'name', 'type']) assert_props(Lala, ['p_employee_number', 'p_id', 'p_name', 'p_type']) assert_props(Fub, ['id', 'type']) assert_props(Frob, ['f_id', 'f_type', 'f_name', ]) # putting the discriminator column in exclude_properties, # very weird. As of 0.7.4 this re-maps it. class Foo(Person): pass assert_props(Empty, ['empty_id']) mapper( Foo, inherits=Person, polymorphic_identity='foo', exclude_properties=('type', ), ) assert hasattr(Foo, "type") assert Foo.type.property.columns[0] is t.c.type @testing.provide_metadata def test_prop_filters_defaults(self): metadata = self.metadata t = Table('t', metadata, Column('id', Integer(), primary_key=True, test_needs_autoincrement=True), Column('x', Integer(), nullable=False, server_default='0') ) t.create() class A(object): pass mapper(A, t, include_properties=['id']) s = Session() s.add(A()) s.commit() def test_we_dont_call_bool(self): class NoBoolAllowed(object): def __nonzero__(self): raise Exception("nope") mapper(NoBoolAllowed, self.tables.users) u1 = NoBoolAllowed() u1.name = "some name" s = Session(testing.db) s.add(u1) s.commit() assert s.query(NoBoolAllowed).get(u1.id) is u1 def test_we_dont_call_eq(self): class NoEqAllowed(object): def __eq__(self, other): raise Exception("nope") addresses, users = self.tables.addresses, self.tables.users Address = self.classes.Address mapper(NoEqAllowed, users, properties={ 'addresses':relationship(Address, backref='user') }) mapper(Address, addresses) u1 = NoEqAllowed() u1.name = "some name" u1.addresses = [Address(id=12, email_address='a1')] s = Session(testing.db) s.add(u1) s.commit() a1 = s.query(Address).filter_by(id=12).one() assert a1.user is u1 def test_mapping_to_join_raises(self): """Test implicit merging of two cols raises.""" addresses, users, User = (self.tables.addresses, self.tables.users, self.classes.User) usersaddresses = sa.join(users, addresses, users.c.id == addresses.c.user_id) assert_raises_message( sa.exc.InvalidRequestError, "Implicitly", mapper, User, usersaddresses, primary_key=[users.c.id] ) def test_mapping_to_join_explicit_prop(self): """Mapping to a join""" User, addresses, users = (self.classes.User, self.tables.addresses, self.tables.users) usersaddresses = sa.join(users, addresses, users.c.id == addresses.c.user_id) mapper(User, usersaddresses, primary_key=[users.c.id], properties={'add_id':addresses.c.id} ) l = create_session().query(User).order_by(users.c.id).all() eq_(l, self.static.user_result[:3]) def test_mapping_to_join_exclude_prop(self): """Mapping to a join""" User, addresses, users = (self.classes.User, self.tables.addresses, self.tables.users) usersaddresses = sa.join(users, addresses, users.c.id == addresses.c.user_id) mapper(User, usersaddresses, primary_key=[users.c.id], exclude_properties=[addresses.c.id] ) l = create_session().query(User).order_by(users.c.id).all() eq_(l, self.static.user_result[:3]) def test_mapping_to_join_no_pk(self): email_bounces, addresses, Address = (self.tables.email_bounces, self.tables.addresses, self.classes.Address) m = mapper(Address, addresses.join(email_bounces), properties={'id':[addresses.c.id, email_bounces.c.id]} ) configure_mappers() assert addresses in m._pks_by_table assert email_bounces not in m._pks_by_table sess = create_session() a = Address(id=10, email_address='e1') sess.add(a) sess.flush() eq_(addresses.count().scalar(), 6) eq_(email_bounces.count().scalar(), 5) def test_mapping_to_outerjoin(self): """Mapping to an outer join with a nullable composite primary key.""" users, addresses, User = (self.tables.users, self.tables.addresses, self.classes.User) mapper(User, users.outerjoin(addresses), primary_key=[users.c.id, addresses.c.id], properties=dict( address_id=addresses.c.id)) session = create_session() l = session.query(User).order_by(User.id, User.address_id).all() eq_(l, [ User(id=7, address_id=1), User(id=8, address_id=2), User(id=8, address_id=3), User(id=8, address_id=4), User(id=9, address_id=5), User(id=10, address_id=None)]) def test_mapping_to_outerjoin_no_partial_pks(self): """test the allow_partial_pks=False flag.""" users, addresses, User = (self.tables.users, self.tables.addresses, self.classes.User) mapper(User, users.outerjoin(addresses), allow_partial_pks=False, primary_key=[users.c.id, addresses.c.id], properties=dict( address_id=addresses.c.id)) session = create_session() l = session.query(User).order_by(User.id, User.address_id).all() eq_(l, [ User(id=7, address_id=1), User(id=8, address_id=2), User(id=8, address_id=3), User(id=8, address_id=4), User(id=9, address_id=5), None]) def test_scalar_pk_arg(self): users, Keyword, items, Item, User, keywords = (self.tables.users, self.classes.Keyword, self.tables.items, self.classes.Item, self.classes.User, self.tables.keywords) m1 = mapper(Item, items, primary_key=[items.c.id]) m2 = mapper(Keyword, keywords, primary_key=keywords.c.id) m3 = mapper(User, users, primary_key=(users.c.id,)) assert m1.primary_key[0] is items.c.id assert m2.primary_key[0] is keywords.c.id assert m3.primary_key[0] is users.c.id def test_custom_join(self): """select_from totally replace the FROM parameters.""" users, items, order_items, orders, Item, User, Order = (self.tables.users, self.tables.items, self.tables.order_items, self.tables.orders, self.classes.Item, self.classes.User, self.classes.Order) mapper(Item, items) mapper(Order, orders, properties=dict( items=relationship(Item, order_items))) mapper(User, users, properties=dict( orders=relationship(Order))) session = create_session() l = (session.query(User). select_from(users.join(orders). join(order_items). join(items)). filter(items.c.description == 'item 4')).all() eq_(l, [self.static.user_result[0]]) def test_cancel_order_by(self): users, User = self.tables.users, self.classes.User mapper(User, users, order_by=users.c.name.desc()) assert "order by users.name desc" in str(create_session().query(User).statement).lower() assert "order by" not in str(create_session().query(User).order_by(None).statement).lower() assert "order by users.name asc" in str(create_session().query(User).order_by(User.name.asc()).statement).lower() eq_( create_session().query(User).all(), [User(id=7, name=u'jack'), User(id=9, name=u'fred'), User(id=8, name=u'ed'), User(id=10, name=u'chuck')] ) eq_( create_session().query(User).order_by(User.name).all(), [User(id=10, name=u'chuck'), User(id=8, name=u'ed'), User(id=9, name=u'fred'), User(id=7, name=u'jack')] ) # 'Raises a "expression evaluation not supported" error at prepare time @testing.fails_on('firebird', 'FIXME: unknown') def test_function(self): """Mapping to a SELECT statement that has functions in it.""" addresses, users, User = (self.tables.addresses, self.tables.users, self.classes.User) s = sa.select([users, (users.c.id * 2).label('concat'), sa.func.count(addresses.c.id).label('count')], users.c.id == addresses.c.user_id, group_by=[c for c in users.c]).alias('myselect') mapper(User, s, order_by=s.c.id) sess = create_session() l = sess.query(User).all() for idx, total in enumerate((14, 16)): eq_(l[idx].concat, l[idx].id * 2) eq_(l[idx].concat, total) def test_count(self): """The count function on Query.""" User, users = self.classes.User, self.tables.users mapper(User, users) session = create_session() q = session.query(User) eq_(q.count(), 4) eq_(q.filter(User.id.in_([8,9])).count(), 2) eq_(q.filter(users.c.id.in_([8,9])).count(), 2) eq_(session.query(User.id).count(), 4) eq_(session.query(User.id).filter(User.id.in_((8, 9))).count(), 2) def test_many_to_many_count(self): keywords, items, item_keywords, Keyword, Item = (self.tables.keywords, self.tables.items, self.tables.item_keywords, self.classes.Keyword, self.classes.Item) mapper(Keyword, keywords) mapper(Item, items, properties=dict( keywords = relationship(Keyword, item_keywords, lazy='select'))) session = create_session() q = (session.query(Item). join('keywords'). distinct(). filter(Keyword.name == "red")) eq_(q.count(), 2) def test_override_1(self): """Overriding a column raises an error.""" Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) def go(): mapper(User, users, properties=dict( name=relationship(mapper(Address, addresses)))) assert_raises(sa.exc.ArgumentError, go) def test_override_2(self): """exclude_properties cancels the error.""" Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(User, users, exclude_properties=['name'], properties=dict( name=relationship(mapper(Address, addresses)))) assert bool(User.name) def test_override_3(self): """The column being named elsewhere also cancels the error,""" Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(User, users, properties=dict( name=relationship(mapper(Address, addresses)), foo=users.c.name)) def test_synonym(self): users, addresses, Address = (self.tables.users, self.tables.addresses, self.classes.Address) assert_col = [] class extendedproperty(property): attribute = 123 def __getitem__(self, key): return 'value' class User(object): def _get_name(self): assert_col.append(('get', self.name)) return self.name def _set_name(self, name): assert_col.append(('set', name)) self.name = name uname = extendedproperty(_get_name, _set_name) mapper(User, users, properties=dict( addresses = relationship(mapper(Address, addresses), lazy='select'), uname = synonym('name'), adlist = synonym('addresses'), adname = synonym('addresses') )) # ensure the synonym can get at the proxied comparators without # an explicit compile User.name == 'ed' User.adname.any() assert hasattr(User, 'adlist') # as of 0.4.2, synonyms always create a property assert hasattr(User, 'adname') # test compile assert not isinstance(User.uname == 'jack', bool) assert User.uname.property assert User.adlist.property sess = create_session() # test RowTuple names row = sess.query(User.id, User.uname).first() assert row.uname == row[1] u = sess.query(User).filter(User.uname=='jack').one() fixture = self.static.user_address_result[0].addresses eq_(u.adlist, fixture) addr = sess.query(Address).filter_by(id=fixture[0].id).one() u = sess.query(User).filter(User.adname.contains(addr)).one() u2 = sess.query(User).filter(User.adlist.contains(addr)).one() assert u is u2 assert u not in sess.dirty u.uname = "some user name" assert len(assert_col) > 0 eq_(assert_col, [('set', 'some user name')]) eq_(u.uname, "some user name") eq_(assert_col, [('set', 'some user name'), ('get', 'some user name')]) eq_(u.name, "some user name") assert u in sess.dirty eq_(User.uname.attribute, 123) eq_(User.uname['key'], 'value') def test_synonym_of_synonym(self): users, User = (self.tables.users, self.classes.User) mapper(User, users, properties={ 'x':synonym('id'), 'y':synonym('x') }) s = Session() u = s.query(User).filter(User.y==8).one() eq_(u.y, 8) def test_synonym_column_location(self): users, User = self.tables.users, self.classes.User def go(): mapper(User, users, properties={ 'not_name':synonym('_name', map_column=True)}) assert_raises_message( sa.exc.ArgumentError, ("Can't compile synonym '_name': no column on table " "'users' named 'not_name'"), go) def test_column_synonyms(self): """Synonyms which automatically instrument properties, set up aliased column, etc.""" addresses, users, Address = (self.tables.addresses, self.tables.users, self.classes.Address) assert_col = [] class User(object): def _get_name(self): assert_col.append(('get', self._name)) return self._name def _set_name(self, name): assert_col.append(('set', name)) self._name = name name = property(_get_name, _set_name) mapper(Address, addresses) mapper(User, users, properties = { 'addresses':relationship(Address, lazy='select'), 'name':synonym('_name', map_column=True) }) # test compile assert not isinstance(User.name == 'jack', bool) assert hasattr(User, 'name') assert hasattr(User, '_name') sess = create_session() u = sess.query(User).filter(User.name == 'jack').one() eq_(u.name, 'jack') u.name = 'foo' eq_(u.name, 'foo') eq_(assert_col, [('get', 'jack'), ('set', 'foo'), ('get', 'foo')]) def test_synonym_map_column_conflict(self): users, User = self.tables.users, self.classes.User assert_raises( sa.exc.ArgumentError, mapper, User, users, properties=util.OrderedDict([ ('_user_id', users.c.id), ('id', synonym('_user_id', map_column=True)), ]) ) assert_raises( sa.exc.ArgumentError, mapper, User, users, properties=util.OrderedDict([ ('id', synonym('_user_id', map_column=True)), ('_user_id', users.c.id), ]) ) def test_comparable(self): users = self.tables.users class extendedproperty(property): attribute = 123 def method1(self): return "method1" def __getitem__(self, key): return 'value' class UCComparator(sa.orm.PropComparator): __hash__ = None def method1(self): return "uccmethod1" def method2(self, other): return "method2" def __eq__(self, other): cls = self.prop.parent.class_ col = getattr(cls, 'name') if other is None: return col == None else: return sa.func.upper(col) == sa.func.upper(other) def map_(with_explicit_property): class User(object): @extendedproperty def uc_name(self): if self.name is None: return None return self.name.upper() if with_explicit_property: args = (UCComparator, User.uc_name) else: args = (UCComparator,) mapper(User, users, properties=dict( uc_name = sa.orm.comparable_property(*args))) return User for User in (map_(True), map_(False)): sess = create_session() sess.begin() q = sess.query(User) assert hasattr(User, 'name') assert hasattr(User, 'uc_name') eq_(User.uc_name.method1(), "method1") eq_(User.uc_name.method2('x'), "method2") assert_raises_message( AttributeError, "Neither 'extendedproperty' object nor 'UCComparator' object has an attribute 'nonexistent'", getattr, User.uc_name, 'nonexistent') # test compile assert not isinstance(User.uc_name == 'jack', bool) u = q.filter(User.uc_name=='JACK').one() assert u.uc_name == "JACK" assert u not in sess.dirty u.name = "some user name" eq_(u.name, "some user name") assert u in sess.dirty eq_(u.uc_name, "SOME USER NAME") sess.flush() sess.expunge_all() q = sess.query(User) u2 = q.filter(User.name=='some user name').one() u3 = q.filter(User.uc_name=='SOME USER NAME').one() assert u2 is u3 eq_(User.uc_name.attribute, 123) eq_(User.uc_name['key'], 'value') sess.rollback() def test_comparable_column(self): users, User = self.tables.users, self.classes.User class MyComparator(sa.orm.properties.ColumnProperty.Comparator): __hash__ = None def __eq__(self, other): # lower case comparison return func.lower(self.__clause_element__()) == func.lower(other) def intersects(self, other): # non-standard comparator return self.__clause_element__().op('&=')(other) mapper(User, users, properties={ 'name':sa.orm.column_property(users.c.name, comparator_factory=MyComparator) }) assert_raises_message( AttributeError, "Neither 'InstrumentedAttribute' object nor 'MyComparator' object has an attribute 'nonexistent'", getattr, User.name, "nonexistent") eq_(str((User.name == 'ed').compile(dialect=sa.engine.default.DefaultDialect())) , "lower(users.name) = lower(:lower_1)") eq_(str((User.name.intersects('ed')).compile(dialect=sa.engine.default.DefaultDialect())), "users.name &= :name_1") def test_reentrant_compile(self): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) class MyFakeProperty(sa.orm.properties.ColumnProperty): def post_instrument_class(self, mapper): super(MyFakeProperty, self).post_instrument_class(mapper) configure_mappers() m1 = mapper(User, users, properties={ 'name':MyFakeProperty(users.c.name) }) m2 = mapper(Address, addresses) configure_mappers() sa.orm.clear_mappers() class MyFakeProperty(sa.orm.properties.ColumnProperty): def post_instrument_class(self, mapper): super(MyFakeProperty, self).post_instrument_class(mapper) configure_mappers() m1 = mapper(User, users, properties={ 'name':MyFakeProperty(users.c.name) }) m2 = mapper(Address, addresses) configure_mappers() def test_reconstructor(self): users = self.tables.users recon = [] class User(object): @reconstructor def reconstruct(self): recon.append('go') mapper(User, users) User() eq_(recon, []) create_session().query(User).first() eq_(recon, ['go']) def test_reconstructor_inheritance(self): users = self.tables.users recon = [] class A(object): @reconstructor def reconstruct(self): assert isinstance(self, A) recon.append('A') class B(A): @reconstructor def reconstruct(self): assert isinstance(self, B) recon.append('B') class C(A): @reconstructor def reconstruct(self): assert isinstance(self, C) recon.append('C') mapper(A, users, polymorphic_on=users.c.name, polymorphic_identity='jack') mapper(B, inherits=A, polymorphic_identity='ed') mapper(C, inherits=A, polymorphic_identity='chuck') A() B() C() eq_(recon, []) sess = create_session() sess.query(A).first() sess.query(B).first() sess.query(C).first() eq_(recon, ['A', 'B', 'C']) def test_unmapped_reconstructor_inheritance(self): users = self.tables.users recon = [] class Base(object): @reconstructor def reconstruct(self): recon.append('go') class User(Base): pass mapper(User, users) User() eq_(recon, []) create_session().query(User).first() eq_(recon, ['go']) def test_unmapped_error(self): Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(Address, addresses) sa.orm.clear_mappers() mapper(User, users, properties={ 'addresses':relationship(Address) }) assert_raises_message( sa.orm.exc.UnmappedClassError, "Class 'test.orm._fixtures.Address' is not mapped", sa.orm.configure_mappers) def test_unmapped_not_type_error(self): assert_raises_message( sa.exc.ArgumentError, "Class object expected, got '5'.", class_mapper, 5 ) def test_unmapped_subclass_error_postmap(self): users = self.tables.users class Base(object): pass class Sub(Base): pass mapper(Base, users) sa.orm.configure_mappers() # we can create new instances, set attributes. s = Sub() s.name = 'foo' eq_(s.name, 'foo') eq_( attributes.get_history(s, 'name'), (['foo'], (), ()) ) # using it with an ORM operation, raises assert_raises(sa.orm.exc.UnmappedClassError, create_session().add, Sub()) def test_unmapped_subclass_error_premap(self): users = self.tables.users class Base(object): pass mapper(Base, users) class Sub(Base): pass sa.orm.configure_mappers() # we can create new instances, set attributes. s = Sub() s.name = 'foo' eq_(s.name, 'foo') eq_( attributes.get_history(s, 'name'), (['foo'], (), ()) ) # using it with an ORM operation, raises assert_raises(sa.orm.exc.UnmappedClassError, create_session().add, Sub()) def test_oldstyle_mixin(self): users = self.tables.users class OldStyle: pass class NewStyle(object): pass class A(NewStyle, OldStyle): pass mapper(A, users) class B(OldStyle, NewStyle): pass mapper(B, users) class DocumentTest(fixtures.TestBase): def test_doc_propagate(self): metadata = MetaData() t1 = Table('t1', metadata, Column('col1', Integer, primary_key=True, doc="primary key column"), Column('col2', String, doc="data col"), Column('col3', String, doc="data col 2"), Column('col4', String, doc="data col 3"), Column('col5', String), ) t2 = Table('t2', metadata, Column('col1', Integer, primary_key=True, doc="primary key column"), Column('col2', String, doc="data col"), Column('col3', Integer, ForeignKey('t1.col1'), doc="foreign key to t1.col1") ) class Foo(object): pass class Bar(object): pass mapper(Foo, t1, properties={ 'bars':relationship(Bar, doc="bar relationship", backref=backref('foo',doc='foo relationship') ), 'foober':column_property(t1.c.col3, doc='alternate data col'), 'hoho':synonym("col4", doc="syn of col4") }) mapper(Bar, t2) configure_mappers() eq_(Foo.col1.__doc__, "primary key column") eq_(Foo.col2.__doc__, "data col") eq_(Foo.col5.__doc__, None) eq_(Foo.foober.__doc__, "alternate data col") eq_(Foo.bars.__doc__, "bar relationship") eq_(Foo.hoho.__doc__, "syn of col4") eq_(Bar.col1.__doc__, "primary key column") eq_(Bar.foo.__doc__, "foo relationship") class ORMLoggingTest(_fixtures.FixtureTest): def setup(self): self.buf = logging.handlers.BufferingHandler(100) for log in [ logging.getLogger('sqlalchemy.orm'), ]: log.addHandler(self.buf) def teardown(self): for log in [ logging.getLogger('sqlalchemy.orm'), ]: log.removeHandler(self.buf) def _current_messages(self): return [b.getMessage() for b in self.buf.buffer] def test_mapper_info_aliased(self): User, users = self.classes.User, self.tables.users tb = users.select().alias() mapper(User, tb) s = Session() s.add(User(name='ed')) s.commit() for msg in self._current_messages(): assert msg.startswith('(User|%%(%d anon)s) ' % id(tb)) class OptionsTest(_fixtures.FixtureTest): @testing.fails_on('maxdb', 'FIXME: unknown') def test_synonym_options(self): Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(User, users, properties=dict( addresses = relationship(mapper(Address, addresses), lazy='select', order_by=addresses.c.id), adlist = synonym('addresses'))) def go(): sess = create_session() u = (sess.query(User). order_by(User.id). options(sa.orm.joinedload('adlist')). filter_by(name='jack')).one() eq_(u.adlist, [self.static.user_address_result[0].addresses[0]]) self.assert_sql_count(testing.db, go, 1) def test_eager_options(self): """A lazy relationship can be upgraded to an eager relationship.""" Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(User, users, properties=dict( addresses = relationship(mapper(Address, addresses), order_by=addresses.c.id))) sess = create_session() l = (sess.query(User). order_by(User.id). options(sa.orm.joinedload('addresses'))).all() def go(): eq_(l, self.static.user_address_result) self.sql_count_(0, go) @testing.fails_on('maxdb', 'FIXME: unknown') def test_eager_options_with_limit(self): Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(User, users, properties=dict( addresses=relationship(mapper(Address, addresses), lazy='select'))) sess = create_session() u = (sess.query(User). options(sa.orm.joinedload('addresses')). filter_by(id=8)).one() def go(): eq_(u.id, 8) eq_(len(u.addresses), 3) self.sql_count_(0, go) sess.expunge_all() u = sess.query(User).filter_by(id=8).one() eq_(u.id, 8) eq_(len(u.addresses), 3) @testing.fails_on('maxdb', 'FIXME: unknown') def test_lazy_options_with_limit(self): Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(User, users, properties=dict( addresses = relationship(mapper(Address, addresses), lazy='joined'))) sess = create_session() u = (sess.query(User). options(sa.orm.lazyload('addresses')). filter_by(id=8)).one() def go(): eq_(u.id, 8) eq_(len(u.addresses), 3) self.sql_count_(1, go) def test_eager_degrade(self): """An eager relationship automatically degrades to a lazy relationship if eager columns are not available""" Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(User, users, properties=dict( addresses = relationship(mapper(Address, addresses), lazy='joined', order_by=addresses.c.id))) sess = create_session() # first test straight eager load, 1 statement def go(): l = sess.query(User).order_by(User.id).all() eq_(l, self.static.user_address_result) self.sql_count_(1, go) sess.expunge_all() # then select just from users. run it into instances. # then assert the data, which will launch 3 more lazy loads # (previous users in session fell out of scope and were removed from # session's identity map) r = users.select().order_by(users.c.id).execute() def go(): l = list(sess.query(User).instances(r)) eq_(l, self.static.user_address_result) self.sql_count_(4, go) def test_eager_degrade_deep(self): users, Keyword, items, order_items, orders, Item, User, Address, keywords, item_keywords, Order, addresses = (self.tables.users, self.classes.Keyword, self.tables.items, self.tables.order_items, self.tables.orders, self.classes.Item, self.classes.User, self.classes.Address, self.tables.keywords, self.tables.item_keywords, self.classes.Order, self.tables.addresses) # test with a deeper set of eager loads. when we first load the three # users, they will have no addresses or orders. the number of lazy # loads when traversing the whole thing will be three for the # addresses and three for the orders. mapper(Address, addresses) mapper(Keyword, keywords) mapper(Item, items, properties=dict( keywords=relationship(Keyword, secondary=item_keywords, lazy='joined', order_by=item_keywords.c.keyword_id))) mapper(Order, orders, properties=dict( items=relationship(Item, secondary=order_items, lazy='joined', order_by=order_items.c.item_id))) mapper(User, users, properties=dict( addresses=relationship(Address, lazy='joined', order_by=addresses.c.id), orders=relationship(Order, lazy='joined', order_by=orders.c.id))) sess = create_session() # first test straight eager load, 1 statement def go(): l = sess.query(User).order_by(User.id).all() eq_(l, self.static.user_all_result) self.assert_sql_count(testing.db, go, 1) sess.expunge_all() # then select just from users. run it into instances. # then assert the data, which will launch 6 more lazy loads r = users.select().execute() def go(): l = list(sess.query(User).instances(r)) eq_(l, self.static.user_all_result) self.assert_sql_count(testing.db, go, 6) def test_lazy_options(self): """An eager relationship can be upgraded to a lazy relationship.""" Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) mapper(User, users, properties=dict( addresses = relationship(mapper(Address, addresses), lazy='joined') )) sess = create_session() l = (sess.query(User). order_by(User.id). options(sa.orm.lazyload('addresses'))).all() def go(): eq_(l, self.static.user_address_result) self.sql_count_(4, go) def test_option_propagate(self): users, items, order_items, Order, Item, User, orders = (self.tables.users, self.tables.items, self.tables.order_items, self.classes.Order, self.classes.Item, self.classes.User, self.tables.orders) mapper(User, users, properties=dict( orders = relationship(Order) )) mapper(Order, orders, properties=dict( items = relationship(Item, secondary=order_items) )) mapper(Item, items) sess = create_session() oalias = aliased(Order) opt1 = sa.orm.joinedload(User.orders, Order.items) opt2a, opt2b = sa.orm.contains_eager(User.orders, Order.items, alias=oalias) u1 = sess.query(User).join(oalias, User.orders).options(opt1, opt2a, opt2b).first() ustate = attributes.instance_state(u1) assert opt1 in ustate.load_options assert opt2a not in ustate.load_options assert opt2b not in ustate.load_options class DeepOptionsTest(_fixtures.FixtureTest): @classmethod def setup_mappers(cls): users, Keyword, items, order_items, Order, Item, User, keywords, item_keywords, orders = (cls.tables.users, cls.classes.Keyword, cls.tables.items, cls.tables.order_items, cls.classes.Order, cls.classes.Item, cls.classes.User, cls.tables.keywords, cls.tables.item_keywords, cls.tables.orders) mapper(Keyword, keywords) mapper(Item, items, properties=dict( keywords=relationship(Keyword, item_keywords, order_by=item_keywords.c.item_id))) mapper(Order, orders, properties=dict( items=relationship(Item, order_items, order_by=items.c.id))) mapper(User, users, order_by=users.c.id, properties=dict( orders=relationship(Order, order_by=orders.c.id))) def test_deep_options_1(self): User = self.classes.User sess = create_session() # joinedload nothing. u = sess.query(User).all() def go(): x = u[0].orders[1].items[0].keywords[1] self.assert_sql_count(testing.db, go, 3) def test_deep_options_2(self): """test (joined|subquery)load_all() options""" User = self.classes.User sess = create_session() l = (sess.query(User). options(sa.orm.joinedload_all('orders.items.keywords'))).all() def go(): x = l[0].orders[1].items[0].keywords[1] self.sql_count_(0, go) sess = create_session() l = (sess.query(User). options(sa.orm.subqueryload_all('orders.items.keywords'))).all() def go(): x = l[0].orders[1].items[0].keywords[1] self.sql_count_(0, go) def test_deep_options_3(self): User = self.classes.User sess = create_session() # same thing, with separate options calls q2 = (sess.query(User). options(sa.orm.joinedload('orders')). options(sa.orm.joinedload('orders.items')). options(sa.orm.joinedload('orders.items.keywords'))) u = q2.all() def go(): x = u[0].orders[1].items[0].keywords[1] self.sql_count_(0, go) def test_deep_options_4(self): Item, User, Order = (self.classes.Item, self.classes.User, self.classes.Order) sess = create_session() assert_raises_message( sa.exc.ArgumentError, "Can't find property 'items' on any entity " "specified in this Query.", sess.query(User).options, sa.orm.joinedload(Order.items)) # joinedload "keywords" on items. it will lazy load "orders", then # lazy load the "items" on the order, but on "items" it will eager # load the "keywords" q3 = sess.query(User).options(sa.orm.joinedload('orders.items.keywords')) u = q3.all() def go(): x = u[0].orders[1].items[0].keywords[1] self.sql_count_(2, go) sess = create_session() q3 = sess.query(User).options( sa.orm.joinedload(User.orders, Order.items, Item.keywords)) u = q3.all() def go(): x = u[0].orders[1].items[0].keywords[1] self.sql_count_(2, go) class ValidatorTest(_fixtures.FixtureTest): def test_scalar(self): users = self.tables.users class User(fixtures.ComparableEntity): @validates('name') def validate_name(self, key, name): assert name != 'fred' return name + ' modified' mapper(User, users) sess = create_session() u1 = User(name='ed') eq_(u1.name, 'ed modified') assert_raises(AssertionError, setattr, u1, "name", "fred") eq_(u1.name, 'ed modified') sess.add(u1) sess.flush() sess.expunge_all() eq_(sess.query(User).filter_by(name='ed modified').one(), User(name='ed')) def test_collection(self): users, addresses, Address = (self.tables.users, self.tables.addresses, self.classes.Address) class User(fixtures.ComparableEntity): @validates('addresses') def validate_address(self, key, ad): assert '@' in ad.email_address return ad mapper(User, users, properties={'addresses':relationship(Address)}) mapper(Address, addresses) sess = create_session() u1 = User(name='edward') assert_raises(AssertionError, u1.addresses.append, Address(email_address='noemail')) u1.addresses.append(Address(id=15, email_address='foo@bar.com')) sess.add(u1) sess.flush() sess.expunge_all() eq_( sess.query(User).filter_by(name='edward').one(), User(name='edward', addresses=[Address(email_address='foo@bar.com')]) ) def test_validators_dict(self): users, addresses, Address = (self.tables.users, self.tables.addresses, self.classes.Address) class User(fixtures.ComparableEntity): @validates('name') def validate_name(self, key, name): assert name != 'fred' return name + ' modified' @validates('addresses') def validate_address(self, key, ad): assert '@' in ad.email_address return ad def simple_function(self, key, value): return key, value u_m = mapper(User, users, properties={'addresses':relationship(Address)}) mapper(Address, addresses) eq_( dict((k, v.__name__) for k, v in u_m.validators.items()), {'name':'validate_name', 'addresses':'validate_address'} ) class ComparatorFactoryTest(_fixtures.FixtureTest, AssertsCompiledSQL): def test_kwarg_accepted(self): users, Address = self.tables.users, self.classes.Address class DummyComposite(object): def __init__(self, x, y): pass from sqlalchemy.orm.interfaces import PropComparator class MyFactory(PropComparator): pass for args in ( (column_property, users.c.name), (deferred, users.c.name), (synonym, 'name'), (composite, DummyComposite, users.c.id, users.c.name), (relationship, Address), (backref, 'address'), (comparable_property, ), (dynamic_loader, Address) ): fn = args[0] args = args[1:] fn(comparator_factory=MyFactory, *args) def test_column(self): User, users = self.classes.User, self.tables.users from sqlalchemy.orm.properties import ColumnProperty class MyFactory(ColumnProperty.Comparator): __hash__ = None def __eq__(self, other): return func.foobar(self.__clause_element__()) == func.foobar(other) mapper(User, users, properties={'name':column_property(users.c.name, comparator_factory=MyFactory)}) self.assert_compile(User.name == 'ed', "foobar(users.name) = foobar(:foobar_1)", dialect=default.DefaultDialect()) self.assert_compile(aliased(User).name == 'ed', "foobar(users_1.name) = foobar(:foobar_1)", dialect=default.DefaultDialect()) def test_synonym(self): users, User = self.tables.users, self.classes.User from sqlalchemy.orm.properties import ColumnProperty class MyFactory(ColumnProperty.Comparator): __hash__ = None def __eq__(self, other): return func.foobar(self.__clause_element__()) ==\ func.foobar(other) mapper(User, users, properties={ 'name':synonym('_name', map_column=True, comparator_factory=MyFactory) }) self.assert_compile( User.name == 'ed', "foobar(users.name) = foobar(:foobar_1)", dialect=default.DefaultDialect()) self.assert_compile( aliased(User).name == 'ed', "foobar(users_1.name) = foobar(:foobar_1)", dialect=default.DefaultDialect()) def test_relationship(self): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) from sqlalchemy.orm.properties import PropertyLoader class MyFactory(PropertyLoader.Comparator): __hash__ = None def __eq__(self, other): return func.foobar(self.__clause_element__().c.user_id) == func.foobar(other.id) class MyFactory2(PropertyLoader.Comparator): __hash__ = None def __eq__(self, other): return func.foobar(self.__clause_element__().c.id) == func.foobar(other.user_id) mapper(User, users) mapper(Address, addresses, properties={ 'user':relationship(User, comparator_factory=MyFactory, backref=backref("addresses", comparator_factory=MyFactory2) ) } ) self.assert_compile(Address.user == User(id=5), "foobar(addresses.user_id) = foobar(:foobar_1)", dialect=default.DefaultDialect()) self.assert_compile(User.addresses == Address(id=5, user_id=7), "foobar(users.id) = foobar(:foobar_1)", dialect=default.DefaultDialect()) self.assert_compile(aliased(Address).user == User(id=5), "foobar(addresses_1.user_id) = foobar(:foobar_1)", dialect=default.DefaultDialect()) self.assert_compile(aliased(User).addresses == Address(id=5, user_id=7), "foobar(users_1.id) = foobar(:foobar_1)", dialect=default.DefaultDialect()) class DeferredTest(_fixtures.FixtureTest): def test_basic(self): """A basic deferred load.""" Order, orders = self.classes.Order, self.tables.orders mapper(Order, orders, order_by=orders.c.id, properties={ 'description': deferred(orders.c.description)}) o = Order() self.assert_(o.description is None) q = create_session().query(Order) def go(): l = q.all() o2 = l[2] x = o2.description self.sql_eq_(go, [ ("SELECT orders.id AS orders_id, " "orders.user_id AS orders_user_id, " "orders.address_id AS orders_address_id, " "orders.isopen AS orders_isopen " "FROM orders ORDER BY orders.id", {}), ("SELECT orders.description AS orders_description " "FROM orders WHERE orders.id = :param_1", {'param_1':3})]) def test_unsaved(self): """Deferred loading does not kick in when just PK cols are set.""" Order, orders = self.classes.Order, self.tables.orders mapper(Order, orders, properties={ 'description': deferred(orders.c.description)}) sess = create_session() o = Order() sess.add(o) o.id = 7 def go(): o.description = "some description" self.sql_count_(0, go) def test_synonym_group_bug(self): orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders, properties={ 'isopen':synonym('_isopen', map_column=True), 'description':deferred(orders.c.description, group='foo') }) sess = create_session() o1 = sess.query(Order).get(1) eq_(o1.description, "order 1") def test_unsaved_2(self): Order, orders = self.classes.Order, self.tables.orders mapper(Order, orders, properties={ 'description': deferred(orders.c.description)}) sess = create_session() o = Order() sess.add(o) def go(): o.description = "some description" self.sql_count_(0, go) def test_unsaved_group(self): """Deferred loading doesnt kick in when just PK cols are set""" orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders, order_by=orders.c.id, properties=dict( description=deferred(orders.c.description, group='primary'), opened=deferred(orders.c.isopen, group='primary'))) sess = create_session() o = Order() sess.add(o) o.id = 7 def go(): o.description = "some description" self.sql_count_(0, go) def test_unsaved_group_2(self): orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders, order_by=orders.c.id, properties=dict( description=deferred(orders.c.description, group='primary'), opened=deferred(orders.c.isopen, group='primary'))) sess = create_session() o = Order() sess.add(o) def go(): o.description = "some description" self.sql_count_(0, go) def test_save(self): Order, orders = self.classes.Order, self.tables.orders m = mapper(Order, orders, properties={ 'description': deferred(orders.c.description)}) sess = create_session() o2 = sess.query(Order).get(2) o2.isopen = 1 sess.flush() def test_group(self): """Deferred load with a group""" orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders, properties=util.OrderedDict([ ('userident', deferred(orders.c.user_id, group='primary')), ('addrident', deferred(orders.c.address_id, group='primary')), ('description', deferred(orders.c.description, group='primary')), ('opened', deferred(orders.c.isopen, group='primary')) ])) sess = create_session() q = sess.query(Order).order_by(Order.id) def go(): l = q.all() o2 = l[2] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, 'order 3') self.sql_eq_(go, [ ("SELECT orders.id AS orders_id " "FROM orders ORDER BY orders.id", {}), ("SELECT orders.user_id AS orders_user_id, " "orders.address_id AS orders_address_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen " "FROM orders WHERE orders.id = :param_1", {'param_1':3})]) o2 = q.all()[2] eq_(o2.description, 'order 3') assert o2 not in sess.dirty o2.description = 'order 3' def go(): sess.flush() self.sql_count_(0, go) def test_preserve_changes(self): """A deferred load operation doesn't revert modifications on attributes""" orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders, properties = { 'userident': deferred(orders.c.user_id, group='primary'), 'description': deferred(orders.c.description, group='primary'), 'opened': deferred(orders.c.isopen, group='primary') }) sess = create_session() o = sess.query(Order).get(3) assert 'userident' not in o.__dict__ o.description = 'somenewdescription' eq_(o.description, 'somenewdescription') def go(): eq_(o.opened, 1) self.assert_sql_count(testing.db, go, 1) eq_(o.description, 'somenewdescription') assert o in sess.dirty def test_commits_state(self): """ When deferred elements are loaded via a group, they get the proper CommittedState and don't result in changes being committed """ orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders, properties = { 'userident':deferred(orders.c.user_id, group='primary'), 'description':deferred(orders.c.description, group='primary'), 'opened':deferred(orders.c.isopen, group='primary')}) sess = create_session() o2 = sess.query(Order).get(3) # this will load the group of attributes eq_(o2.description, 'order 3') assert o2 not in sess.dirty # this will mark it as 'dirty', but nothing actually changed o2.description = 'order 3' # therefore the flush() shouldnt actually issue any SQL self.assert_sql_count(testing.db, sess.flush, 0) def test_options(self): """Options on a mapper to create deferred and undeferred columns""" orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders) sess = create_session() q = sess.query(Order).order_by(Order.id).options(defer('user_id')) def go(): q.all()[0].user_id self.sql_eq_(go, [ ("SELECT orders.id AS orders_id, " "orders.address_id AS orders_address_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen " "FROM orders ORDER BY orders.id", {}), ("SELECT orders.user_id AS orders_user_id " "FROM orders WHERE orders.id = :param_1", {'param_1':1})]) sess.expunge_all() q2 = q.options(sa.orm.undefer('user_id')) self.sql_eq_(q2.all, [ ("SELECT orders.id AS orders_id, " "orders.user_id AS orders_user_id, " "orders.address_id AS orders_address_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen " "FROM orders ORDER BY orders.id", {})]) def test_undefer_group(self): orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders, properties=util.OrderedDict([ ('userident',deferred(orders.c.user_id, group='primary')), ('description',deferred(orders.c.description, group='primary')), ('opened',deferred(orders.c.isopen, group='primary')) ] )) sess = create_session() q = sess.query(Order).order_by(Order.id) def go(): l = q.options(sa.orm.undefer_group('primary')).all() o2 = l[2] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, 'order 3') self.sql_eq_(go, [ ("SELECT orders.user_id AS orders_user_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen, " "orders.id AS orders_id, " "orders.address_id AS orders_address_id " "FROM orders ORDER BY orders.id", {})]) def test_locates_col(self): """Manually adding a column to the result undefers the column.""" orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders, properties={ 'description':deferred(orders.c.description)}) sess = create_session() o1 = sess.query(Order).order_by(Order.id).first() def go(): eq_(o1.description, 'order 1') self.sql_count_(1, go) sess = create_session() o1 = (sess.query(Order). order_by(Order.id). add_column(orders.c.description).first())[0] def go(): eq_(o1.description, 'order 1') self.sql_count_(0, go) def test_map_selectable_wo_deferred(self): """test mapping to a selectable with deferred cols, the selectable doesn't include the deferred col. """ Order, orders = self.classes.Order, self.tables.orders order_select = sa.select([ orders.c.id, orders.c.user_id, orders.c.address_id, orders.c.description, orders.c.isopen]).alias() mapper(Order, order_select, properties={ 'description':deferred(order_select.c.description) }) sess = Session() o1 = sess.query(Order).order_by(Order.id).first() assert 'description' not in o1.__dict__ eq_(o1.description, 'order 1') def test_deep_options(self): users, items, order_items, Order, Item, User, orders = (self.tables.users, self.tables.items, self.tables.order_items, self.classes.Order, self.classes.Item, self.classes.User, self.tables.orders) mapper(Item, items, properties=dict( description=deferred(items.c.description))) mapper(Order, orders, properties=dict( items=relationship(Item, secondary=order_items))) mapper(User, users, properties=dict( orders=relationship(Order, order_by=orders.c.id))) sess = create_session() q = sess.query(User).order_by(User.id) l = q.all() item = l[0].orders[1].items[1] def go(): eq_(item.description, 'item 4') self.sql_count_(1, go) eq_(item.description, 'item 4') sess.expunge_all() l = q.options(sa.orm.undefer('orders.items.description')).all() item = l[0].orders[1].items[1] def go(): eq_(item.description, 'item 4') self.sql_count_(0, go) eq_(item.description, 'item 4') class SecondaryOptionsTest(fixtures.MappedTest): """test that the contains_eager() option doesn't bleed into a secondary load.""" run_inserts = 'once' run_deletes = None @classmethod def define_tables(cls, metadata): Table("base", metadata, Column('id', Integer, primary_key=True), Column('type', String(50), nullable=False) ) Table("child1", metadata, Column('id', Integer, ForeignKey('base.id'), primary_key=True), Column('child2id', Integer, ForeignKey('child2.id'), nullable=False) ) Table("child2", metadata, Column('id', Integer, ForeignKey('base.id'), primary_key=True), ) Table('related', metadata, Column('id', Integer, ForeignKey('base.id'), primary_key=True), ) @classmethod def setup_mappers(cls): child1, child2, base, related = (cls.tables.child1, cls.tables.child2, cls.tables.base, cls.tables.related) class Base(cls.Comparable): pass class Child1(Base): pass class Child2(Base): pass class Related(cls.Comparable): pass mapper(Base, base, polymorphic_on=base.c.type, properties={ 'related':relationship(Related, uselist=False) }) mapper(Child1, child1, inherits=Base, polymorphic_identity='child1', properties={ 'child2':relationship(Child2, primaryjoin=child1.c.child2id==base.c.id, foreign_keys=child1.c.child2id) }) mapper(Child2, child2, inherits=Base, polymorphic_identity='child2') mapper(Related, related) @classmethod def insert_data(cls): child1, child2, base, related = (cls.tables.child1, cls.tables.child2, cls.tables.base, cls.tables.related) base.insert().execute([ {'id':1, 'type':'child1'}, {'id':2, 'type':'child1'}, {'id':3, 'type':'child1'}, {'id':4, 'type':'child2'}, {'id':5, 'type':'child2'}, {'id':6, 'type':'child2'}, ]) child2.insert().execute([ {'id':4}, {'id':5}, {'id':6}, ]) child1.insert().execute([ {'id':1, 'child2id':4}, {'id':2, 'child2id':5}, {'id':3, 'child2id':6}, ]) related.insert().execute([ {'id':1}, {'id':2}, {'id':3}, {'id':4}, {'id':5}, {'id':6}, ]) def test_contains_eager(self): Child1, Related = self.classes.Child1, self.classes.Related sess = create_session() child1s = sess.query(Child1).\ join(Child1.related).\ options(sa.orm.contains_eager(Child1.related)).\ order_by(Child1.id) def go(): eq_( child1s.all(), [ Child1(id=1, related=Related(id=1)), Child1(id=2, related=Related(id=2)), Child1(id=3, related=Related(id=3)) ] ) self.assert_sql_count(testing.db, go, 1) c1 = child1s[0] self.assert_sql_execution( testing.db, lambda: c1.child2, CompiledSQL( "SELECT child2.id AS child2_id, base.id AS base_id, base.type AS base_type " "FROM base JOIN child2 ON base.id = child2.id " "WHERE base.id = :param_1", {'param_1':4} ) ) def test_joinedload_on_other(self): Child1, Related = self.classes.Child1, self.classes.Related sess = create_session() child1s = sess.query(Child1).join(Child1.related).options(sa.orm.joinedload(Child1.related)).order_by(Child1.id) def go(): eq_( child1s.all(), [Child1(id=1, related=Related(id=1)), Child1(id=2, related=Related(id=2)), Child1(id=3, related=Related(id=3))] ) self.assert_sql_count(testing.db, go, 1) c1 = child1s[0] self.assert_sql_execution( testing.db, lambda: c1.child2, CompiledSQL( "SELECT child2.id AS child2_id, base.id AS base_id, base.type AS base_type " "FROM base JOIN child2 ON base.id = child2.id WHERE base.id = :param_1", # joinedload- this shouldn't happen # "SELECT base.id AS base_id, child2.id AS child2_id, base.type AS base_type, " # "related_1.id AS related_1_id FROM base JOIN child2 ON base.id = child2.id " # "LEFT OUTER JOIN related AS related_1 ON base.id = related_1.id WHERE base.id = :param_1", {'param_1':4} ) ) def test_joinedload_on_same(self): Child1, Child2, Related = (self.classes.Child1, self.classes.Child2, self.classes.Related) sess = create_session() child1s = sess.query(Child1).join(Child1.related).options(sa.orm.joinedload(Child1.child2, Child2.related)).order_by(Child1.id) def go(): eq_( child1s.all(), [Child1(id=1, related=Related(id=1)), Child1(id=2, related=Related(id=2)), Child1(id=3, related=Related(id=3))] ) self.assert_sql_count(testing.db, go, 4) c1 = child1s[0] # this *does* joinedload self.assert_sql_execution( testing.db, lambda: c1.child2, CompiledSQL( "SELECT child2.id AS child2_id, base.id AS base_id, base.type AS base_type, " "related_1.id AS related_1_id FROM base JOIN child2 ON base.id = child2.id " "LEFT OUTER JOIN related AS related_1 ON base.id = related_1.id WHERE base.id = :param_1", {'param_1':4} ) ) class DeferredPopulationTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): Table("thing", metadata, Column("id", Integer, primary_key=True, test_needs_autoincrement=True), Column("name", String(20))) Table("human", metadata, Column("id", Integer, primary_key=True, test_needs_autoincrement=True), Column("thing_id", Integer, ForeignKey("thing.id")), Column("name", String(20))) @classmethod def setup_mappers(cls): thing, human = cls.tables.thing, cls.tables.human class Human(cls.Basic): pass class Thing(cls.Basic): pass mapper(Human, human, properties={"thing": relationship(Thing)}) mapper(Thing, thing, properties={"name": deferred(thing.c.name)}) @classmethod def insert_data(cls): thing, human = cls.tables.thing, cls.tables.human thing.insert().execute([ {"id": 1, "name": "Chair"}, ]) human.insert().execute([ {"id": 1, "thing_id": 1, "name": "Clark Kent"}, ]) def _test(self, thing): assert "name" in attributes.instance_state(thing).dict def test_no_previous_query(self): Thing = self.classes.Thing session = create_session() thing = session.query(Thing).options(sa.orm.undefer("name")).first() self._test(thing) def test_query_twice_with_clear(self): Thing = self.classes.Thing session = create_session() result = session.query(Thing).first() session.expunge_all() thing = session.query(Thing).options(sa.orm.undefer("name")).first() self._test(thing) def test_query_twice_no_clear(self): Thing = self.classes.Thing session = create_session() result = session.query(Thing).first() thing = session.query(Thing).options(sa.orm.undefer("name")).first() self._test(thing) def test_joinedload_with_clear(self): Thing, Human = self.classes.Thing, self.classes.Human session = create_session() human = session.query(Human).options(sa.orm.joinedload("thing")).first() session.expunge_all() thing = session.query(Thing).options(sa.orm.undefer("name")).first() self._test(thing) def test_joinedload_no_clear(self): Thing, Human = self.classes.Thing, self.classes.Human session = create_session() human = session.query(Human).options(sa.orm.joinedload("thing")).first() thing = session.query(Thing).options(sa.orm.undefer("name")).first() self._test(thing) def test_join_with_clear(self): Thing, Human = self.classes.Thing, self.classes.Human session = create_session() result = session.query(Human).add_entity(Thing).join("thing").first() session.expunge_all() thing = session.query(Thing).options(sa.orm.undefer("name")).first() self._test(thing) def test_join_no_clear(self): Thing, Human = self.classes.Thing, self.classes.Human session = create_session() result = session.query(Human).add_entity(Thing).join("thing").first() thing = session.query(Thing).options(sa.orm.undefer("name")).first() self._test(thing) class NoLoadTest(_fixtures.FixtureTest): run_inserts = 'once' run_deletes = None def test_basic(self): """A basic one-to-many lazy load""" Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) m = mapper(User, users, properties=dict( addresses = relationship(mapper(Address, addresses), lazy='noload') )) q = create_session().query(m) l = [None] def go(): x = q.filter(User.id == 7).all() x[0].addresses l[0] = x self.assert_sql_count(testing.db, go, 1) self.assert_result(l[0], User, {'id' : 7, 'addresses' : (Address, [])}, ) def test_options(self): Address, addresses, users, User = (self.classes.Address, self.tables.addresses, self.tables.users, self.classes.User) m = mapper(User, users, properties=dict( addresses = relationship(mapper(Address, addresses), lazy='noload') )) q = create_session().query(m).options(sa.orm.lazyload('addresses')) l = [None] def go(): x = q.filter(User.id == 7).all() x[0].addresses l[0] = x self.sql_count_(2, go) self.assert_result(l[0], User, {'id' : 7, 'addresses' : (Address, [{'id' : 1}])}, ) class RequirementsTest(fixtures.MappedTest): """Tests the contract for user classes.""" @classmethod def define_tables(cls, metadata): Table('ht1', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('value', String(10))) Table('ht2', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('ht1_id', Integer, ForeignKey('ht1.id')), Column('value', String(10))) Table('ht3', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('value', String(10))) Table('ht4', metadata, Column('ht1_id', Integer, ForeignKey('ht1.id'), primary_key=True), Column('ht3_id', Integer, ForeignKey('ht3.id'), primary_key=True)) Table('ht5', metadata, Column('ht1_id', Integer, ForeignKey('ht1.id'), primary_key=True)) Table('ht6', metadata, Column('ht1a_id', Integer, ForeignKey('ht1.id'), primary_key=True), Column('ht1b_id', Integer, ForeignKey('ht1.id'), primary_key=True), Column('value', String(10))) # Py2K def test_baseclass(self): ht1 = self.tables.ht1 class OldStyle: pass assert_raises(sa.exc.ArgumentError, mapper, OldStyle, ht1) assert_raises(sa.exc.ArgumentError, mapper, 123) class NoWeakrefSupport(str): pass # TODO: is weakref support detectable without an instance? #self.assertRaises(sa.exc.ArgumentError, mapper, NoWeakrefSupport, t2) # end Py2K class _ValueBase(object): def __init__(self, value='abc', id=None): self.id = id self.value = value def __nonzero__(self): return False def __hash__(self): return hash(self.value) def __eq__(self, other): if isinstance(other, type(self)): return self.value == other.value return False def test_comparison_overrides(self): """Simple tests to ensure users can supply comparison __methods__. The suite-level test --options are better suited to detect problems- they add selected __methods__ across the board on all ORM tests. This test simply shoves a variety of operations through the ORM to catch basic regressions early in a standard test run. """ ht6, ht5, ht4, ht3, ht2, ht1 = (self.tables.ht6, self.tables.ht5, self.tables.ht4, self.tables.ht3, self.tables.ht2, self.tables.ht1) class H1(self._ValueBase): pass class H2(self._ValueBase): pass class H3(self._ValueBase): pass class H6(self._ValueBase): pass mapper(H1, ht1, properties={ 'h2s': relationship(H2, backref='h1'), 'h3s': relationship(H3, secondary=ht4, backref='h1s'), 'h1s': relationship(H1, secondary=ht5, backref='parent_h1'), 't6a': relationship(H6, backref='h1a', primaryjoin=ht1.c.id==ht6.c.ht1a_id), 't6b': relationship(H6, backref='h1b', primaryjoin=ht1.c.id==ht6.c.ht1b_id), }) mapper(H2, ht2) mapper(H3, ht3) mapper(H6, ht6) s = create_session() s.add_all([ H1('abc'), H1('def'), ]) h1 = H1('ghi') s.add(h1) h1.h2s.append(H2('abc')) h1.h3s.extend([H3(), H3()]) h1.h1s.append(H1()) s.flush() eq_(ht1.count().scalar(), 4) h6 = H6() h6.h1a = h1 h6.h1b = h1 h6 = H6() h6.h1a = h1 h6.h1b = x = H1() assert x in s h6.h1b.h2s.append(H2('def')) s.flush() h1.h2s.extend([H2('abc'), H2('def')]) s.flush() h1s = s.query(H1).options(sa.orm.joinedload('h2s')).all() eq_(len(h1s), 5) self.assert_unordered_result(h1s, H1, {'h2s': []}, {'h2s': []}, {'h2s': (H2, [{'value': 'abc'}, {'value': 'def'}, {'value': 'abc'}])}, {'h2s': []}, {'h2s': (H2, [{'value': 'def'}])}) h1s = s.query(H1).options(sa.orm.joinedload('h3s')).all() eq_(len(h1s), 5) h1s = s.query(H1).options(sa.orm.joinedload_all('t6a.h1b'), sa.orm.joinedload('h2s'), sa.orm.joinedload_all('h3s.h1s')).all() eq_(len(h1s), 5) def test_composite_results(self): ht2, ht1 = (self.tables.ht2, self.tables.ht1) class H1(self._ValueBase): def __init__(self, value, id, h2s): self.value = value self.id = id self.h2s = h2s class H2(self._ValueBase): def __init__(self, value, id): self.value = value self.id = id mapper(H1, ht1, properties={ 'h2s': relationship(H2, backref='h1'), }) mapper(H2, ht2) s = Session() s.add_all([ H1('abc', 1, h2s=[ H2('abc', id=1), H2('def', id=2), H2('def', id=3), ]), H1('def', 2, h2s=[ H2('abc', id=4), H2('abc', id=5), H2('def', id=6), ]), ]) s.commit() eq_( [(h1.value, h1.id, h2.value, h2.id) for h1, h2 in s.query(H1, H2).join(H1.h2s).order_by(H1.id, H2.id)], [ ('abc', 1, 'abc', 1), ('abc', 1, 'def', 2), ('abc', 1, 'def', 3), ('def', 2, 'abc', 4), ('def', 2, 'abc', 5), ('def', 2, 'def', 6), ] ) def test_nonzero_len_recursion(self): ht1 = self.tables.ht1 class H1(object): def __len__(self): return len(self.get_value()) def get_value(self): self.value = "foobar" return self.value class H2(object): def __nonzero__(self): return bool(self.get_value()) def get_value(self): self.value = "foobar" return self.value mapper(H1, ht1) mapper(H2, ht1) h1 = H1() h1.value = "Asdf" h1.value = "asdf asdf" # ding h2 = H2() h2.value = "Asdf" h2.value = "asdf asdf" # ding class MagicNamesTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): Table('cartographers', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('name', String(50)), Column('alias', String(50)), Column('quip', String(100))) Table('maps', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('cart_id', Integer, ForeignKey('cartographers.id')), Column('state', String(2)), Column('data', sa.Text)) @classmethod def setup_classes(cls): class Cartographer(cls.Basic): pass class Map(cls.Basic): pass def test_mappish(self): maps, Cartographer, cartographers, Map = (self.tables.maps, self.classes.Cartographer, self.tables.cartographers, self.classes.Map) mapper(Cartographer, cartographers, properties=dict( query=cartographers.c.quip)) mapper(Map, maps, properties=dict( mapper=relationship(Cartographer, backref='maps'))) c = Cartographer(name='Lenny', alias='The Dude', query='Where be dragons?') m = Map(state='AK', mapper=c) sess = create_session() sess.add(c) sess.flush() sess.expunge_all() for C, M in ((Cartographer, Map), (sa.orm.aliased(Cartographer), sa.orm.aliased(Map))): c1 = (sess.query(C). filter(C.alias=='The Dude'). filter(C.query=='Where be dragons?')).one() m1 = sess.query(M).filter(M.mapper==c1).one() def test_direct_stateish(self): for reserved in (sa.orm.instrumentation.ClassManager.STATE_ATTR, sa.orm.instrumentation.ClassManager.MANAGER_ATTR): t = Table('t', sa.MetaData(), Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column(reserved, Integer)) class T(object): pass assert_raises_message( KeyError, ('%r: requested attribute name conflicts with ' 'instrumentation attribute of the same name.' % reserved), mapper, T, t) def test_indirect_stateish(self): maps = self.tables.maps for reserved in (sa.orm.instrumentation.ClassManager.STATE_ATTR, sa.orm.instrumentation.ClassManager.MANAGER_ATTR): class M(object): pass assert_raises_message( KeyError, ('requested attribute name conflicts with ' 'instrumentation attribute of the same name'), mapper, M, maps, properties={ reserved: maps.c.state})