unitofwork.py

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

            )

    def test_dicts(self):
        """dictionaries dont pickle the same way twice, sigh."""

        class Foo(object):pass
        mapper(Foo, table)
        f1 = Foo()
        f1.data = [{'personne': {'nom': u'Smith', 'pers_id': 1, 'prenom': u'john', 'civilite': u'Mr', \
                    'int_3': False, 'int_2': False, 'int_1': u'23', 'VenSoir': True, 'str_1': u'Test', \
                    'SamMidi': False, 'str_2': u'chien', 'DimMidi': False, 'SamSoir': True, 'SamAcc': False}}]

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

        f1.data = [{'personne': {'nom': u'Smith', 'pers_id': 1, 'prenom': u'john', 'civilite': u'Mr', \
                    'int_3': False, 'int_2': False, 'int_1': u'23', 'VenSoir': True, 'str_1': u'Test', \
                    'SamMidi': False, 'str_2': u'chien', 'DimMidi': False, 'SamSoir': True, 'SamAcc': False}}]

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

        f1.data[0]['personne']['VenSoir']= False
        def go():
            Session.commit()
        self.assert_sql_count(testing.db, go, 1)

        Session.clear()
        f = Session.query(Foo).get(f1.id)
        assert f.data == [{'personne': {'nom': u'Smith', 'pers_id': 1, 'prenom': u'john', 'civilite': u'Mr', \
                    'int_3': False, 'int_2': False, 'int_1': u'23', 'VenSoir': False, 'str_1': u'Test', \
                    'SamMidi': False, 'str_2': u'chien', 'DimMidi': False, 'SamSoir': True, 'SamAcc': False}}]

class PKTest(ORMTest):
    def define_tables(self, metadata):
        global table, table2, table3

        table = Table(
            'multipk', metadata,
            Column('multi_id', Integer, Sequence("multi_id_seq", optional=True), primary_key=True),
            Column('multi_rev', Integer, primary_key=True),
            Column('name', String(50), nullable=False),
            Column('value', String(100))
        )

        table2 = Table('multipk2', metadata,
            Column('pk_col_1', String(30), primary_key=True),
            Column('pk_col_2', String(30), primary_key=True),
            Column('data', String(30), )
            )
        table3 = Table('multipk3', metadata,
            Column('pri_code', String(30), key='primary', primary_key=True),
            Column('sec_code', String(30), key='secondary', primary_key=True),
            Column('date_assigned', Date, key='assigned', primary_key=True),
            Column('data', String(30), )
            )

    # not supported on sqlite since sqlite's auto-pk generation only works with
    # single column primary keys
    @testing.fails_on('sqlite')
    def test_primarykey(self):
        class Entry(object):
            pass
        Entry.mapper = mapper(Entry, table)
        e = Entry()
        e.name = 'entry1'
        e.value = 'this is entry 1'
        e.multi_rev = 2
        Session.commit()
        Session.close()
        e2 = Query(Entry).get((e.multi_id, 2))
        self.assert_(e is not e2 and e._instance_key == e2._instance_key)

    # this one works with sqlite since we are manually setting up pk values
    def test_manualpk(self):
        class Entry(object):
            pass
        Entry.mapper = mapper(Entry, table2)
        e = Entry()
        e.pk_col_1 = 'pk1'
        e.pk_col_2 = 'pk1_related'
        e.data = 'im the data'
        Session.commit()

    def test_keypks(self):
        import datetime
        class Entity(object):
            pass
        Entity.mapper = mapper(Entity, table3)
        e = Entity()
        e.primary = 'pk1'
        e.secondary = 'pk2'
        e.assigned = datetime.date.today()
        e.data = 'some more data'
        Session.commit()

