and :meth:`.BatchOperations.create_check_constraint`.
fixes #305
- table keyword arguments are copied from the original reflected table,
such as the "mysql_engine" keyword argument.
def prep_table_for_batch(self, table):
for constraint in table.constraints:
- self.drop_constraint(constraint)
+ if constraint.name is not None:
+ self.drop_constraint(constraint)
def compare_server_default(self, inspector_column,
metadata_column,
self.recreate = recreate
self.copy_from = copy_from
self.table_args = table_args
- self.table_kwargs = table_kwargs
+ self.table_kwargs = dict(table_kwargs)
self.reflect_args = reflect_args
self.reflect_kwargs = reflect_kwargs
self.naming_convention = naming_convention
for idx in self.table.indexes:
self.indexes[idx.name] = idx
+ for k in self.table.kwargs:
+ self.table_kwargs.setdefault(k, self.table.kwargs[k])
+
def _transfer_elements_to_new_table(self):
assert self.new_table is None, "Can only create new table once"
m = MetaData()
schema = self.table.schema
+
self.new_table = new_table = Table(
'_alembic_batch_temp', m,
*(list(self.columns.values()) + list(self.table_args)),
def add_constraint(self, const):
if not const.name:
raise ValueError("Constraint must have a name")
+ if isinstance(const, sql_schema.PrimaryKeyConstraint):
+ if self.table.primary_key in self.unnamed_constraints:
+ self.unnamed_constraints.remove(self.table.primary_key)
+
self.named_constraints[const.name] = const
def drop_constraint(self, const):
:meth:`.Operations.create_primary_key`
"""
- raise NotImplementedError("not yet implemented")
+ op = cls(
+ constraint_name, operations.impl.table_name, columns,
+ schema=operations.impl.schema
+ )
+ return operations.invoke(op)
@Operations.register_operation("create_unique_constraint")
:meth:`.Operations.create_check_constraint`
"""
- raise NotImplementedError("not yet implemented")
+ op = cls(
+ constraint_name, operations.impl.table_name,
+ condition, schema=operations.impl.schema, **kw)
+ return operations.invoke(op)
@Operations.register_operation("create_index")
def primary_key_constraint(self, name, table_name, cols, schema=None):
m = self.metadata()
columns = [sa_schema.Column(n, NULLTYPE) for n in cols]
- t1 = sa_schema.Table(table_name, m,
- *columns,
- schema=schema)
- p = sa_schema.PrimaryKeyConstraint(*columns, name=name)
- t1.append_constraint(p)
+ t = sa_schema.Table(
+ table_name, m,
+ *columns,
+ schema=schema)
+ p = sa_schema.PrimaryKeyConstraint(
+ *[t.c[n] for n in cols], name=name)
+ t.append_constraint(p)
return p
def foreign_key_constraint(
"MATCH for foreign keys added in SQLAlchemy 0.8.0"
)
+ @property
+ def check_constraints_w_enforcement(self):
+ """Target database must support check constraints
+ and also enforce them."""
+
+ return exclusions.open()
+
+ @property
+ def reflects_pk_names(self):
+ return exclusions.closed()
+
@property
def fail_before_sqla_079(self):
return exclusions.fails_if(
.. changelog::
:version: 0.7.7
+ .. change::
+ :tags: feature, batch
+ :tickets: 305
+
+ Implemented support for :meth:`.BatchOperations.create_primary_key`
+ and :meth:`.BatchOperations.create_check_constraint`. Additionally,
+ table keyword arguments are copied from the original reflected table,
+ such as the "mysql_engine" keyword argument.
+
.. change::
:tags: bug, environment
:tickets: 300
and not util.sqla_100
)
+ @property
+ def check_constraints_w_enforcement(self):
+ return exclusions.fails_on("mysql")
+
@property
def unnamed_constraints(self):
"""constraints without names are supported."""
@property
def reflects_unique_constraints_unambiguously(self):
return exclusions.fails_on("mysql")
+
+ @property
+ def reflects_pk_names(self):
+ """Target driver reflects the name of primary key constraints."""
+
+ return exclusions.fails_on_everything_except(
+ 'postgresql', 'oracle', 'mssql', 'sybase')
import re
from alembic.testing import exclusions
+from alembic.testing import assert_raises_message
from alembic.testing import TestBase, eq_, config
from alembic.testing.fixtures import op_fixture
from alembic.testing import mock
UniqueConstraint, ForeignKeyConstraint, Index, Boolean, CheckConstraint, \
Enum
from sqlalchemy.engine.reflection import Inspector
-from sqlalchemy.sql import column
+from sqlalchemy.sql import column, text
from sqlalchemy.schema import CreateTable, CreateIndex
+from sqlalchemy import exc
class BatchApplyTest(TestBase):
self.mock_schema.UniqueConstraint())]
)
+ def test_create_pk(self):
+ with self._fixture() as batch:
+ batch.create_primary_key('pk1', ['a', 'b'])
+
+ eq_(
+ self.mock_schema.Table().c.__getitem__.mock_calls,
+ [mock.call('a'), mock.call('b')]
+ )
+
+ eq_(
+ self.mock_schema.PrimaryKeyConstraint.mock_calls,
+ [
+ mock.call(
+ self.mock_schema.Table().c.__getitem__(),
+ self.mock_schema.Table().c.__getitem__(),
+ name='pk1'
+ )
+ ]
+ )
+ eq_(
+ batch.impl.operations.impl.mock_calls,
+ [mock.call.add_constraint(
+ self.mock_schema.PrimaryKeyConstraint())]
+ )
+
+ def test_create_check(self):
+ expr = text("a > b")
+ with self._fixture() as batch:
+ batch.create_check_constraint('ck1', expr)
+
+ eq_(
+ self.mock_schema.CheckConstraint.mock_calls,
+ [
+ mock.call(
+ expr, name="ck1"
+ )
+ ]
+ )
+ eq_(
+ batch.impl.operations.impl.mock_calls,
+ [mock.call.add_constraint(
+ self.mock_schema.CheckConstraint())]
+ )
+
def test_drop_constraint(self):
with self._fixture() as batch:
batch.drop_constraint('uq1')
context = MigrationContext.configure(self.conn)
self.op = Operations(context)
+ def _no_pk_fixture(self):
+ nopk = Table(
+ 'nopk', self.metadata,
+ Column('a', Integer),
+ Column('b', Integer),
+ Column('c', Integer),
+ mysql_engine='InnoDB'
+ )
+ nopk.create(self.conn)
+ self.conn.execute(
+ nopk.insert(),
+ [
+ {"a": 1, "b": 2, "c": 3},
+ {"a": 2, "b": 4, "c": 5},
+ ]
+
+ )
+ return nopk
+
def tearDown(self):
self.metadata.drop_all(self.conn)
self.conn.close()
{"id": 5, "x": 9}
])
+ def test_add_pk_constraint(self):
+ self._no_pk_fixture()
+ with self.op.batch_alter_table("nopk", recreate="always") as batch_op:
+ batch_op.create_primary_key('newpk', ['a', 'b'])
+
+ pk_const = Inspector.from_engine(self.conn).get_pk_constraint('nopk')
+ with config.requirements.reflects_pk_names.fail_if():
+ eq_(pk_const['name'], 'newpk')
+ eq_(pk_const['constrained_columns'], ['a', 'b'])
+
+ @config.requirements.check_constraints_w_enforcement
+ def test_add_ck_constraint(self):
+ with self.op.batch_alter_table("foo", recreate="always") as batch_op:
+ batch_op.create_check_constraint("newck", text("x > 0"))
+
+ # we dont support reflection of CHECK constraints
+ # so test this by just running invalid data in
+ foo = self.metadata.tables['foo']
+
+ assert_raises_message(
+ exc.IntegrityError,
+ "newck",
+ self.conn.execute,
+ foo.insert(), {"id": 6, "data": 5, "x": -2}
+ )
+
@config.requirements.sqlalchemy_094
@config.requirements.unnamed_constraints
def test_drop_foreign_key(self):
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
