def _create(self, op_impl):
self._transfer_elements_to_new_table()
+
+ op_impl.prep_table_for_batch(self.table)
op_impl.create_table(self.new_table)
op_impl._exec(
"""
return False
+ def prep_table_for_batch(self, table):
+ """perform any operations needed on a table before a new
+ one is created to replace it in batch mode.
+
+ the PG dialect uses this to drop constraints on the table
+ before the new one uses those same names.
+
+ """
+
+
@property
def bind(self):
return self.connection
__dialect__ = 'postgresql'
transactional_ddl = True
+ def prep_table_for_batch(self, table):
+ for constraint in table.constraints:
+ self.drop_constraint(constraint)
+
def compare_server_default(self, inspector_column,
metadata_column,
rendered_metadata_default,
Column('x', Integer)
)
t1.create(self.conn)
+
self.conn.execute(
t1.insert(),
[
{"id": 5, "x": 9}
])
+ def test_drop_column_fk_recreate(self):
+ with self.op.batch_alter_table("foo", recreate='always') as batch_op:
+ batch_op.drop_column('data')
+
+ self._assert_data([
+ {"id": 1, "x": 5},
+ {"id": 2, "x": 6},
+ {"id": 3, "x": 7},
+ {"id": 4, "x": 8},
+ {"id": 5, "x": 9}
+ ])
+
def test_rename_column(self):
with self.op.batch_alter_table("foo") as batch_op:
batch_op.alter_column('x', new_column_name='y')
def test_change_type(self):
super(BatchRoundTripPostgresqlTest, self).test_change_type()
- @exclusions.fails()
- def test_add_column_recreate(self):
- super(BatchRoundTripPostgresqlTest, self).test_add_column_recreate()