class ForeignPKTest(ORMTest):
    """tests mapper detection of the relationship direction when parent/child tables are joined on their
    primary keys"""

    def define_tables(self, metadata):
        global people, peoplesites

        people = Table("people", metadata,
           Column('person', String(10), primary_key=True),
           Column('firstname', String(10)),
           Column('lastname', String(10)),
        )

        peoplesites = Table("peoplesites", metadata,
            Column('person', String(10), ForeignKey("people.person"),
        primary_key=True),
            Column('site', String(10)),
        )

    def test_basic(self):
        class PersonSite(object):pass
        class Person(object):pass
        m1 = mapper(PersonSite, peoplesites)

        m2 = mapper(Person, people,
              properties = {
                      'sites' : relation(PersonSite),
              },
            )
        compile_mappers()
        assert list(m2.get_property('sites').foreign_keys) == [peoplesites.c.person]
        p = Person()
        p.person = 'im the key'
        p.firstname = 'asdf'
        ps = PersonSite()
        ps.site = 'asdf'
        p.sites.append(ps)
        Session.commit()
        assert people.count(people.c.person=='im the key').scalar() == peoplesites.count(peoplesites.c.person=='im the key').scalar() == 1

class ClauseAttributesTest(ORMTest):
    def define_tables(self, metadata):
        global users_table
        users_table = Table('users', metadata,
            Column('id', Integer, Sequence('users_id_seq', optional=True), primary_key=True),
            Column('name', String(30)),
            Column('counter', Integer, default=1))

    def test_update(self):
        class User(object):
            pass
        mapper(User, users_table)
        u = User(name='test')
        sess = Session()
        sess.save(u)
        sess.flush()
        assert u.counter == 1
        u.counter = User.counter + 1
        sess.flush()
        def go():
            assert u.counter == 2
        self.assert_sql_count(testing.db, go, 1)

    def test_multi_update(self):
        class User(object):
            pass
        mapper(User, users_table)
        u = User(name='test')
        sess = Session()
        sess.save(u)
        sess.flush()
        assert u.counter == 1
        u.name = 'test2'
        u.counter = User.counter + 1
        sess.flush()
        def go():
            assert u.name == 'test2'
            assert u.counter == 2
        self.assert_sql_count(testing.db, go, 1)

        sess.clear()
        u = sess.query(User).get(u.id)
        assert u.name == 'test2'
        assert u.counter == 2

    @testing.unsupported('mssql')
    def test_insert(self):
        class User(object):
            pass
        mapper(User, users_table)
        u = User(name='test', counter=select([5]))
        sess = Session()
        sess.save(u)
        sess.flush()
        assert u.counter == 5


class PassiveDeletesTest(ORMTest):
    def define_tables(self, metadata):
        global mytable,myothertable

        mytable = Table('mytable', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)),
            test_needs_fk=True,
            )

        myothertable = Table('myothertable', metadata,
            Column('id', Integer, primary_key=True),
            Column('parent_id', Integer),
            Column('data', String(30)),
            ForeignKeyConstraint(['parent_id'],['mytable.id'], ondelete="CASCADE"),
            test_needs_fk=True,
            )

    @testing.unsupported('sqlite')
    def test_basic(self):
        class MyClass(object):
            pass
        class MyOtherClass(object):
            pass

        mapper(MyOtherClass, myothertable)

        mapper(MyClass, mytable, properties={
            'children':relation(MyOtherClass, passive_deletes=True, cascade="all")
        })

        sess = Session
        mc = MyClass()
        mc.children.append(MyOtherClass())
        mc.children.append(MyOtherClass())
        mc.children.append(MyOtherClass())
        mc.children.append(MyOtherClass())
        sess.save(mc)
        sess.commit()
        sess.close()
        assert myothertable.count().scalar() == 4
        mc = sess.query(MyClass).get(mc.id)
        sess.delete(mc)
        sess.commit()
        assert mytable.count().scalar() == 0
        assert myothertable.count().scalar() == 0

