unitofwork.py

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

# coding: utf-8

"""Tests unitofwork operations."""

import testenv; testenv.configure_for_tests()
import pickleable
from sqlalchemy import *
from sqlalchemy import exceptions, sql
from sqlalchemy.orm import *
from testlib import *
from testlib.tables import *
from testlib import engines, tables, fixtures


# TODO: convert suite to not use Session.mapper, use fixtures.Base
# with explicit session.save()
Session = scoped_session(sessionmaker(autoflush=True, transactional=True))
orm_mapper = mapper
mapper = Session.mapper

class UnitOfWorkTest(object):
    pass

class HistoryTest(ORMTest):
    metadata = tables.metadata
    def define_tables(self, metadata):
        pass

    def test_backref(self):
        s = Session()
        class User(object):pass
        class Address(object):pass
        am = mapper(Address, addresses)
        m = mapper(User, users, properties = dict(
            addresses = relation(am, backref='user', lazy=False))
        )

        u = User(_sa_session=s)
        a = Address(_sa_session=s)
        a.user = u

        self.assert_(u.addresses == [a])
        s.commit()

        s.close()
        u = s.query(m).all()[0]
        print u.addresses[0].user

class VersioningTest(ORMTest):
    def define_tables(self, metadata):
        global version_table
        version_table = Table('version_test', metadata,
        Column('id', Integer, Sequence('version_test_seq', optional=True),
               primary_key=True ),
        Column('version_id', Integer, nullable=False),
        Column('value', String(40), nullable=False)
        )

    @engines.close_open_connections
    def test_basic(self):
        s = Session(scope=None)
        class Foo(object):pass
        mapper(Foo, version_table, version_id_col=version_table.c.version_id)
        f1 = Foo(value='f1', _sa_session=s)
        f2 = Foo(value='f2', _sa_session=s)
        s.commit()

        f1.value='f1rev2'
        s.commit()
        s2 = Session()
        f1_s = s2.query(Foo).get(f1.id)
        f1_s.value='f1rev3'
        s2.commit()

        f1.value='f1rev3mine'
        success = False
        try:
            # a concurrent session has modified this, should throw
            # an exception
            s.commit()
        except exceptions.ConcurrentModificationError, e:
            #print e
            success = True

        # Only dialects with a sane rowcount can detect the ConcurrentModificationError
        if testing.db.dialect.supports_sane_rowcount:
            assert success

        s.close()
        f1 = s.query(Foo).get(f1.id)
        f2 = s.query(Foo).get(f2.id)

        f1_s.value='f1rev4'
        s2.commit()

        s.delete(f1)
        s.delete(f2)
        success = False
        try:
            s.commit()
        except exceptions.ConcurrentModificationError, e:
            #print e
            success = True
        if testing.db.dialect.supports_sane_multi_rowcount:
            assert success

    @engines.close_open_connections
    def test_versioncheck(self):
        """test that query.with_lockmode performs a 'version check' on an already loaded instance"""
        s1 = Session(scope=None)
        class Foo(object):pass
        mapper(Foo, version_table, version_id_col=version_table.c.version_id)
        f1s1 =Foo(value='f1', _sa_session=s1)
        s1.commit()
        s2 = Session()
        f1s2 = s2.query(Foo).get(f1s1.id)
        f1s2.value='f1 new value'
        s2.commit()
        try:
            # load, version is wrong
            s1.query(Foo).with_lockmode('read').get(f1s1.id)
            assert False
        except exceptions.ConcurrentModificationError, e:
            assert True
        # reload it
        s1.query(Foo).load(f1s1.id)
        # now assert version OK
        s1.query(Foo).with_lockmode('read').get(f1s1.id)

        # assert brand new load is OK too
        s1.close()
        s1.query(Foo).with_lockmode('read').get(f1s1.id)

    @engines.close_open_connections
    def test_noversioncheck(self):
        """test that query.with_lockmode works OK when the mapper has no version id col"""
        s1 = Session()
        class Foo(object):pass
        mapper(Foo, version_table)
        f1s1 =Foo(value='f1', _sa_session=s1)
        f1s1.version_id=0
        s1.commit()
        s2 = Session()
        f1s2 = s2.query(Foo).with_lockmode('read').get(f1s1.id)
        assert f1s2.id == f1s1.id
        assert f1s2.value == f1s1.value

