mysql.py

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

                roundtrip([1] * 8)
                roundtrip([sql.text("b'1'")] * 8, [1] * 8)

                i = 255
                roundtrip([0, 0, 0, 0, i, i, i, i])
                i = 2**32 - 1
                roundtrip([0, 0, 0, 0, 0, i, i, i])
                i = 2**63 - 1
                roundtrip([0, 0, 0, 0, 0, 0, i, i])
                i = 2**64 - 1
                roundtrip([0, 0, 0, 0, 0, 0, 0, i])
        finally:
            meta.drop_all()

    def test_boolean(self):
        """Test BOOL/TINYINT(1) compatability and reflection."""

        meta = MetaData(testing.db)
        bool_table = Table('mysql_bool', meta,
                           Column('b1', BOOLEAN),
                           Column('b2', mysql.MSBoolean),
                           Column('b3', mysql.MSTinyInteger(1)),
                           Column('b4', mysql.MSTinyInteger))

        self.assert_eq(colspec(bool_table.c.b1), 'b1 BOOL')
        self.assert_eq(colspec(bool_table.c.b2), 'b2 BOOL')
        self.assert_eq(colspec(bool_table.c.b3), 'b3 TINYINT(1)')
        self.assert_eq(colspec(bool_table.c.b4), 'b4 TINYINT')

        for col in bool_table.c:
            self.assert_(repr(col))
        try:
            meta.create_all()

            table = bool_table
            def roundtrip(store, expected=None):
                expected = expected or store
                table.insert(store).execute()
                row = list(table.select().execute())[0]
                try:
                    self.assert_(list(row) == expected)
                    for i, val in enumerate(expected):
                        if isinstance(val, bool):
                            self.assert_(val is row[i])
                except:
                    print "Storing %s" % store
                    print "Expected %s" % expected
                    print "Found %s" % list(row)
                    raise
                table.delete().execute()


            roundtrip([None, None, None, None])
            roundtrip([True, True, 1, 1])
            roundtrip([False, False, 0, 0])
            roundtrip([True, True, True, True], [True, True, 1, 1])
            roundtrip([False, False, 0, 0], [False, False, 0, 0])

            meta2 = MetaData(testing.db)
            # replace with reflected
            table = Table('mysql_bool', meta2, autoload=True)
            self.assert_eq(colspec(table.c.b3), 'b3 BOOL')

            roundtrip([None, None, None, None])
            roundtrip([True, True, 1, 1], [True, True, True, 1])
            roundtrip([False, False, 0, 0], [False, False, False, 0])
            roundtrip([True, True, True, True], [True, True, True, 1])
            roundtrip([False, False, 0, 0], [False, False, False, 0])
        finally:
            meta.drop_all()

    @testing.exclude('mysql', '<', (4, 1, 0))
    def test_timestamp(self):
        """Exercise funky TIMESTAMP default syntax."""

        meta = MetaData(testing.db)

        try:
            columns = [
                ([TIMESTAMP],
                 'TIMESTAMP'),
                ([mysql.MSTimeStamp],
                 'TIMESTAMP'),
                ([mysql.MSTimeStamp,
                  PassiveDefault(sql.text('CURRENT_TIMESTAMP'))],
                 "TIMESTAMP DEFAULT CURRENT_TIMESTAMP"),
                ([mysql.MSTimeStamp,
                  PassiveDefault(sql.text("'1999-09-09 09:09:09'"))],
                 "TIMESTAMP DEFAULT '1999-09-09 09:09:09'"),
                ([mysql.MSTimeStamp,
                  PassiveDefault(sql.text("'1999-09-09 09:09:09' "
                                          "ON UPDATE CURRENT_TIMESTAMP"))],
                 "TIMESTAMP DEFAULT '1999-09-09 09:09:09' "
                 "ON UPDATE CURRENT_TIMESTAMP"),
                ([mysql.MSTimeStamp,
                  PassiveDefault(sql.text("CURRENT_TIMESTAMP "
                                          "ON UPDATE CURRENT_TIMESTAMP"))],
                 "TIMESTAMP DEFAULT CURRENT_TIMESTAMP "
                 "ON UPDATE CURRENT_TIMESTAMP"),
                ]
            for idx, (spec, expected) in enumerate(columns):
                t = Table('mysql_ts%s' % idx, meta,
                          Column('id', Integer, primary_key=True),
                          Column('t', *spec))
                self.assert_eq(colspec(t.c.t), "t %s" % expected)
                self.assert_(repr(t.c.t))
                t.create()
                r = Table('mysql_ts%s' % idx, MetaData(testing.db),
                          autoload=True)
                if len(spec) > 1:
                    self.assert_(r.c.t is not None)
        finally:
            meta.drop_all()

    def test_year(self):
        """Exercise YEAR."""

        meta = MetaData(testing.db)
        year_table = Table('mysql_year', meta,
                           Column('y1', mysql.MSYear),
                           Column('y2', mysql.MSYear),
                           Column('y3', mysql.MSYear),
                           Column('y4', mysql.MSYear(2)),
                           Column('y5', mysql.MSYear(4)))

        for col in year_table.c:
            self.assert_(repr(col))
        try:
            year_table.create()
            reflected = Table('mysql_year', MetaData(testing.db),
                              autoload=True)

            for table in year_table, reflected:
                table.insert(['1950', '50', None, 50, 1950]).execute()
                row = list(table.select().execute())[0]
                self.assert_eq(list(row), [1950, 2050, None, 50, 1950])
                table.delete().execute()
                self.assert_(colspec(table.c.y1).startswith('y1 YEAR'))
                self.assert_eq(colspec(table.c.y4), 'y4 YEAR(2)')
                self.assert_eq(colspec(table.c.y5), 'y5 YEAR(4)')
        finally:
            meta.drop_all()


    def test_set(self):
        """Exercise the SET type."""

        meta = MetaData(testing.db)
        set_table = Table('mysql_set', meta,
                          Column('s1', mysql.MSSet("'dq'", "'sq'")),
                          Column('s2', mysql.MSSet("'a'")),
                          Column('s3', mysql.MSSet("'5'", "'7'", "'9'")))

        self.assert_eq(colspec(set_table.c.s1), "s1 SET('dq','sq')")
        self.assert_eq(colspec(set_table.c.s2), "s2 SET('a')")
        self.assert_eq(colspec(set_table.c.s3), "s3 SET('5','7','9')")

        for col in set_table.c:
            self.assert_(repr(col))
        try:
            set_table.create()
            reflected = Table('mysql_set', MetaData(testing.db),
                              autoload=True)

            for table in set_table, reflected:
                def roundtrip(store, expected=None):
                    expected = expected or store
                    table.insert(store).execute()
                    row = list(table.select().execute())[0]
                    try:
                        self.assert_(list(row) == expected)
                    except:
                        print "Storing %s" % store
                        print "Expected %s" % expected
                        print "Found %s" % list(row)
                        raise
                    table.delete().execute()

                roundtrip([None, None, None],[None] * 3)
                roundtrip(['', '', ''], [set([''])] * 3)

                roundtrip([set(['dq']), set(['a']), set(['5'])])
                roundtrip(['dq', 'a', '5'],
                          [set(['dq']), set(['a']), set(['5'])])
                roundtrip([1, 1, 1],
                          [set(['dq']), set(['a']), set(['5'])])
                roundtrip([set(['dq', 'sq']), None, set(['9', '5', '7'])])

            set_table.insert().execute({'s3':set(['5'])},
                                       {'s3':set(['5', '7'])},
                                       {'s3':set(['5', '7', '9'])},
                                       {'s3':set(['7', '9'])})
            rows = list(select(
                [set_table.c.s3],
                set_table.c.s3.in_([set(['5']), set(['5', '7'])])).execute())
            found = set([frozenset(row[0]) for row in rows])
            self.assertEquals(found,
                              set([frozenset(['5']), frozenset(['5', '7'])]))
        finally:
            meta.drop_all()

    def test_enum(self):
        """Exercise the ENUM type."""

        db = testing.db
        enum_table = Table('mysql_enum', MetaData(testing.db),
            Column('e1', mysql.MSEnum("'a'", "'b'")),
            Column('e2', mysql.MSEnum("'a'", "'b'"),
                   nullable=False),
            Column('e3', mysql.MSEnum("'a'", "'b'", strict=True)),
            Column('e4', mysql.MSEnum("'a'", "'b'", strict=True),
                   nullable=False))

        self.assert_eq(colspec(enum_table.c.e1),
                       "e1 ENUM('a','b')")
        self.assert_eq(colspec(enum_table.c.e2),
                       "e2 ENUM('a','b') NOT NULL")
        self.assert_eq(colspec(enum_table.c.e3),
                       "e3 ENUM('a','b')")
        self.assert_eq(colspec(enum_table.c.e4),
                       "e4 ENUM('a','b') NOT NULL")
        enum_table.drop(checkfirst=True)
        enum_table.create()

        try:
            enum_table.insert().execute(e1=None, e2=None, e3=None, e4=None)
            self.assert_(False)
        except exceptions.SQLError:
            self.assert_(True)

        try:
            enum_table.insert().execute(e1='c', e2='c', e3='c', e4='c')
            self.assert_(False)
        except exceptions.InvalidRequestError:
            self.assert_(True)

        enum_table.insert().execute()
        enum_table.insert().execute(e1='a', e2='a', e3='a', e4='a')
        enum_table.insert().execute(e1='b', e2='b', e3='b', e4='b')

        res = enum_table.select().execute().fetchall()

        expected = [(None, 'a', None, 'a'),
                    ('a', 'a', 'a', 'a'),
                    ('b', 'b', 'b', 'b')]

        # This is known to fail with MySQLDB 1.2.2 beta versions
        # which return these as sets.Set(['a']), sets.Set(['b'])
        # (even on Pythons with __builtin__.set)
        if testing.db.dialect.dbapi.version_info < (1, 2, 2, 'beta', 3) and \
           testing.db.dialect.dbapi.version_info >= (1, 2, 2):
            # these mysqldb seem to always uses 'sets', even on later pythons
            import sets
            def convert(value):
                if value is None:
                    return value
                if value == '':
                    return sets.Set([])
                else:
                    return sets.Set([value])

            e = []
            for row in expected:
                e.append(tuple([convert(c) for c in row]))
            expected = e

        self.assert_eq(res, expected)
        enum_table.drop()

    @testing.exclude('mysql', '>', (3))
    def test_enum_parse(self):
        """More exercises for the ENUM type."""

        # MySQL 3.23 can't handle an ENUM of ''....

        enum_table = Table('mysql_enum', MetaData(testing.db),
            Column('e1', mysql.MSEnum("'a'")),
            Column('e2', mysql.MSEnum("''")),
            Column('e3', mysql.MSEnum("'a'", "''")),
            Column('e4', mysql.MSEnum("''", "'a'")),
            Column('e5', mysql.MSEnum("''", "'''a'''", "'b''b'", "''''")))

        for col in enum_table.c:
            self.assert_(repr(col))
        try:
            enum_table.create()
            reflected = Table('mysql_enum', MetaData(testing.db),
                              autoload=True)
            for t in enum_table, reflected:
                assert t.c.e1.type.enums == ["a"]
                assert t.c.e2.type.enums == [""]
                assert t.c.e3.type.enums == ["a", ""]
                assert t.c.e4.type.enums == ["", "a"]
                assert t.c.e5.type.enums == ["", "'a'", "b'b", "'"]
        finally:
            enum_table.drop()

    def test_default_reflection(self):
        """Test reflection of column defaults."""

        def_table = Table('mysql_def', MetaData(testing.db),
            Column('c1', String(10), PassiveDefault('')),
            Column('c2', String(10), PassiveDefault('0')),
            Column('c3', String(10), PassiveDefault('abc')))

        try:
            def_table.create()
            reflected = Table('mysql_def', MetaData(testing.db),
                              autoload=True)
            for t in def_table, reflected:
                assert t.c.c1.default.arg == ''
                assert t.c.c2.default.arg == '0'