mapper.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

"""tests general mapper operations with an emphasis on selecting/loading"""

import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy import exceptions, sql
from sqlalchemy.orm import *
from sqlalchemy.ext.sessioncontext import SessionContext, SessionContextExt
from testlib import *
from testlib.tables import *
import testlib.tables as tables


class MapperSuperTest(TestBase, AssertsExecutionResults):
    def setUpAll(self):
        tables.create()
        tables.data()
    def tearDownAll(self):
        tables.drop()
    def tearDown(self):
        clear_mappers()
    def setUp(self):
        pass

class MapperTest(MapperSuperTest):

    def test_propconflict(self):
        """test that a backref created against an existing mapper with a property name
        conflict raises a decent error message"""
        mapper(Address, addresses)
        mapper(User, users,
            properties={
            'addresses':relation(Address, backref='email_address')
        })
        try:
            class_mapper(Address)
            class_mapper(User)
            assert False
        except exceptions.ArgumentError:
            pass

    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."

    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'"

    def test_columnprefix(self):
        mapper(User, users, column_prefix='_', properties={
            'user_name':synonym('_user_name')
        })

        s = create_session()
        u = s.get(User, 7)
        assert u._user_name=='jack'
        assert u._user_id ==7
        u2 = s.query(User).filter_by(user_name='jack').one()
        assert u is u2

    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)

    def test_compileonsession(self):
        m = mapper(User, users)
        session = create_session()
        session.connection(m)

    def test_incompletecolumns(self):
        """test loading from a select which does not contain all columns"""
        mapper(Address, addresses)
        s = create_session()
        a = s.query(Address).from_statement(select([addresses.c.address_id, addresses.c.user_id])).first()
        assert a.user_id == 7
        assert a.address_id == 1
        # email address auto-defers
        assert 'email_addres' not in a.__dict__
        assert a.email_address == '[e-mail]'

    def test_badconstructor(self):
        """test that if the construction of a mapped class fails, the instnace does not get placed in the session"""
        class Foo(object):
            def __init__(self, one, two):
                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

    @testing.uses_deprecated('SessionContext', 'SessionContextExt')
    def test_constructorexceptions(self):
        """test that exceptions raised in the mapped class are not masked by sa decorations"""
        ex = AssertionError('oops')
        sess = create_session()

        class Foo(object):
            def __init__(self):
                raise ex
        mapper(Foo, users)

        try:
            Foo()
            assert False
        except Exception, e:
            assert e is ex

        clear_mappers()
        mapper(Foo, users, extension=SessionContextExt(SessionContext()))
        def bad_expunge(foo):
            raise Exception("this exception should be stated as a warning")

        sess.expunge = bad_expunge
        try:
            Foo(_sa_session=sess)
            assert False
        except Exception, e:
            assert isinstance(e, exceptions.SAWarning)

        clear_mappers()

        # test that TypeError is raised for illegal constructor args,
        # whether or not explicit __init__ is present [ticket:908]
        class Foo(object):
            def __init__(self):
                pass
        class Bar(object):
            pass

        mapper(Foo, users)
        mapper(Bar, addresses)
        try:
            Foo(x=5)
            assert False
        except TypeError:
            assert True

        try:
            Bar(x=5)
            assert False
        except TypeError:
            assert True

    def test_props(self):
        m = mapper(User, users, properties = {
            'addresses' : relation(mapper(Address, addresses))
        }).compile()
        self.assert_(User.addresses.property is m.get_property('addresses'))

    def test_compileonprop(self):
        mapper(User, users, properties = {
            'addresses' : relation(mapper(Address, addresses))
        })
        User.addresses.any(Address.email_address=='[e-mail]')
        clear_mappers()

        mapper(User, users, properties = {
            'addresses' : relation(mapper(Address, addresses))
        })
        assert (User.user_id==3).compare(users.c.user_id==3)

        clear_mappers()

        class Foo(User):pass
        mapper(User, users)
        mapper(Foo, addresses, inherits=User)
        assert getattr(Foo().__class__, 'user_name').impl is not None

    def test_compileon_getprops(self):
        m =mapper(User, users)

        assert not m.compiled
        assert list(m.iterate_properties)
        assert m.compiled
        clear_mappers()

        m= mapper(User, users)
        assert not m.compiled
        assert m.get_property('user_name')
        assert m.compiled

    def test_add_property(self):
        assert_col = []
        class User(object):
            def _get_user_name(self):
                assert_col.append(('get', self._user_name))
                return self._user_name
            def _set_user_name(self, name):
                assert_col.append(('set', name))
                self._user_name = name
            user_name = property(_get_user_name, _set_user_name)

        m = mapper(User, users)
        mapper(Address, addresses)
        m.add_property('_user_name', deferred(users.c.user_name))
        m.add_property('user_name', synonym('_user_name'))
        m.add_property('addresses', relation(Address))

        sess = create_session(transactional=True)
        assert sess.query(User).get(7)

        u = sess.query(User).filter_by(user_name='jack').one()

        def go():
            self.assert_result([u], User, user_address_result[0])
            assert u.user_name == 'jack'
            assert assert_col == [('get', 'jack')], str(assert_col)
        self.assert_sql_count(testing.db, go, 2)

        u.name = 'ed'
        u3 = User()
        u3.user_name = 'some user'
        sess.save(u3)
        sess.flush()
        sess.rollback()

    def test_replace_property(self):
        m = mapper(User, users)
        m.add_property('_user_name',users.c.user_name)
        m.add_property('user_name', synonym('_user_name', proxy=True))

        sess = create_session()
        u = sess.query(User).filter_by(user_name='jack').one()
        assert u._user_name == 'jack'
        assert u.user_name == 'jack'
        u.user_name = 'jacko'
        assert m._columntoproperty[users.c.user_name] is m.get_property('_user_name')

        clear_mappers()

        m = mapper(User, users)
        m.add_property('user_name', synonym('_user_name', map_column=True))

        sess.clear()
        u = sess.query(User).filter_by(user_name='jack').one()
        assert u._user_name == 'jack'
        assert u.user_name == 'jack'
        u.user_name = 'jacko'
        assert m._columntoproperty[users.c.user_name] is m.get_property('_user_name')

    def test_synonym_replaces_backref(self):
        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')
        })
        compile_mappers()

        # later, backref sets up the prop
        mapper(User, users, properties={
            'addresses':relation(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()
        assert a1.address_id == 1
        a1.user = u2
        assert a1.user is u2
        self.assertEquals(assert_calls, ["set", "get"])

    def test_self_ref_syn(self):
        t = Table('nodes', MetaData(),
            Column('id', Integer, primary_key=True),
            Column('parent_id', Integer, ForeignKey('nodes.id')))

        class Node(object):
            pass

        mapper(Node, t, properties={
            '_children':relation(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
        self.assertEquals(str(Node.parent == n2), ":param_1 = nodes.parent_id")

    def test_illegal_non_primary(self):
        mapper(User, users)
        mapper(Address, addresses)
        try:
            mapper(User, users, non_primary=True, properties={
                'addresses':relation(Address)
            }).compile()
            assert False
        except exceptions.ArgumentError, e:
            assert "Attempting to assign a new relation 'addresses' to a non-primary mapper on class 'User'" in str(e)

    def test_propfilters(self):
        t = Table('person', MetaData(),
                  Column('id', Integer, primary_key=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

        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': relation(Manager, backref='peon')
          },
          exclude_properties=('vendor_id',))

        m_m = mapper(Manager, inherits=e_m, polymorphic_identity='manager',
                     include_properties=())

        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_")

        p_m.compile()
        #compile_mappers()

        def assert_props(cls, want):
            have = set([n for n in dir(cls) if not n.startswith('_')])
            want = set(want)
            want.add('c')
            self.assert_(have == want, repr(have) + " " + repr(want))

        assert_props(Person, ['id', 'name', 'type'])
        assert_props(Employee, ['boss', 'boss_id', 'employee_number',
                                'id', 'name', 'type'])
        assert_props(Manager, ['boss', 'boss_id', 'employee_number', 'peon',
                               'id', 'name', 'type'])
        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'])

    @testing.uses_deprecated('//select_by', '//join_via', '//list')
    def test_recursive_select_by_deprecated(self):
        """test that no endless loop occurs when traversing for select_by"""
        m = mapper(User, users, properties={
            'orders':relation(mapper(Order, orders), backref='user'),
            'addresses':relation(mapper(Address, addresses), backref='user'),
        })
        q = create_session().query(m)
        q.select_by(email_address='foo')

    def test_mappingtojoin(self):
        """test mapping to a join"""
        usersaddresses = sql.join(users, addresses, users.c.user_id == addresses.c.user_id)
        m = mapper(User, usersaddresses, primary_key=[users.c.user_id])
        q = create_session().query(m)
        l = q.all()