class UnicodeTest(ORMTest):
    def define_tables(self, metadata):
        global uni_table, uni_table2
        uni_table = Table('uni_test', metadata,
            Column('id',  Integer, Sequence("uni_test_id_seq", optional=True), primary_key=True),
            Column('txt', Unicode(50), unique=True))
        uni_table2 = Table('uni2', metadata,
            Column('id',  Integer, Sequence("uni2_test_id_seq", optional=True), primary_key=True),
            Column('txt', Unicode(50), ForeignKey(uni_table.c.txt)))

    def test_basic(self):
        class Test(object):
            def __init__(self, id, txt):
                self.id = id
                self.txt = txt
        mapper(Test, uni_table)

        txt = u"\u0160\u0110\u0106\u010c\u017d"
        t1 = Test(id=1, txt = txt)
        self.assert_(t1.txt == txt)
        Session.commit()
        self.assert_(t1.txt == txt)

    def test_relation(self):
        class Test(object):
            def __init__(self, txt):
                self.txt = txt
        class Test2(object):pass

        mapper(Test, uni_table, properties={
            't2s':relation(Test2)
        })
        mapper(Test2, uni_table2)

        txt = u"\u0160\u0110\u0106\u010c\u017d"
        t1 = Test(txt=txt)
        t1.t2s.append(Test2())
        t1.t2s.append(Test2())
        Session.commit()
        Session.close()
        t1 = Session.query(Test).filter_by(id=t1.id).one()
        assert len(t1.t2s) == 2

class UnicodeSchemaTest(ORMTest):
    __unsupported_on__ = ('oracle', 'mssql', 'firebird', 'sybase',
                          'access', 'maxdb')
    __excluded_on__ = (('mysql', '<', (4, 1, 1)),)

    metadata = MetaData(engines.utf8_engine())

    def define_tables(self, metadata):
        global t1, t2

        t1 = Table('unitable1', metadata,
            Column(u'm茅il', Integer, primary_key=True, key='a'),
            Column(u'\u6e2c\u8a66', Integer, key='b'),
            Column('type',  String(20)),
            test_needs_fk=True,
            )
        t2 = Table(u'Unit茅ble2', metadata,
            Column(u'm茅il', Integer, primary_key=True, key="cc"),
            Column(u'\u6e2c\u8a66', Integer, ForeignKey(u'unitable1.a'), key="d"),
           Column(u'\u6e2c\u8a66_2', Integer, key="e"),
                  test_needs_fk=True,
            )

    def test_mapping(self):
        class A(fixtures.Base):pass
        class B(fixtures.Base):pass

        mapper(A, t1, properties={
            't2s':relation(B),
        })
        mapper(B, t2)
        a1 = A()
        b1 = B()
        a1.t2s.append(b1)
        Session.flush()
        Session.clear()
        new_a1 = Session.query(A).filter(t1.c.a == a1.a).one()
        assert new_a1.a == a1.a
        assert new_a1.t2s[0].d == b1.d
        Session.clear()

        new_a1 = Session.query(A).options(eagerload('t2s')).filter(t1.c.a == a1.a).one()
        assert new_a1.a == a1.a
        assert new_a1.t2s[0].d == b1.d
        Session.clear()

        new_a1 = Session.query(A).filter(A.a == a1.a).one()
        assert new_a1.a == a1.a
        assert new_a1.t2s[0].d == b1.d
        Session.clear()

    def test_inheritance_mapping(self):
        class A(fixtures.Base):pass
        class B(A):pass
        mapper(A, t1, polymorphic_on=t1.c.type, polymorphic_identity='a')
        mapper(B, t2, inherits=A, polymorphic_identity='b')
        a1 = A(b=5)
        b1 = B(e=7)

        Session.flush()
        Session.clear()
        # TODO: somehow, not assigning to "l" first
        # breaks the comparison ?????
        l = Session.query(A).all()
        assert [A(b=5), B(e=7)] == l

