Add full support for Postgresql add_exclude_constraint().
This opens up more of the operations API and serves as a
model for other dialect-specific constraints.
Additionally, gracefully degrade if a given constraint
class is not supported with a warning.
Fixes: #412
Change-Id: I0fb89c840518aaeae97929919356f944479bc756
"""
if name is None:
return name
- elif compat.sqla_09 and isinstance(name, sql.elements.quoted_name):
+ elif util.sqla_09 and isinstance(name, sql.elements.quoted_name):
if compat.py2k:
# the attempt to encode to ascii here isn't super ideal,
# however we are trying to cut down on an explosion of
def _render_potential_expr(value, autogen_context, wrap_in_text=True):
if isinstance(value, sql.ClauseElement):
- if compat.sqla_08:
+ if util.sqla_08:
compile_kw = dict(compile_kwargs={
'literal_binds': True, "include_table": False})
else:
def _get_index_rendered_expressions(idx, autogen_context):
- if compat.sqla_08:
+ if util.sqla_08:
return [repr(_ident(getattr(exp, "name", None)))
if isinstance(exp, sa_schema.Column)
else _render_potential_expr(exp, autogen_context)
def _render_constraint(constraint, autogen_context):
- renderer = _constraint_renderers.dispatch(constraint)
- return renderer(constraint, autogen_context)
+ try:
+ renderer = _constraint_renderers.dispatch(constraint)
+ except ValueError:
+ util.warn("No renderer is established for object %r" % constraint)
+ return "[Unknown Python object %r]" % constraint
+ else:
+ return renderer(constraint, autogen_context)
@_constraint_renderers.dispatch_for(sa_schema.PrimaryKeyConstraint)
from sqlalchemy.dialects.postgresql import INTEGER, BIGINT
from ..autogenerate import render
from sqlalchemy import text, Numeric, Column
+from sqlalchemy.types import NULLTYPE
from sqlalchemy import types as sqltypes
-if compat.sqla_08:
+from ..operations.base import Operations
+from ..operations.base import BatchOperations
+from ..operations import ops
+from ..util import sqla_compat
+from ..operations import schemaobj
+from ..autogenerate import render
+
+import logging
+
+if util.sqla_08:
from sqlalchemy.sql.expression import UnaryExpression
else:
from sqlalchemy.sql.expression import _UnaryExpression as UnaryExpression
-import logging
+if util.sqla_100:
+ from sqlalchemy.dialects.postgresql import ExcludeConstraint
+
log = logging.getLogger(__name__)
existing_autoincrement=existing_autoincrement,
**kw)
-
def autogen_column_reflect(self, inspector, table, column_info):
if column_info.get('default') and \
isinstance(column_info['type'], (INTEGER, BIGINT)):
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
- if compat.sqla_08:
+ if util.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
"TYPE %s" % format_type(compiler, element.type_),
"USING %s" % element.using if element.using else ""
)
+
+
+@Operations.register_operation("create_exclude_constraint")
+@BatchOperations.register_operation(
+ "create_exclude_constraint", "batch_create_exclude_constraint")
+@ops.AddConstraintOp.register_add_constraint("exclude_constraint")
+class CreateExcludeConstraintOp(ops.AddConstraintOp):
+ """Represent a create exclude constraint operation."""
+
+ constraint_type = "exclude"
+
+ def __init__(
+ self, constraint_name, table_name,
+ elements, where=None, schema=None,
+ _orig_constraint=None, **kw):
+ self.constraint_name = constraint_name
+ self.table_name = table_name
+ self.elements = elements
+ self.where = where
+ self.schema = schema
+ self._orig_constraint = _orig_constraint
+ self.kw = kw
+
+ @classmethod
+ def from_constraint(cls, constraint):
+ constraint_table = sqla_compat._table_for_constraint(constraint)
+
+ return cls(
+ constraint.name,
+ constraint_table.name,
+ [(expr, op) for expr, name, op in constraint._render_exprs],
+ where=constraint.where,
+ schema=constraint_table.schema,
+ _orig_constraint=constraint,
+ deferrable=constraint.deferrable,
+ initially=constraint.initially,
+ using=constraint.using
+ )
+
+ def to_constraint(self, migration_context=None):
+ if not util.sqla_100:
+ raise NotImplementedError(
+ "ExcludeConstraint not supported until SQLAlchemy 1.0")
+ if self._orig_constraint is not None:
+ return self._orig_constraint
+ schema_obj = schemaobj.SchemaObjects(migration_context)
+ t = schema_obj.table(self.table_name, schema=self.schema)
+ excl = ExcludeConstraint(
+ *self.elements,
+ name=self.constraint_name,
+ where=self.where,
+ **self.kw
+ )
+ for expr, name, oper in excl._render_exprs:
+ t.append_column(Column(name, NULLTYPE))
+ t.append_constraint(excl)
+ return excl
+
+ @classmethod
+ def create_exclude_constraint(
+ cls, operations,
+ constraint_name, table_name, *elements, **kw):
+ """Issue an alter to create an EXCLUDE constraint using the
+ current migration context.
+
+ .. note:: This method is Postgresql specific, and additionally
+ requires at least SQLAlchemy 1.0.
+
+ e.g.::
+
+ from alembic import op
+
+ op.create_exclude_constraint(
+ "user_excl",
+ "user",
+
+ ("period", '&&'),
+ ("group", '='),
+ where=("group != 'some group'")
+
+ )
+
+ Note that the expressions work the same way as that of
+ the ``ExcludeConstraint`` object itself; if plain strings are
+ passed, quoting rules must be applied manually.
+
+ :param name: Name of the constraint.
+ :param table_name: String name of the source table.
+ :param elements: exclude conditions.
+ :param where: SQL expression or SQL string with optional WHERE
+ clause.
+ :param deferrable: optional bool. If set, emit DEFERRABLE or
+ NOT DEFERRABLE when issuing DDL for this constraint.
+ :param initially: optional string. If set, emit INITIALLY <value>
+ when issuing DDL for this constraint.
+ :param schema: Optional schema name to operate within.
+
+ .. versionadded:: 0.9.0
+
+ """
+ op = cls(constraint_name, table_name, elements, **kw)
+ return operations.invoke(op)
+
+ @classmethod
+ def batch_create_exclude_constraint(
+ cls, operations, constraint_name, *elements, **kw):
+ """Issue a "create exclude constraint" instruction using the
+ current batch migration context.
+
+ .. note:: This method is Postgresql specific, and additionally
+ requires at least SQLAlchemy 1.0.
+
+ .. versionadded:: 0.9.0
+
+ .. seealso::
+
+ :meth:`.Operations.create_exclude_constraint`
+
+ """
+ kw['schema'] = operations.impl.schema
+ op = cls(constraint_name, operations.impl.table_name, elements, **kw)
+ return operations.invoke(op)
+
+
+@render.renderers.dispatch_for(CreateExcludeConstraintOp)
+def _add_exclude_constraint(autogen_context, op):
+ return _exclude_constraint(
+ op.to_constraint(),
+ autogen_context,
+ alter=True
+ )
+
+if util.sqla_100:
+ @render._constraint_renderers.dispatch_for(ExcludeConstraint)
+ def _render_inline_exclude_constraint(constraint, autogen_context):
+ rendered = render._user_defined_render(
+ "exclude", constraint, autogen_context)
+ if rendered is not False:
+ return rendered
+
+ return _exclude_constraint(constraint, autogen_context, False)
+
+
+def _postgresql_autogenerate_prefix(autogen_context):
+
+ imports = autogen_context.imports
+ if imports is not None:
+ imports.add("from sqlalchemy.dialects import postgresql")
+ return "postgresql."
+
+
+def _exclude_constraint(constraint, autogen_context, alter):
+ opts = []
+
+ has_batch = autogen_context._has_batch
+
+ if constraint.deferrable:
+ opts.append(("deferrable", str(constraint.deferrable)))
+ if constraint.initially:
+ opts.append(("initially", str(constraint.initially)))
+ if constraint.using:
+ opts.append(("using", str(constraint.using)))
+ if not has_batch and alter and constraint.table.schema:
+ opts.append(("schema", render._ident(constraint.table.schema)))
+ if not alter and constraint.name:
+ opts.append(
+ ("name",
+ render._render_gen_name(autogen_context, constraint.name)))
+
+ if alter:
+ args = [
+ repr(render._render_gen_name(
+ autogen_context, constraint.name))]
+ if not has_batch:
+ args += [repr(render._ident(constraint.table.name))]
+ args.extend([
+ "(%s, %r)" % (
+ render._render_potential_expr(
+ sqltext, autogen_context, wrap_in_text=False),
+ opstring
+ )
+ for sqltext, name, opstring in constraint._render_exprs
+ ])
+ if constraint.where is not None:
+ args.append(
+ "where=%s" % render._render_potential_expr(
+ constraint.where, autogen_context)
+ )
+ args.extend(["%s=%r" % (k, v) for k, v in opts])
+ return "%(prefix)screate_exclude_constraint(%(args)s)" % {
+ 'prefix': render._alembic_autogenerate_prefix(autogen_context),
+ 'args': ", ".join(args)
+ }
+ else:
+ args = [
+ "(%s, %r)" % (
+ render._render_potential_expr(
+ sqltext, autogen_context, wrap_in_text=False),
+ opstring
+ ) for sqltext, name, opstring in constraint._render_exprs
+ ]
+ if constraint.where is not None:
+ args.append(
+ "where=%s" % render._render_potential_expr(
+ constraint.where, autogen_context)
+ )
+ args.extend(["%s=%r" % (k, v) for k, v in opts])
+ return "%(prefix)sExcludeConstraint(%(args)s)" % {
+ "prefix": _postgresql_autogenerate_prefix(autogen_context),
+ "args": ", ".join(args)
+ }
class AddConstraintOp(MigrateOperation):
"""Represent an add constraint operation."""
+ add_constraint_ops = util.Dispatcher()
+
@property
def constraint_type(self):
raise NotImplementedError()
+ @classmethod
+ def register_add_constraint(cls, type_):
+ def go(klass):
+ cls.add_constraint_ops.dispatch_for(type_)(klass.from_constraint)
+ return klass
+ return go
+
@classmethod
def from_constraint(cls, constraint):
- funcs = {
- "unique_constraint": CreateUniqueConstraintOp.from_constraint,
- "foreign_key_constraint": CreateForeignKeyOp.from_constraint,
- "primary_key_constraint": CreatePrimaryKeyOp.from_constraint,
- "check_constraint": CreateCheckConstraintOp.from_constraint,
- "column_check_constraint": CreateCheckConstraintOp.from_constraint,
- }
- return funcs[constraint.__visit_name__](constraint)
+ return cls.add_constraint_ops.dispatch(
+ constraint.__visit_name__)(constraint)
def reverse(self):
return DropConstraintOp.from_constraint(self.to_constraint())
@Operations.register_operation("create_primary_key")
@BatchOperations.register_operation(
"create_primary_key", "batch_create_primary_key")
+@AddConstraintOp.register_add_constraint("primary_key_constraint")
class CreatePrimaryKeyOp(AddConstraintOp):
"""Represent a create primary key operation."""
@Operations.register_operation("create_unique_constraint")
@BatchOperations.register_operation(
"create_unique_constraint", "batch_create_unique_constraint")
+@AddConstraintOp.register_add_constraint("unique_constraint")
class CreateUniqueConstraintOp(AddConstraintOp):
"""Represent a create unique constraint operation."""
@Operations.register_operation("create_foreign_key")
@BatchOperations.register_operation(
"create_foreign_key", "batch_create_foreign_key")
+@AddConstraintOp.register_add_constraint("foreign_key_constraint")
class CreateForeignKeyOp(AddConstraintOp):
"""Represent a create foreign key constraint operation."""
@Operations.register_operation("create_check_constraint")
@BatchOperations.register_operation(
"create_check_constraint", "batch_create_check_constraint")
+@AddConstraintOp.register_add_constraint("check_constraint")
+@AddConstraintOp.register_add_constraint("column_check_constraint")
class CreateCheckConstraintOp(AddConstraintOp):
"""Represent a create check constraint operation."""
"SQLAlchemy 0.9.0 or greater required"
)
+ @property
+ def fail_before_sqla_100(self):
+ return exclusions.fails_if(
+ lambda config: not util.sqla_100,
+ "SQLAlchemy 1.0.0 or greater required"
+ )
+
@property
def fail_before_sqla_099(self):
return exclusions.fails_if(
import io
import sys
-from sqlalchemy import __version__ as sa_version
if sys.version_info < (2, 6):
raise NotImplementedError("Python 2.6 or greater is required.")
-sqla_08 = sa_version >= '0.8.0'
-sqla_09 = sa_version >= '0.9.0'
-
py27 = sys.version_info >= (2, 7)
py2k = sys.version_info < (3, 0)
py3k = sys.version_info >= (3, 0)
Autogenerate can't currently, but **will eventually detect**:
-* Some free-standing constraint additions and removals,
- like CHECK, PRIMARY KEY - these are not fully implemented.
+* Some free-standing constraint additions and removals may not be supported,
+ including PRIMARY KEY, EXCLUDE, CHECK; these are not necessarily implemented
+ within the autogenerate detection system and also may not be supported by
+ the supporting SQLAlchemy dialect.
* Sequence additions, removals - not yet implemented.
function can be used among other things to place a complete
:class:`.MigrationScript` structure in place.
+ .. change:: 412
+ :tags: feature, postgresql
+ :tickets: 412
+
+ Added support for Postgresql EXCLUDE constraints, including the
+ operation directive :meth:`.Operations.create_exclude_constraints`
+ as well as autogenerate render support for the ``ExcludeConstraint``
+ object as present in a ``Table``. Autogenerate detection for an EXCLUDE
+ constraint added or removed to/from an existing table is **not**
+ implemented as the SQLAlchemy Postgresql dialect does not yet support
+ reflection of EXCLUDE constraints.
+
+ Additionally, unknown constraint types now warn when
+ encountered within an autogenerate action rather than raise.
+
.. change:: fk_schema_compare
:tags: bug, operations
import re
import sys
from alembic.testing import TestBase, exclusions, assert_raises
+from alembic.testing import assertions
from alembic.operations import ops
from sqlalchemy import MetaData, Column, Table, String, \
"schema=%r)" % compat.ue('\u0411\u0435\u0437')
)
+ @config.requirements.sqlalchemy_09
+ def test_render_table_w_unsupported_constraint(self):
+ from sqlalchemy.sql.schema import ColumnCollectionConstraint
+
+ class SomeCustomConstraint(ColumnCollectionConstraint):
+ __visit_name__ = 'some_custom'
+
+ m = MetaData()
+
+ t = Table(
+ 't', m, Column('id', Integer),
+ SomeCustomConstraint('id'),
+ )
+ op_obj = ops.CreateTableOp.from_table(t)
+ with assertions.expect_warnings(
+ "No renderer is established for object SomeCustomConstraint"):
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.create_table('t',"
+ "sa.Column('id', sa.Integer(), nullable=True),"
+ "[Unknown Python object "
+ "SomeCustomConstraint(Column('id', Integer(), table=<t>))])"
+ )
+
@patch("alembic.autogenerate.render.MAX_PYTHON_ARGS", 3)
def test_render_table_max_cols(self):
m = MetaData()
'ALTER TABLE some_table ADD COLUMN q SERIAL NOT NULL'
)
+ @config.requirements.fail_before_sqla_100
+ def test_create_exclude_constraint(self):
+ context = op_fixture("postgresql")
+ op.create_exclude_constraint(
+ "ex1", "t1", ('x', '>'), where='x > 5', using="gist")
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT ex1 EXCLUDE USING gist (x WITH >) "
+ "WHERE (x > 5)"
+ )
+
+ @config.requirements.fail_before_sqla_100
+ def test_create_exclude_constraint_quoted_literal(self):
+ context = op_fixture("postgresql")
+ op.create_exclude_constraint(
+ "ex1", "SomeTable", ('"SomeColumn"', '>'),
+ where='"SomeColumn" > 5', using="gist")
+ context.assert_(
+ 'ALTER TABLE "SomeTable" ADD CONSTRAINT ex1 EXCLUDE USING gist '
+ '("SomeColumn" WITH >) WHERE ("SomeColumn" > 5)'
+ )
+
+ @config.requirements.fail_before_sqla_100
+ def test_create_exclude_constraint_quoted_column(self):
+ context = op_fixture("postgresql")
+ op.create_exclude_constraint(
+ "ex1", "SomeTable", (column("SomeColumn"), '>'),
+ where=column("SomeColumn") > 5, using="gist")
+ context.assert_(
+ 'ALTER TABLE "SomeTable" ADD CONSTRAINT ex1 EXCLUDE '
+ 'USING gist ("SomeColumn" WITH >) WHERE ("SomeColumn" > 5)'
+ )
+
+
class PGOfflineEnumTest(TestBase):
def setUp(self):
op_obj = ops.CreateIndexOp.from_index(idx)
- if compat.sqla_08:
+ if util.sqla_08:
eq_ignore_whitespace(
autogenerate.render_op_text(autogen_context, op_obj),
"""op.create_index('foo_idx', 't', \
ARRAY(String), self.autogen_context),
"postgresql.ARRAY(foobar.MYVARCHAR)"
)
+
+ @config.requirements.fail_before_sqla_100
+ def test_add_exclude_constraint(self):
+ from sqlalchemy.dialects.postgresql import ExcludeConstraint
+
+ autogen_context = self.autogen_context
+
+ m = MetaData()
+ t = Table('t', m,
+ Column('x', String),
+ Column('y', String)
+ )
+
+ op_obj = ops.AddConstraintOp.from_constraint(ExcludeConstraint(
+ (t.c.x, ">"),
+ where=t.c.x != 2,
+ using="gist",
+ name="t_excl_x"
+ ))
+
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(autogen_context, op_obj),
+ "op.create_exclude_constraint('t_excl_x', 't', ('x', '>'), "
+ "where=sa.text(!U'x != 2'), using='gist')"
+ )
+
+ @config.requirements.fail_before_sqla_100
+ def test_inline_exclude_constraint(self):
+ from sqlalchemy.dialects.postgresql import ExcludeConstraint
+
+ autogen_context = self.autogen_context
+
+ m = MetaData()
+ t = Table(
+ 't', m,
+ Column('x', String),
+ Column('y', String),
+ ExcludeConstraint(
+ ('x', ">"),
+ using="gist",
+ where='x != 2',
+ name="t_excl_x"
+ )
+ )
+
+ op_obj = ops.CreateTableOp.from_table(t)
+
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(autogen_context, op_obj),
+ "op.create_table('t',sa.Column('x', sa.String(), nullable=True),"
+ "sa.Column('y', sa.String(), nullable=True),"
+ "postgresql.ExcludeConstraint((!U'x', '>'), "
+ "where=sa.text(!U'x != 2'), using='gist', name='t_excl_x')"
+ ")"
+ )
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
