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.
# 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 ""
# 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")
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)
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:
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):
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
"""
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:
# 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'))
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',
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
+#! 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'
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
