我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.exc.CompileError()。
def get_column_specification(self, column, **kwargs): if column.table is None: raise exc.CompileError( "access requires Table-bound columns " "in order to generate DDL") colspec = self.preparer.format_column(column) seq_col = column.table._autoincrement_column if seq_col is column: colspec += " AUTOINCREMENT" else: colspec += " " + self.dialect.type_compiler.process(column.type) if column.nullable is not None and not column.primary_key: if not column.nullable or column.primary_key: colspec += " NOT NULL" else: colspec += " NULL" default = self.get_column_default_string(column) if default is not None: colspec += " DEFAULT " + default return colspec
def get_column_specification(self, column, **kwargs): if column.table is None: raise exc.CompileError( "drill requires Table-bound columns " "in order to generate DDL") colspec = self.preparer.format_column(column) seq_col = column.table._autoincrement_column if seq_col is column: colspec += " AUTOINCREMENT" else: colspec += " " + self.dialect.type_compiler.process(column.type) if column.nullable is not None and not column.primary_key: if not column.nullable or column.primary_key: colspec += " NOT NULL" else: colspec += " NULL" default = self.get_column_default_string(column) if default is not None: colspec += " DEFAULT " + default return colspec
def get_column_specification(self, column, **kwargs): if column.table is None: raise exc.CompileError( "Teradata requires Table-bound columns " "in order to generate DDL") colspec = (self.preparer.format_column(column) + " " +\ self.dialect.type_compiler.process( column.type, type_expression=column)) # Null/NotNull if column.nullable is not None: if not column.nullable or column.primary_key: colspec += " NOT NULL" return colspec
def test_lambda_no_arg_kwargs(self): with self.assertRaises(exc.CompileError) as ex: self.compile(Lambda(lambda x, *args: x * 2)) self.assertEqual( str(ex.exception), 'Lambdas with *args are not supported' ) with self.assertRaises(exc.CompileError) as ex: self.compile(Lambda(lambda x, **kwargs: x * 2)) self.assertEqual( str(ex.exception), 'Lambdas with **kwargs are not supported' )
def visit_lambda(self, lambda_, **kw): func = lambda_.func spec = inspect_getargspec(func) if spec.varargs: raise exc.CompileError('Lambdas with *args are not supported') if spec.keywords: raise exc.CompileError('Lambdas with **kwargs are not supported') text = ', '.join(spec.args) + ' -> ' args = [literal_column(arg) for arg in spec.args] text += self.process(func(*args), **kw) return text
def visit_VARCHAR(self, type_, **kw): if not type_.length: raise exc.CompileError( "VARCHAR requires a length on dialect %s" % self.dialect.name) basic = super(FBTypeCompiler, self).visit_VARCHAR(type_, **kw) return self._extend_string(type_, basic)
def get_column_specification(self, column, **kwargs): colspec = self.preparer.format_column(column) + " " + \ self.dialect.type_compiler.process( column.type, type_expression=column) if column.table is None: raise exc.CompileError( "The Sybase dialect requires Table-bound " "columns in order to generate DDL") seq_col = column.table._autoincrement_column # install a IDENTITY Sequence if we have an implicit IDENTITY column if seq_col is column: sequence = isinstance(column.default, sa_schema.Sequence) \ and column.default if sequence: start, increment = sequence.start or 1, \ sequence.increment or 1 else: start, increment = 1, 1 if (start, increment) == (1, 1): colspec += " IDENTITY" else: # TODO: need correct syntax for this colspec += " IDENTITY(%s,%s)" % (start, increment) else: default = self.get_column_default_string(column) if default is not None: colspec += " DEFAULT " + default if column.nullable is not None: if not column.nullable or column.primary_key: colspec += " NOT NULL" else: colspec += " NULL" return colspec
def visit_VARCHAR(self, type_): if not type_.length: raise exc.CompileError( "VARCHAR requires a length on dialect %s" % self.dialect.name) basic = super(FBTypeCompiler, self).visit_VARCHAR(type_) return self._extend_string(type_, basic)
def get_column_specification(self, column, **kwargs): colspec = self.preparer.format_column(column) + " " + \ self.dialect.type_compiler.process(column.type) if column.table is None: raise exc.CompileError( "The Sybase dialect requires Table-bound " "columns in order to generate DDL") seq_col = column.table._autoincrement_column # install a IDENTITY Sequence if we have an implicit IDENTITY column if seq_col is column: sequence = isinstance(column.default, sa_schema.Sequence) \ and column.default if sequence: start, increment = sequence.start or 1, \ sequence.increment or 1 else: start, increment = 1, 1 if (start, increment) == (1, 1): colspec += " IDENTITY" else: # TODO: need correct syntax for this colspec += " IDENTITY(%s,%s)" % (start, increment) else: default = self.get_column_default_string(column) if default is not None: colspec += " DEFAULT " + default if column.nullable is not None: if not column.nullable or column.primary_key: colspec += " NOT NULL" else: colspec += " NULL" return colspec
def visit_extract(self, extract, **kw): try: return "CAST(STRFTIME('%s', %s) AS INTEGER)" % ( self.extract_map[extract.field], self.process(extract.expr, **kw) ) except KeyError: raise exc.CompileError( "%s is not a valid extract argument." % extract.field)
def compile_sqlite_sign(element, compiler, **kw): args = list(element.clauses) if len(args) != 1: raise CompileError("Sign function takes exactly one argument.") return ( "CASE WHEN {0} < 0 THEN -1 " "ELSE CASE WHEN {0} > 0 THEN 1 " "ELSE 0 END END".format( compiler.process(args[0]) ) )
def test_unsupported_expressoin(self): t1, t2 = self.create_tables(2) query = session.query(t1.c.x).join(t2, literal(True), any=True) with self.assertRaises(exc.CompileError) as ex: self.compile(query) self.assertIn('Only tuple elements are supported', str(ex.exception))
def test_offset_without_limit(self): table = self.create_table() with self.assertRaises(exc.CompileError) as ex: query = session.query(table.c.x).offset(5) self.compile(query, literal_binds=True) self.assertEqual( str(ex.exception), 'OFFSET without LIMIT is not supported' )
def test_else_required(self): with self.assertRaises(exc.CompileError) as ex: self.compile(case([(literal(1), 0)])) self.assertEqual( str(ex.exception), 'ELSE clause is required in CASE' )
def test_create_table_without_engine(self): no_engine_table = Table( 't1', self.metadata(), Column('x', types.Int32, primary_key=True), Column('y', types.String) ) with self.assertRaises(exc.CompileError) as ex: self.compile(CreateTable(no_engine_table)) self.assertEqual(str(ex.exception), "No engine for table 't1'")
def limit_clause(self, select, **kw): text = '' if select._limit_clause is not None: text += ' \n LIMIT ' if select._offset_clause is not None: text += self.process(select._offset_clause, **kw) + ', ' text += self.process(select._limit_clause, **kw) else: if select._offset_clause is not None: raise exc.CompileError('OFFSET without LIMIT is not supported') return text
def visit_join(self, join, asfrom=False, **kwargs): join_type = " " if join.global_: join_type += "GLOBAL " if join.any: join_type += "ANY " if join.all: join_type += "ALL " if join.isouter: join_type += "LEFT OUTER JOIN " else: join_type += "INNER JOIN " if not isinstance(join.onclause, elements.Tuple): raise exc.CompileError( "Only tuple elements are supported. " "Got: %s" % type(join.onclause) ) return ( join.left._compiler_dispatch(self, asfrom=True, **kwargs) + join_type + join.right._compiler_dispatch(self, asfrom=True, **kwargs) + " USING " + join.onclause._compiler_dispatch(self, **kwargs) )
def post_create_table(self, table): engine = getattr(table, 'engine', None) if not engine: raise exc.CompileError("No engine for table '%s'" % table.name) return ' ENGINE = ' + self.process(engine)
