Uses a straight CheckConstraint with a generic expression. Preparing for boolean
constraint in [ticket:1589]
- CheckConstraint now accepts SQL expressions, though support for quoting of values
will be very limited. we don't want to get into formatting dates and such.
return ' '.join(colspec)
- def visit_enum_constraint(self, constraint):
- if not constraint.type.native_enum:
- return super(MySQLDDLCompiler, self).visit_enum_constraint(constraint)
- else:
- return None
-
def post_create_table(self, table):
"""Build table-level CREATE options like ENGINE and COLLATE."""
# identifiers are 64, however aliases can be 255...
max_identifier_length = 255
+ supports_native_enum = True
+ supports_native_boolean = True
+
supports_sane_rowcount = True
supports_sane_multi_rowcount = False
from sqlalchemy import schema as sa_schema
from sqlalchemy import sql, schema, exc, util
from sqlalchemy.engine import base, default, reflection
-from sqlalchemy.sql import compiler, expression
+from sqlalchemy.sql import compiler, expression, util as sql_util
from sqlalchemy.sql import operators as sql_operators
from sqlalchemy import types as sqltypes
colspec += " NOT NULL"
return colspec
- def visit_enum_constraint(self, constraint):
- if not constraint.type.native_enum:
- return super(PGDDLCompiler, self).visit_enum_constraint(constraint)
-
def visit_create_enum_type(self, create):
type_ = create.element
whereclause = None
if whereclause is not None:
- compiler = self._compile(whereclause, None)
- # this might belong to the compiler class
- inlined_clause = str(compiler) % dict(
- [(key,bind.value) for key,bind in compiler.binds.iteritems()])
- text += " WHERE " + inlined_clause
+ whereclause = sql_util.expression_as_ddl(whereclause)
+ where_compiled = self.sql_compiler.process(whereclause)
+ text += " WHERE " + where_compiled
return text
max_identifier_length = 63
supports_sane_rowcount = True
+ supports_native_enum = True
+
supports_sequences = True
sequences_optional = True
preexecute_autoincrement_sequences = True
supports_default_values
Indicates if the construct ``INSERT INTO tablename DEFAULT
VALUES`` is supported
+
+ supports_sequences
+ Indicates if the dialect supports CREATE SEQUENCE or similar.
+
+ sequences_optional
+ If True, indicates if the "optional" flag on the Sequence() construct
+ should signal to not generate a CREATE SEQUENCE. Applies only to
+ dialects that support sequences. Currently used only to allow Postgresql
+ SERIAL to be used on a column that specifies Sequence() for usage on
+ other backends.
+
+ supports_native_enum
+ Indicates if the dialect supports a native ENUM construct.
+ This will prevent types.Enum from generating a CHECK
+ constraint when that type is used.
+
+ supports_native_boolean
+ Indicates if the dialect supports a native boolean construct.
+ This will prevent types.Boolean from generating a CHECK
+ constraint when that type is used.
+
"""
def create_connect_args(self, url):
postfetch_lastrowid = True
implicit_returning = False
+ supports_native_enum = False
+ supports_native_boolean = False
+
# Py3K
#supports_unicode_statements = True
#supports_unicode_binds = True
__visit_name__ = 'constraint'
- def __init__(self, name=None, deferrable=None, initially=None, inline_ddl=True):
+ def __init__(self, name=None, deferrable=None, initially=None,
+ _create_rule=None):
"""Create a SQL constraint.
name
Optional string. If set, emit INITIALLY <value> when issuing DDL
for this constraint.
- inline_ddl
- if True, DDL for this Constraint will be generated within the span of a
- CREATE TABLE or DROP TABLE statement, when the associated table's
- DDL is generated. if False, no DDL is issued within that process.
- Instead, it is expected that an AddConstraint or DropConstraint
- construct will be used to issue DDL for this Contraint.
- The AddConstraint/DropConstraint constructs set this flag automatically
- as well.
+ _create_rule
+ a callable which is passed the DDLCompiler object during
+ compilation. Returns True or False to signal inline generation of
+ this Constraint.
+
+ The AddConstraint and DropConstraint DDL constructs provide
+ DDLElement's more comprehensive "conditional DDL" approach that is
+ passed a database connection when DDL is being issued. _create_rule
+ is instead called during any CREATE TABLE compilation, where there
+ may not be any transaction/connection in progress. However, it
+ allows conditional compilation of the constraint even for backends
+ which do not support addition of constraints through ALTER TABLE,
+ which currently includes SQLite.
+
+ _create_rule is used by some types to create constraints.
+ Currently, its call signature is subject to change at any time.
+
"""
self.name = name
self.deferrable = deferrable
self.initially = initially
- self.inline_ddl = inline_ddl
+ self._create_rule = _create_rule
@property
def table(self):
Can be included in the definition of a Table or Column.
"""
- def __init__(self, sqltext, name=None, deferrable=None, initially=None, table=None):
+ def __init__(self, sqltext, name=None, deferrable=None, initially=None, table=None, _create_rule=None):
"""Construct a CHECK constraint.
sqltext
- A string containing the constraint definition. Will be used
- verbatim.
-
+ A string containing the constraint definition, which will be used
+ verbatim, or a SQL expression construct.
+
name
Optional, the in-database name of the constraint.
"""
- super(CheckConstraint, self).__init__(name, deferrable, initially)
- if not isinstance(sqltext, basestring):
- raise exc.ArgumentError(
- "sqltext must be a string and will be used verbatim.")
- self.sqltext = sqltext
+ super(CheckConstraint, self).__init__(name, deferrable, initially, _create_rule)
+ self.sqltext = expression._literal_as_text(sqltext)
if table is not None:
self._set_parent(table)
self._check_ddl_on(on)
self.on = on
self.bind = bind
- element.inline_ddl = False
class CreateTable(_CreateDropBase):
"""Represent a CREATE TABLE statement."""
__visit_name__ = "add_constraint"
+ def __init__(self, element, *args, **kw):
+ super(AddConstraint, self).__init__(element, *args, **kw)
+ element._create_rule = lambda compiler: False
+
class DropConstraint(_CreateDropBase):
"""Represent an ALTER TABLE DROP CONSTRAINT statement."""
def __init__(self, element, cascade=False, **kw):
self.cascade = cascade
super(DropConstraint, self).__init__(element, **kw)
+ element._create_rule = lambda compiler: False
def _bind_or_error(schemaitem):
bind = schemaitem.bind
class DDLCompiler(engine.Compiled):
+
+ @util.memoized_property
+ def sql_compiler(self):
+ return self.dialect.statement_compiler(self.dialect, self.statement)
+
@property
def preparer(self):
return self.dialect.identifier_preparer
const = ", \n\t".join(p for p in
(self.process(constraint) for constraint in table.constraints
if constraint is not table.primary_key
- and constraint.inline_ddl
+ and (
+ constraint._create_rule is None or
+ constraint._create_rule(self))
and (
not self.dialect.supports_alter or
not getattr(constraint, 'use_alter', False)
def post_create_table(self, table):
return ''
- def _compile(self, tocompile, parameters):
- """compile the given string/parameters using this SchemaGenerator's dialect."""
-
- compiler = self.dialect.statement_compiler(self.dialect, tocompile, parameters)
- compiler.compile()
- return compiler
-
def _validate_identifier(self, ident, truncate):
if truncate:
if len(ident) > self.dialect.max_identifier_length:
if isinstance(column.server_default.arg, basestring):
return "'%s'" % column.server_default.arg
else:
- return unicode(self._compile(column.server_default.arg, None))
+ return self.sql_compiler.process(column.server_default.arg)
else:
return None
if constraint.name is not None:
text += "CONSTRAINT %s " % \
self.preparer.format_constraint(constraint)
- text += " CHECK (%s)" % constraint.sqltext
+ sqltext = sql_util.expression_as_ddl(constraint.sqltext)
+ text += "CHECK (%s)" % self.sql_compiler.process(sqltext)
text += self.define_constraint_deferrability(constraint)
return text
text += self.define_constraint_deferrability(constraint)
return text
- def visit_enum_constraint(self, constraint):
- text = ""
- if constraint.name is not None:
- text += "CONSTRAINT %s " % \
- self.preparer.format_constraint(constraint)
- text += " CHECK (%s IN (%s))" % (
- self.preparer.format_column(constraint.column),
- ",".join("'%s'" % x for x in constraint.type.enums)
- )
- return text
-
def define_constraint_cascades(self, constraint):
text = ""
if constraint.ondelete is not None:
visitors.traverse(clause, {}, {'column':cols.add})
return cols
+def expression_as_ddl(clause):
+ """Given a SQL expression, convert for usage in DDL, such as
+ CREATE INDEX and CHECK CONSTRAINT.
+
+ Converts bind params into quoted literals, column identifiers
+ into detached column constructs so that the parent table
+ identifier is not included.
+
+ """
+ def repl(element):
+ if isinstance(element, expression._BindParamClause):
+ return expression.literal_column(repr(element.value))
+ elif isinstance(element, expression.ColumnClause) and \
+ element.table is not None:
+ return expression.column(element.name)
+ else:
+ return None
+
+ return visitors.replacement_traverse(clause, {}, repl)
+
def adapt_criterion_to_null(crit, nulls):
"""given criterion containing bind params, convert selected elements to IS NULL."""
return dbapi.BINARY
class SchemaType(object):
- """Mark a type as possibly requiring schema-level DDL for usage."""
+ """Mark a type as possibly requiring schema-level DDL for usage.
+
+ Supports types that must be explicitly created/dropped (i.e. PG ENUM type)
+ as well as types that are complimented by table or schema level
+ constraints, triggers, and other rules.
+
+ """
def __init__(self, **kw):
self.name = kw.pop('name', None)
return self.metadata and self.metadata.bind or None
def create(self, bind=None, checkfirst=False):
+ """Issue CREATE ddl for this type, if applicable."""
+
from sqlalchemy.schema import _bind_or_error
if bind is None:
bind = _bind_or_error(self)
t.create(bind=bind, checkfirst=checkfirst)
def drop(self, bind=None, checkfirst=False):
+ """Issue DROP ddl for this type, if applicable."""
+
from sqlalchemy.schema import _bind_or_error
if bind is None:
bind = _bind_or_error(self)
if self.native_enum:
SchemaType._set_table(self, table, column)
- # this constraint DDL object is conditionally
- # compiled by MySQL, Postgresql based on
- # the native_enum flag.
- table.append_constraint(
- EnumConstraint(self, column)
- )
+ def should_create_constraint(compiler):
+ return not self.native_enum or \
+ not compiler.dialect.supports_native_enum
+
+ e = schema.CheckConstraint(
+ column.in_(self.enums),
+ name=self.name,
+ _create_rule=should_create_constraint
+ )
+ table.append_constraint(e)
def adapt(self, impltype):
return impltype(name=self.name,
*self.enums
)
-class EnumConstraint(schema.CheckConstraint):
- __visit_name__ = 'enum_constraint'
-
- def __init__(self, type_, column, **kw):
- super(EnumConstraint, self).__init__('', name=type_.name, **kw)
- self.type = type_
- self.column = column
-
class PickleType(MutableType, TypeDecorator):
"""Holds Python objects.
schema.CreateTable(t1),
"CREATE TABLE sometable ("
"somecolumn VARCHAR(1), "
- " CHECK (somecolumn IN ('x','y','z'))"
+ "CHECK (somecolumn IN ('x', 'y', 'z'))"
")"
)
self.assert_compile(i, "INSERT INTO mytable (name) VALUES (%(name)s) RETURNING mytable.myid, mytable.name", dialect=dialect)
def test_create_partial_index(self):
- tbl = Table('testtbl', MetaData(), Column('data',Integer))
+ m = MetaData()
+ tbl = Table('testtbl', m, Column('data',Integer))
idx = Index('test_idx1', tbl.c.data, postgresql_where=and_(tbl.c.data > 5, tbl.c.data < 10))
self.assert_compile(schema.CreateIndex(idx),
- "CREATE INDEX test_idx1 ON testtbl (data) WHERE testtbl.data > 5 AND testtbl.data < 10", dialect=postgresql.dialect())
-
+ "CREATE INDEX test_idx1 ON testtbl (data) WHERE data > 5 AND data < 10", dialect=postgresql.dialect())
+
@testing.uses_deprecated(r".*'postgres_where' argument has been renamed.*")
def test_old_create_partial_index(self):
tbl = Table('testtbl', MetaData(), Column('data',Integer))
idx = Index('test_idx1', tbl.c.data, postgres_where=and_(tbl.c.data > 5, tbl.c.data < 10))
self.assert_compile(schema.CreateIndex(idx),
- "CREATE INDEX test_idx1 ON testtbl (data) WHERE testtbl.data > 5 AND testtbl.data < 10", dialect=postgresql.dialect())
+ "CREATE INDEX test_idx1 ON testtbl (data) WHERE data > 5 AND data < 10", dialect=postgresql.dialect())
def test_extract(self):
t = table('t', column('col1'))
schema.CreateTable(t1),
"CREATE TABLE sometable ("
"somecolumn VARCHAR(1), "
- " CHECK (somecolumn IN ('x','y','z'))"
+ "CHECK (somecolumn IN ('x', 'y', 'z'))"
")"
)
def test_time_microseconds(self):
dt = datetime.datetime(2008, 6, 27, 12, 0, 0, 125) # 125 usec
eq_(str(dt), '2008-06-27 12:00:00.000125')
- sldt = sqlite._SLDateTime()
+ sldt = sqlite.DATETIME()
bp = sldt.bind_processor(None)
eq_(bp(dt), '2008-06-27 12:00:00.000125')
rp = sldt.result_processor(None, None)
eq_(rp(bp(dt)), dt)
- sldt.__legacy_microseconds__ = True
- bp = sldt.bind_processor(None)
- eq_(bp(dt), '2008-06-27 12:00:00.125')
- eq_(rp(bp(dt)), dt)
-
def test_no_convert_unicode(self):
"""test no utf-8 encoding occurs"""
break
else:
assert False
- assert c.sqltext=="description='hi'"
+ assert str(c.sqltext)=="description='hi'"
for c in table_c.constraints:
if isinstance(c, UniqueConstraint):
constraint = CheckConstraint('a < b',name="my_test_constraint", deferrable=True,initially='DEFERRED', table=t)
self.assert_compile(
schema.AddConstraint(constraint),
- "ALTER TABLE tbl ADD CONSTRAINT my_test_constraint CHECK (a < b) DEFERRABLE INITIALLY DEFERRED"
+ "ALTER TABLE tbl ADD CONSTRAINT my_test_constraint CHECK (a < b) DEFERRABLE INITIALLY DEFERRED"
)
self.assert_compile(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