class ExtraPassiveDeletesTest(ORMTest):
    def define_tables(self, metadata):
        global mytable,myothertable

        mytable = Table('mytable', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)),
            test_needs_fk=True,
            )

        myothertable = Table('myothertable', metadata,
            Column('id', Integer, primary_key=True),
            Column('parent_id', Integer),
            Column('data', String(30)),
            ForeignKeyConstraint(['parent_id'],['mytable.id']),  # no CASCADE, the same as ON DELETE RESTRICT
            test_needs_fk=True,
            )

    def test_assertions(self):
        class MyClass(object):
            pass
        class MyOtherClass(object):
            pass

        mapper(MyOtherClass, myothertable)

        try:
            mapper(MyClass, mytable, properties={
                'children':relation(MyOtherClass, passive_deletes='all', cascade="all")
            })
            assert False
        except exceptions.ArgumentError, e:
            assert str(e) == "Can't set passive_deletes='all' in conjunction with 'delete' or 'delete-orphan' cascade"

    @testing.unsupported('sqlite')
    def test_extra_passive(self):
        class MyClass(object):
            pass
        class MyOtherClass(object):
            pass

        mapper(MyOtherClass, myothertable)

        mapper(MyClass, mytable, properties={
            'children':relation(MyOtherClass, passive_deletes='all', cascade="save-update")
        })

        sess = Session
        mc = MyClass()
        mc.children.append(MyOtherClass())
        mc.children.append(MyOtherClass())
        mc.children.append(MyOtherClass())
        mc.children.append(MyOtherClass())
        sess.save(mc)
        sess.commit()

        assert myothertable.count().scalar() == 4
        mc = sess.query(MyClass).get(mc.id)
        sess.delete(mc)
        try:
            sess.commit()
            assert False
        except exceptions.DBAPIError:
            assert True


class DefaultTest(ORMTest):
    """tests that when saving objects whose table contains DefaultGenerators, either python-side, preexec or database-side,
    the newly saved instances receive all the default values either through a post-fetch or getting the pre-exec'ed
    defaults back from the engine."""

    def define_tables(self, metadata):
        db = testing.db
        use_string_defaults = testing.against('postgres', 'oracle', 'sqlite') 
        global hohoval, althohoval

        if use_string_defaults:
            hohotype = String(30)
            hohoval = "im hoho"
            althohoval = "im different hoho"
        else:
            hohotype = Integer
            hohoval = 9
            althohoval = 15

        global default_table, secondary_table
        default_table = Table('default_test', metadata,
            Column('id', Integer, Sequence("dt_seq", optional=True), primary_key=True),
            Column('hoho', hohotype, PassiveDefault(str(hohoval))),
            Column('counter', Integer, default=func.length("1234567")),
            Column('foober', String(30), default="im foober", onupdate="im the update"),
        )
        
        secondary_table = Table('secondary_table', metadata, 
            Column('id', Integer, primary_key=True),
            Column('data', String(50))
            )
        
        if testing.against('postgres', 'oracle'):
            default_table.append_column(Column('secondary_id', Integer, Sequence('sec_id_seq'), unique=True))
            secondary_table.append_column(Column('fk_val', Integer, ForeignKey('default_test.secondary_id')))
        else:
            secondary_table.append_column(Column('hoho', hohotype, ForeignKey('default_test.hoho')))

    def test_insert(self):
        class Hoho(object):pass
        mapper(Hoho, default_table)

        h1 = Hoho(hoho=althohoval)
        h2 = Hoho(counter=12)
        h3 = Hoho(hoho=althohoval, counter=12)
        h4 = Hoho()
        h5 = Hoho(foober='im the new foober')
        Session.commit()

        self.assert_(h1.hoho==althohoval)
        self.assert_(h3.hoho==althohoval)

        def go():
            # test deferred load of attribues, one select per instance
            self.assert_(h2.hoho==h4.hoho==h5.hoho==hohoval)
        self.assert_sql_count(testing.db, go, 3)

        def go():
            self.assert_(h1.counter ==  h4.counter==h5.counter==7)