defaults.py

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

             CREATE TABLE speedy_users
             (
                 speedy_user_id   SERIAL     PRIMARY KEY,

                 user_name        VARCHAR    NOT NULL,
                 user_password    VARCHAR    NOT NULL
             );
            """, None)

            t = Table("speedy_users", meta, autoload=True)
            t.insert().execute(user_name='user', user_password='lala')
            l = t.select().execute().fetchall()
            self.assert_(l == [(1, 'user', 'lala')])
        finally:
            testing.db.execute("drop table speedy_users", None)

class PKDefaultTest(TestBase):
    def setUpAll(self):
        global metadata, t1, t2

        metadata = MetaData(testing.db)

        t2 = Table('t2', metadata,
            Column('nextid', Integer))

        t1 = Table('t1', metadata,
            Column('id', Integer, primary_key=True, default=select([func.max(t2.c.nextid)]).as_scalar()),
            Column('data', String(30)))

        metadata.create_all()

    def tearDownAll(self):
        metadata.drop_all()

    @testing.unsupported('mssql')
    def test_basic(self):
        t2.insert().execute(nextid=1)
        r = t1.insert().execute(data='hi')
        assert r.last_inserted_ids() == [1]

        t2.insert().execute(nextid=2)
        r = t1.insert().execute(data='there')
        assert r.last_inserted_ids() == [2]


class AutoIncrementTest(TestBase):
    def setUp(self):
        global aitable, aimeta

        aimeta = MetaData(testing.db)
        aitable = Table("aitest", aimeta,
            Column('id', Integer, Sequence('ai_id_seq', optional=True),
                   primary_key=True),
            Column('int1', Integer),
            Column('str1', String(20)))
        aimeta.create_all()

    def tearDown(self):
        aimeta.drop_all()

    # should fail everywhere... was: @supported('postgres', 'mysql', 'maxdb')
    @testing.fails_on('sqlite')
    def testnonautoincrement(self):
        # sqlite INT primary keys can be non-unique! (only for ints)
        meta = MetaData(testing.db)
        nonai_table = Table("nonaitest", meta,
            Column('id', Integer, autoincrement=False, primary_key=True),
            Column('data', String(20)))
        nonai_table.create(checkfirst=True)
        try:
            try:
                # postgres + mysql strict will fail on first row,
                # mysql in legacy mode fails on second row
                nonai_table.insert().execute(data='row 1')
                nonai_table.insert().execute(data='row 2')
                assert False
            except exceptions.SQLError, e:
                print "Got exception", str(e)
                assert True

            nonai_table.insert().execute(id=1, data='row 1')
        finally:
            nonai_table.drop()

    # TODO: add coverage for increment on a secondary column in a key
    def _test_autoincrement(self, bind):
        ids = set()
        rs = bind.execute(aitable.insert(), int1=1)
        last = rs.last_inserted_ids()[0]
        self.assert_(last)
        self.assert_(last not in ids)
        ids.add(last)

        rs = bind.execute(aitable.insert(), str1='row 2')
        last = rs.last_inserted_ids()[0]
        self.assert_(last)
        self.assert_(last not in ids)
        ids.add(last)

        rs = bind.execute(aitable.insert(), int1=3, str1='row 3')
        last = rs.last_inserted_ids()[0]
        self.assert_(last)
        self.assert_(last not in ids)
        ids.add(last)

        rs = bind.execute(aitable.insert(values={'int1':func.length('four')}))
        last = rs.last_inserted_ids()[0]
        self.assert_(last)
        self.assert_(last not in ids)
        ids.add(last)

        self.assert_(
            list(bind.execute(aitable.select().order_by(aitable.c.id))) ==
            [(1, 1, None), (2, None, 'row 2'), (3, 3, 'row 3'), (4, 4, None)])

    def test_autoincrement_autocommit(self):
        self._test_autoincrement(testing.db)

    def test_autoincrement_transaction(self):
        con = testing.db.connect()
        tx = con.begin()
        try:
            try:
                self._test_autoincrement(con)
            except:
                try:
                    tx.rollback()
                except:
                    pass
                raise
            else:
                tx.commit()
        finally:
            con.close()

    def test_autoincrement_fk(self):
        if not testing.db.dialect.supports_pk_autoincrement:
            return True

        metadata = MetaData(testing.db)

        # No optional sequence here.
        nodes = Table('nodes', metadata,
            Column('id', Integer, primary_key=True),
            Column('parent_id', Integer, ForeignKey('nodes.id')),
            Column('data', String(30)))
        metadata.create_all()
        try:
            r = nodes.insert().execute(data='foo')
            id_ = r.last_inserted_ids()[0]
            nodes.insert().execute(data='bar', parent_id=id_)
        finally:
            metadata.drop_all()


class SequenceTest(TestBase):
    __unsupported_on__ = ('sqlite', 'mysql', 'mssql', 'firebird',
                          'sybase', 'access')

    def setUpAll(self):
        global cartitems, sometable, metadata
        metadata = MetaData(testing.db)
        cartitems = Table("cartitems", metadata,
            Column("cart_id", Integer, Sequence('cart_id_seq'), primary_key=True),
            Column("description", String(40)),
            Column("createdate", DateTime())
        )
        sometable = Table( 'Manager', metadata,
               Column('obj_id', Integer, Sequence('obj_id_seq'), ),
               Column('name', String(128)),
               Column('id', Integer, Sequence('Manager_id_seq', optional=True),
                      primary_key=True),
           )

        metadata.create_all()

    def testseqnonpk(self):
        """test sequences fire off as defaults on non-pk columns"""

        result = sometable.insert().execute(name="somename")
        assert 'id' in result.postfetch_cols()

        result = sometable.insert().execute(name="someother")
        assert 'id' in result.postfetch_cols()

        sometable.insert().execute(
            {'name':'name3'},
            {'name':'name4'}
        )
        assert sometable.select().execute().fetchall() == [
            (1, "somename", 1),
            (2, "someother", 2),
            (3, "name3", 3),
            (4, "name4", 4),
        ]

    def testsequence(self):
        cartitems.insert().execute(description='hi')
        cartitems.insert().execute(description='there')
        r = cartitems.insert().execute(description='lala')

        assert r.last_inserted_ids() and r.last_inserted_ids()[0] is not None
        id_ = r.last_inserted_ids()[0]

        assert select([func.count(cartitems.c.cart_id)],
                      and_(cartitems.c.description == 'lala',
                           cartitems.c.cart_id == id_)).scalar() == 1

        cartitems.select().execute().fetchall()


    @testing.fails_on('maxdb')
    # maxdb db-api seems to double-execute NEXTVAL internally somewhere,
    # throwing off the numbers for these tests...
    def test_implicit_sequence_exec(self):
        s = Sequence("my_sequence", metadata=MetaData(testing.db))
        s.create()
        try:
            x = s.execute()
            self.assert_(x == 1)
        finally:
            s.drop()

    @testing.fails_on('maxdb')
    def teststandalone_explicit(self):
        s = Sequence("my_sequence")
        s.create(bind=testing.db)
        try:
            x = s.execute(testing.db)
            self.assert_(x == 1)
        finally:
            s.drop(testing.db)

    def test_checkfirst(self):
        s = Sequence("my_sequence")
        s.create(testing.db, checkfirst=False)
        s.create(testing.db, checkfirst=True)
        s.drop(testing.db, checkfirst=False)
        s.drop(testing.db, checkfirst=True)

    @testing.fails_on('maxdb')
    def teststandalone2(self):
        x = cartitems.c.cart_id.sequence.execute()
        self.assert_(1 <= x <= 4)

    def tearDownAll(self):
        metadata.drop_all()


if __name__ == "__main__":
    testenv.main()