1. diana clarke
  2. sqlalchemy-2385


Mike Bayer  committed a5fb8ce

fixes to types so that database-specific types more easily used;
fixes to mysql text types to work with this methodology

  • Participants
  • Parent commits 54b4fec
  • Branches default

Comments (0)

Files changed (4)


View file
 - SelectResults will use a subselect, when calling an aggregate (i.e.
 max, min, etc.) on a SelectResults that has an ORDER BY clause
+- fixes to types so that database-specific types more easily used;
+fixes to mysql text types to work with this methodology
 - big overhaul to schema to allow truly composite primary and foreign

File lib/sqlalchemy/databases/mysql.py

View file
         super(MSText, self).__init__()
     def get_col_spec(self):
         return "TEXT"
-class MSTinyText(sqltypes.TEXT):
-    def __init__(self, **kw):
-        self.binary = 'binary' in kw
-        super(MSTinyText, self).__init__()
+class MSTinyText(MSText):
     def get_col_spec(self):
         if self.binary:
             return "TEXT BINARY"
            return "TEXT"
-class MSMediumText(sqltypes.TEXT):
-    def __init__(self, **kw):
-        self.binary = 'binary' in kw
-        super(MSMediumText, self).__init__()
+class MSMediumText(MSText):
     def get_col_spec(self):
         if self.binary:
             return "MEDIUMTEXT BINARY"
             return "MEDIUMTEXT"
-class MSLongText(sqltypes.TEXT):
-    def __init__(self, **kw):
-        self.binary = 'binary' in kw
-        super(MSLongText, self).__init__()
+class MSLongText(MSText):
     def get_col_spec(self):
         if self.binary:
             return "LONGTEXT BINARY"
 class MySQLSchemaGenerator(ansisql.ANSISchemaGenerator):
     def get_column_specification(self, column, override_pk=False, first_pk=False):
+        t = column.type.engine_impl(self.engine)
         colspec = self.preparer.format_column(column) + " " + column.type.engine_impl(self.engine).get_col_spec()
         default = self.get_column_default_string(column)
         if default is not None:

File lib/sqlalchemy/types.py

View file
     def dialect_impl(self, dialect):
             return self.impl_dict[dialect]
-        except:
+        except KeyError:
             return self.impl_dict.setdefault(dialect, dialect.type_descriptor(self))
     def _get_impl(self):
         if hasattr(self, '_impl'):
 def adapt_type(typeobj, colspecs):
     if isinstance(typeobj, type):
         typeobj = typeobj()
     for t in typeobj.__class__.__mro__[0:-1]:
             impltype = colspecs[t]
         # couldnt adapt...raise exception ?
         return typeobj
+    # if we adapted the given generic type to a database-specific type, 
+    # but it turns out the originally given "generic" type
+    # is actually a subclass of our resulting type, then we were already
+    # were given a more specific type than that required; so use that.
+    if (issubclass(typeobj.__class__, impltype)):
+        return typeobj
     return typeobj.adapt(impltype)
 class NullTypeEngine(TypeEngine):

File test/engine/reflection.py

View file
 from sqlalchemy import *
 from sqlalchemy.exceptions import NoSuchTableError
+import sqlalchemy.databases.mysql as mysql
 import unittest, re, StringIO
+    @testbase.supported('mysql')
+    def testmysqltypes(self):
+        meta1 = BoundMetaData(testbase.db)
+        table = Table(
+            'mysql_types', meta1,
+            Column('id', Integer, primary_key=True),
+            Column('num1', mysql.MSInteger(unsigned=True)),
+            Column('text1', mysql.MSLongText),
+            Column('text2', mysql.MSLongText())
+            )
+        try:
+            table.create(checkfirst=True)
+            meta2 = BoundMetaData(testbase.db)
+            t2 = Table('mysql_types', meta2, autoload=True)
+            assert isinstance(t2.c.num1.type, mysql.MSInteger)
+            assert t2.c.num1.type.unsigned
+            assert isinstance(t2.c.text1.type, mysql.MSLongText)
+            assert isinstance(t2.c.text2.type, mysql.MSLongText)
+            t2.drop()
+            t2.create()
+        finally:
+            table.drop(checkfirst=True)
     def testmultipk(self):
         table = Table(
             'engine_multi', testbase.db,