Commits

jason kirtland committed 0953ab8

- Emit BOOL rather than BOOLEAN for MySQL booleans in DDL, for old versions
of MySQL (#583)
- MySQL columns (such as times) with colons in their default values couldn't
be roundtripped, fixed (also in Postgres, but not fixed here.)
- BINARY/VARBINARY columns aren't really binary at all on ancient versions
of MySQL. The type.Binary(123) passthrough now always makes BLOBs.
Removed the short-lived MSBaseBinary.
- Added mysql.get_version_info, given a connectable returns a tuple of server
version info.
- Backed off on the reflection tests for older versions of MySQL, for now.

  • Participants
  • Parent commits abb1f2e

Comments (0)

Files changed (4)

 - mysql
     - Nearly all MySQL column types are now supported for declaration and
       reflection. Added NCHAR, NVARCHAR, VARBINARY, TINYBLOB, LONGBLOB, YEAR
-    - The sqltypes.Binary passthrough now builds a VARBINARY rather than a
-      BINARY if given a length
+    - The sqltypes.Binary passthrough now always builds a BLOB, avoiding
+      problems with very old database versions
     - support for column-level CHARACTER SET and COLLATE declarations,
       as well as ASCII, UNICODE, NATIONAL and BINARY shorthand.
 - firebird

File lib/sqlalchemy/databases/mysql.py

         # We'll actually generate the equiv. "NATIONAL CHAR" instead of "NCHAR".
         return self._extend("CHAR(%(length)s)" % {'length': self.length})
 
-class MSBaseBinary(sqltypes.Binary):
-    """Flexible binary type"""
-
-    def __init__(self, length=None, **kw):
-        """Flexibly construct a binary column type.  Will construct a
-        VARBINARY or BLOB depending on the length requested, if any.
-
-        length
-          Maximum data length, in bytes.
-        """
-        super(MSBaseBinary, self).__init__(length, **kw)
+class _BinaryType(sqltypes.Binary):
+    """MySQL binary types"""
 
     def get_col_spec(self):
-        if self.length and self.length <= 255:
-            return "VARBINARY(%d)" % self.length
+        if self.length:
+            return "BLOB(%d)" % self.length
         else:
             return "BLOB"
 
         else:
             return buffer(value)
 
-class MSVarBinary(MSBaseBinary):
+class MSVarBinary(_BinaryType):
     """MySQL VARBINARY type, for variable length binary data"""
 
     def __init__(self, length=None, **kw):
         else:
             return "BLOB"
 
-class MSBinary(MSBaseBinary):
+class MSBinary(_BinaryType):
     """MySQL BINARY type, for fixed length binary data"""
 
     def __init__(self, length=None, **kw):
         else:
             return buffer(value)
 
-class MSBlob(MSBaseBinary):
+class MSBlob(_BinaryType):
     """MySQL BLOB type, for binary data up to 2^16 bytes""" 
 
 
 
 class MSBoolean(sqltypes.Boolean):
     def get_col_spec(self):
-        return "BOOLEAN"
+        return "BOOL"
 
     def convert_result_value(self, value, dialect):
         if value is None:
     sqltypes.Date : MSDate,
     sqltypes.Time : MSTime,
     sqltypes.String : MSString,
-    sqltypes.Binary : MSVarBinary,
+    sqltypes.Binary : MSBlob,
     sqltypes.Boolean : MSBoolean,
     sqltypes.TEXT : MSText,
     sqltypes.CHAR: MSChar,
     sqltypes.NCHAR: MSNChar,
     sqltypes.TIMESTAMP: MSTimeStamp,
     sqltypes.BLOB: MSBlob,
-    MSBaseBinary: MSBaseBinary,
+    _BinaryType: _BinaryType,
 }
 
 
             else:
                 raise
 
+    def get_version_info(self, connectable):
+        if hasattr(connectable, 'connect'):
+            con = connectable.connect().connection
+        else:
+            con = connectable
+        version = []
+        for n in con.get_server_info().split('.'):
+            try:
+                version.append(int(n))
+            except ValueError:
+                version.append(n)
+        return tuple(version)
+
     def reflecttable(self, connection, table):
         # reference:  http://dev.mysql.com/doc/refman/5.0/en/name-case-sensitivity.html
         cs = connection.execute("show variables like 'lower_case_table_names'").fetchone()[1]
 
             colargs= []
             if default:
-                colargs.append(schema.PassiveDefault(sql.text(default)))
+                if col_type == 'timestamp' and default == 'CURRENT_TIMESTAMP':
+                    arg = sql.text(default)
+                else:
+                    arg = default
+                colargs.append(schema.PassiveDefault(arg))
             table.append_column(schema.Column(name, coltype, *colargs,
                                             **dict(primary_key=primary_key,
                                                    nullable=nullable,

File test/dialect/mysql.py

 
     @testbase.supported('mysql')
     def test_type_reflection(self):
+        # FIXME: older versions need their own test
+        if db.dialect.get_version_info(db) < (5, 0):
+            return
+
         # (ask_for, roundtripped_as_if_different)
         specs = [( String(), mysql.MSText(), ),
                  ( String(1), mysql.MSString(1), ),
                  ( Smallinteger(4), mysql.MSSmallInteger(4), ),
                  ( mysql.MSSmallInteger(), ),
                  ( mysql.MSSmallInteger(4), mysql.MSSmallInteger(4), ),
-                 ( Binary(3), mysql.MSVarBinary(3), ),
+                 ( Binary(3), mysql.MSBlob(3), ),
                  ( Binary(), mysql.MSBlob() ),
                  ( mysql.MSBinary(3), mysql.MSBinary(3), ),
-                 ( mysql.MSBaseBinary(), mysql.MSBlob(), ),
-                 ( mysql.MSBaseBinary(3), mysql.MSVarBinary(3), ),
                  ( mysql.MSVarBinary(3),),
                  ( mysql.MSVarBinary(), mysql.MSBlob()),
                  ( mysql.MSTinyBlob(),),
         m2 = BoundMetaData(db)
         rt = Table('mysql_types', m2, autoload=True)
 
+        #print
         expected = [len(c) > 1 and c[1] or c[0] for c in specs]
         for i, reflected in enumerate(rt.c):
             #print (reflected, specs[i][0], '->',
             #       reflected.type, '==', expected[i])
-            assert type(reflected.type) == type(expected[i])
+            assert isinstance(reflected.type, type(expected[i]))
 
-        #m.drop_all()
+        m.drop_all()
 
 if __name__ == "__main__":
     testbase.main()

File test/engine/reflection.py

         if use_string_defaults:
             deftype2 = String
             defval2 = "im a default"
+            deftype3 = DateTime
+            defval3 = '1999-09-09 00:00:00'
         else:
-            deftype2 = Integer
-            defval2 = "15"
+            deftype2, deftype3 = Integer, Integer
+            defval2, defval3 = "15", "16"
         
         meta = BoundMetaData(testbase.db)
         
             Column('test_passivedefault', deftype, PassiveDefault(defval)),
             Column('test_passivedefault2', Integer, PassiveDefault("5")),
             Column('test_passivedefault3', deftype2, PassiveDefault(defval2)),
+            Column('test_passivedefault4', deftype3, PassiveDefault(defval3)),
             Column('test9', Binary(100)),
             Column('test_numeric', Numeric(None, None)),
             mysql_engine='InnoDB'
             mysql_engine='InnoDB'
         )
         meta.drop_all()
-        
+
         users.create()
         addresses.create()