Mike Bayer avatar Mike Bayer committed bf62102

- [feature] Dialect-specific compilers now raise
CompileException for all type/statement compilation
issues, instead of InvalidRequestError or ArgumentError.
The DDL for CREATE TABLE will re-raise
CompileExceptions to include table/column information
for the problematic column. [ticket:2361]

Comments (0)

Files changed (12)

     constructs to sqlalchemy.sql namespace, though
     not part of __all__ as of yet.
 
+  - [feature] Dialect-specific compilers now raise
+    CompileException for all type/statement compilation 
+    issues, instead of InvalidRequestError or ArgumentError. 
+    The DDL for CREATE TABLE will re-raise 
+    CompileExceptions to include table/column information
+    for the problematic column.  [ticket:2361]
+
   - [bug] Fixed issue where the "required" exception
     would not be raised for bindparam() with required=True,
     if the statement were given no parameters at all.

lib/sqlalchemy/dialects/maxdb/base.py

                 # LIMIT.  Right?  Other dialects seem to get away with
                 # dropping order.
                 if select._limit:
-                    raise exc.InvalidRequestError(
+                    raise exc.CompileError(
                         "MaxDB does not support ORDER BY in subqueries")
                 else:
                     return ""

lib/sqlalchemy/dialects/mssql/base.py

             # to use ROW_NUMBER(), an ORDER BY is required.
             orderby = self.process(select._order_by_clause)
             if not orderby:
-                raise exc.InvalidRequestError('MSSQL requires an order_by when '
+                raise exc.CompileError('MSSQL requires an order_by when '
                                               'using an offset.')
 
             _offset = select._offset
                 colspec += " NULL"
 
         if column.table is None:
-            raise exc.InvalidRequestError(
+            raise exc.CompileError(
                             "mssql requires Table-bound columns " 
                             "in order to generate DDL")
 

lib/sqlalchemy/dialects/mysql/base.py

         if type_.length:
             return self._extend_string(type_, {}, "VARCHAR(%d)" % type_.length)
         else:
-            raise exc.InvalidRequestError(
+            raise exc.CompileError(
                     "VARCHAR requires a length on dialect %s" % 
                     self.dialect.name)
 
         if type_.length:
             return self._extend_string(type_, {'national':True}, "VARCHAR(%(length)s)" % {'length': type_.length})
         else:
-            raise exc.InvalidRequestError(
+            raise exc.CompileError(
                     "NVARCHAR requires a length on dialect %s" % 
                     self.dialect.name)
 

lib/sqlalchemy/dialects/postgresql/base.py

 
     def format_type(self, type_, use_schema=True):
         if not type_.name:
-            raise exc.ArgumentError("Postgresql ENUM type requires a name.")
+            raise exc.CompileError("Postgresql ENUM type requires a name.")
 
         name = self.quote(type_.name, type_.quote)
         if not self.omit_schema and use_schema and type_.schema is not None:

lib/sqlalchemy/dialects/sqlite/base.py

             return "CAST(STRFTIME('%s', %s) AS INTEGER)" % (
                 self.extract_map[extract.field], self.process(extract.expr, **kw))
         except KeyError:
-            raise exc.ArgumentError(
+            raise exc.CompileError(
                 "%s is not a valid extract argument." % extract.field)
 
     def limit_clause(self, select):

lib/sqlalchemy/dialects/sybase/base.py

                         self.dialect.type_compiler.process(column.type)
 
         if column.table is None:
-            raise exc.InvalidRequestError(
+            raise exc.CompileError(
                         "The Sybase dialect requires Table-bound "
                        "columns in order to generate DDL")
         seq_col = column.table._autoincrement_column

lib/sqlalchemy/sql/compiler.py

 """
 
 import re
+import sys
 from sqlalchemy import schema, engine, util, exc
 from sqlalchemy.sql import operators, functions, util as sql_util, \
     visitors
         # if only one primary key, specify it along with the column
         first_pk = False
         for column in table.columns:
-            text += separator
-            separator = ", \n"
-            text += "\t" + self.get_column_specification(
-                                            column, 
-                                            first_pk=column.primary_key and \
-                                            not first_pk
-                                        )
-            if column.primary_key:
-                first_pk = True
-            const = " ".join(self.process(constraint) \
-                            for constraint in column.constraints)
-            if const:
-                text += " " + const
+            try:
+                text += separator
+                separator = ", \n"
+                text += "\t" + self.get_column_specification(
+                                                column, 
+                                                first_pk=column.primary_key and \
+                                                not first_pk
+                                            )
+                if column.primary_key:
+                    first_pk = True
+                const = " ".join(self.process(constraint) \
+                                for constraint in column.constraints)
+                if const:
+                    text += " " + const
+            except exc.CompileError, ce:
+                # Py3K
+                #raise exc.CompileError("(in table '%s', column '%s'): %s" 
+                #                             % (
+                #                                table.description, 
+                #                                column.name, 
+                #                                ce.args[0]
+                #                            )) from ce
+                # Py2K
+                raise exc.CompileError("(in table '%s', column '%s'): %s" 
+                                            % (
+                                                table.description, 
+                                                column.name,
+                                                ce.args[0]
+                                            )), None, sys.exc_info()[2]
 
         const = self.create_table_constraints(table)
         if const:

test/dialect/test_mysql.py

 # coding: utf-8
 
-from test.lib.testing import eq_, assert_raises
+from test.lib.testing import eq_, assert_raises, assert_raises_message
 
 # Py2K
 import sets
             Unicode(),
         ):
             type_ = sqltypes.to_instance(type_)
-            assert_raises(exc.InvalidRequestError, type_.compile, dialect=mysql.dialect())
+            assert_raises_message(
+                exc.CompileError, 
+                "VARCHAR requires a length on dialect mysql",
+                type_.compile, 
+            dialect=mysql.dialect())
+
+            t1 = Table('sometable', MetaData(),
+                Column('somecolumn', type_)
+            )
+            assert_raises_message(
+                exc.CompileError,
+                r"\(in table 'sometable', column 'somecolumn'\)\: "
+                r"(?:N)?VARCHAR requires a length on dialect mysql",
+                schema.CreateTable(t1).compile,
+                dialect=mysql.dialect()
+            )
 
     def test_update_limit(self):
         t = sql.table('t', sql.column('col1'), sql.column('col2'))

test/dialect/test_postgresql.py

     def test_name_required(self):
         metadata = MetaData(testing.db)
         etype = Enum('four', 'five', 'six', metadata=metadata)
-        assert_raises(exc.ArgumentError, etype.create)
-        assert_raises(exc.ArgumentError, etype.compile,
+        assert_raises(exc.CompileError, etype.create)
+        assert_raises(exc.CompileError, etype.compile,
                       dialect=postgresql.dialect())
 
     @testing.fails_on('postgresql+zxjdbc',

test/lib/testing.py

         callable_(*args, **kwargs)
         assert False, "Callable did not raise an exception"
     except except_cls, e:
-        assert re.search(msg, str(e)), "%r !~ %s" % (msg, e)
-        print str(e)
+        assert re.search(msg, unicode(e)), u"%r !~ %s" % (msg, e)
+        print unicode(e).encode('utf-8')
 
 def fail(msg):
     assert False, msg

test/sql/test_compiler.py

+#! coding:utf-8
+
 from test.lib.testing import eq_, assert_raises, assert_raises_message
 import datetime, re, operator, decimal
 from sqlalchemy import *
-from sqlalchemy import exc, sql, util
+from sqlalchemy import exc, sql, util, types, schema
 from sqlalchemy.sql import table, column, label, compiler
 from sqlalchemy.sql.expression import ClauseList, _literal_as_text
 from sqlalchemy.engine import default
 from sqlalchemy.databases import *
 from test.lib import *
+from sqlalchemy.ext.compiler import compiles
 
 table1 = table('mytable',
     column('myid', Integer),
             "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :foo_id_1"
         )
 
+class DDLTest(fixtures.TestBase, AssertsCompiledSQL):
+    __dialect__ = 'default'
+
+    def _illegal_type_fixture(self):
+        class MyType(types.TypeEngine):
+            pass
+        @compiles(MyType)
+        def compile(element, compiler, **kw):
+            raise exc.CompileError("Couldn't compile type")
+        return MyType
+
+    def test_reraise_of_column_spec_issue(self):
+        MyType = self._illegal_type_fixture()
+        t1 = Table('t', MetaData(),
+            Column('x', MyType())
+        )
+        assert_raises_message(
+            exc.CompileError,
+            r"\(in table 't', column 'x'\): Couldn't compile type",
+            schema.CreateTable(t1).compile
+        )
+
+    def test_reraise_of_column_spec_issue_unicode(self):
+        MyType = self._illegal_type_fixture()
+        t1 = Table('t', MetaData(),
+            Column(u'méil', MyType())
+        )
+        assert_raises_message(
+            exc.CompileError,
+            ur"\(in table 't', column 'méil'\): Couldn't compile type",
+            schema.CreateTable(t1).compile
+        )
+
+
 class InlineDefaultTest(fixtures.TestBase, AssertsCompiledSQL):
     __dialect__ = 'default'
 
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.