class MutableTypesTest(ORMTest):
    def define_tables(self, metadata):
        global table
        table = Table('mutabletest', metadata,
            Column('id', Integer, Sequence('mutableidseq', optional=True), primary_key=True),
            Column('data', PickleType),
            Column('val', Unicode(30)))

    def test_basic(self):
        """test that types marked as MutableType get changes detected on them"""
        class Foo(object):pass
        mapper(Foo, table)
        f1 = Foo()
        f1.data = pickleable.Bar(4,5)
        Session.commit()
        Session.close()
        f2 = Session.query(Foo).filter_by(id=f1.id).one()
        assert f2.data == f1.data
        f2.data.y = 19
        assert f2 in Session.dirty
        Session.commit()
        Session.close()
        f3 = Session.query(Foo).filter_by(id=f1.id).one()
        print f2.data, f3.data
        assert f3.data != f1.data
        assert f3.data == pickleable.Bar(4, 19)

    def test_mutablechanges(self):
        """test that mutable changes are detected or not detected correctly"""
        class Foo(object):pass
        mapper(Foo, table)
        f1 = Foo()
        f1.data = pickleable.Bar(4,5)
        f1.val = unicode('hi')
        Session.commit()
        def go():
            Session.commit()
        self.assert_sql_count(testing.db, go, 0)
        f1.val = unicode('someothervalue')
        self.assert_sql(testing.db, lambda: Session.commit(), [
            (
                "UPDATE mutabletest SET val=:val WHERE mutabletest.id = :mutabletest_id",
                {'mutabletest_id': f1.id, 'val': u'someothervalue'}
            ),
        ])
        f1.val = unicode('hi')
        f1.data.x = 9
        self.assert_sql(testing.db, lambda: Session.commit(), [
            (
                "UPDATE mutabletest SET data=:data, val=:val WHERE mutabletest.id = :mutabletest_id",
                {'mutabletest_id': f1.id, 'val': u'hi', 'data':f1.data}
            ),
        ])

    def test_nocomparison(self):
        """test that types marked as MutableType get changes detected on them when the type has no __eq__ method"""
        class Foo(object):pass
        mapper(Foo, table)
        f1 = Foo()
        f1.data = pickleable.BarWithoutCompare(4,5)
        Session.commit()

        def go():
            Session.commit()
        self.assert_sql_count(testing.db, go, 0)

        Session.close()

        f2 = Session.query(Foo).filter_by(id=f1.id).one()

        def go():
            Session.commit()
        self.assert_sql_count(testing.db, go, 0)

        f2.data.y = 19
        def go():
            Session.commit()
        self.assert_sql_count(testing.db, go, 1)

        Session.close()
        f3 = Session.query(Foo).filter_by(id=f1.id).one()
        print f2.data, f3.data
        assert (f3.data.x, f3.data.y) == (4,19)

        def go():
            Session.commit()
        self.assert_sql_count(testing.db, go, 0)

    def test_unicode(self):
        """test that two equivalent unicode values dont get flagged as changed.

        apparently two equal unicode objects dont compare via "is" in all cases, so this
        tests the compare_values() call on types.String and its usage via types.Unicode."""
        class Foo(object):pass
        mapper(Foo, table)
        f1 = Foo()
        f1.val = u'hi'
        Session.commit()
        Session.close()
        f1 = Session.get(Foo, f1.id)
        f1.val = u'hi'
        def go():
            Session.commit()
        self.assert_sql_count(testing.db, go, 0)

class MutableTypesTest2(ORMTest):
    def define_tables(self, metadata):
        global table
        import operator
        table = Table('mutabletest', metadata,
            Column('id', Integer, Sequence('mutableidseq', optional=True), primary_key=True),
            Column('data', PickleType(comparator=operator.eq)),