diff options
author | CaselIT <cfederico87@gmail.com> | 2023-02-23 21:45:11 +0100 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2023-02-28 13:24:23 -0500 |
commit | 7ebd32d194c86170130f6925a18d944431cbce00 (patch) | |
tree | 4aefc0241b38c7634b9859645df6845691b2c5c5 | |
parent | 919297555f17efedcdcc7cc936204a1b86b804d3 (diff) | |
download | alembic-7ebd32d194c86170130f6925a18d944431cbce00.tar.gz |
Escape sql server constriant names
Properly escape constraint name on SQL Server when dropping
a column while specifying ``mssql_drop_default=True`` or
``mssql_drop_check=True`` or ``mssql_drop_foreign_key=True``.
Fixes: #1187
Change-Id: I060442bc63c4e53f64724985e20e6e15e4335b6b
-rw-r--r-- | alembic/ddl/mssql.py | 5 | ||||
-rw-r--r-- | docs/build/unreleased/1187.rst | 8 | ||||
-rw-r--r-- | tests/test_mssql.py | 73 |
3 files changed, 75 insertions, 11 deletions
diff --git a/alembic/ddl/mssql.py b/alembic/ddl/mssql.py index bdf215d..ebf4db1 100644 --- a/alembic/ddl/mssql.py +++ b/alembic/ddl/mssql.py @@ -306,9 +306,8 @@ def _exec_drop_col_constraint( ) # from http://www.mssqltips.com/sqlservertip/1425/\ # working-with-default-constraints-in-sql-server/ - # TODO: needs table formatting, etc. return """declare @const_name varchar(256) -select @const_name = [name] from %(type)s +select @const_name = QUOTENAME([name]) from %(type)s where parent_object_id = object_id('%(schema_dot)s%(tname)s') and col_name(parent_object_id, parent_column_id) = '%(colname)s' exec('alter table %(tname_quoted)s drop constraint ' + @const_name)""" % { @@ -327,7 +326,7 @@ def _exec_drop_col_fk_constraint( schema, tname, colname = element.schema, element.tname, element.colname return """declare @const_name varchar(256) -select @const_name = [name] from +select @const_name = QUOTENAME([name]) from sys.foreign_keys fk join sys.foreign_key_columns fkc on fk.object_id=fkc.constraint_object_id where fkc.parent_object_id = object_id('%(schema_dot)s%(tname)s') diff --git a/docs/build/unreleased/1187.rst b/docs/build/unreleased/1187.rst new file mode 100644 index 0000000..0d9c7dd --- /dev/null +++ b/docs/build/unreleased/1187.rst @@ -0,0 +1,8 @@ +.. change:: + :tags: bug, mssql + :tickets: 1187 + + Properly escape constraint name on SQL Server when dropping + a column while specifying ``mssql_drop_default=True`` or + ``mssql_drop_check=True`` or ``mssql_drop_foreign_key=True``. + diff --git a/tests/test_mssql.py b/tests/test_mssql.py index 4bb618c..640e168 100644 --- a/tests/test_mssql.py +++ b/tests/test_mssql.py @@ -4,10 +4,16 @@ from __future__ import annotations from typing import Any from typing import Dict +from sqlalchemy import CheckConstraint from sqlalchemy import Column from sqlalchemy import exc +from sqlalchemy import ForeignKey +from sqlalchemy import inspect from sqlalchemy import Integer +from sqlalchemy import MetaData from sqlalchemy import String +from sqlalchemy import Table +from sqlalchemy import text from alembic import command from alembic import op @@ -17,6 +23,7 @@ from alembic.testing import combinations from alembic.testing import config from alembic.testing import eq_ from alembic.testing import expect_warnings +from alembic.testing import fixture from alembic.testing.env import _no_sql_testing_config from alembic.testing.env import clear_staging_env from alembic.testing.env import staging_env @@ -181,9 +188,10 @@ class OpTest(TestBase): server_default=None, ) context.assert_contains( - "declare @const_name varchar(256)select @const_name = [name] " - "from sys.default_constraintswhere parent_object_id = " - "object_id('t')and col_name(parent_object_id, " + "declare @const_name varchar(256)\n" + "select @const_name = QUOTENAME([name]) " + "from sys.default_constraints\nwhere parent_object_id = " + "object_id('t')\nand col_name(parent_object_id, " "parent_column_id) = 'c'" ) context.assert_contains( @@ -199,9 +207,10 @@ class OpTest(TestBase): schema="xyz", ) context.assert_contains( - "declare @const_name varchar(256)select @const_name = [name] " - "from sys.default_constraintswhere parent_object_id = " - "object_id('xyz.t')and col_name(parent_object_id, " + "declare @const_name varchar(256)\n" + "select @const_name = QUOTENAME([name]) from " + "sys.default_constraints\nwhere parent_object_id = " + "object_id('xyz.t')\nand col_name(parent_object_id, " "parent_column_id) = 'c'" ) context.assert_contains( @@ -259,7 +268,7 @@ class OpTest(TestBase): op.drop_column("t1", "c1", mssql_drop_foreign_key=True) context.assert_contains( "declare @const_name varchar(256)\n" - "select @const_name = [name] from\n" + "select @const_name = QUOTENAME([name]) from\n" "sys.foreign_keys fk join sys.foreign_key_columns fkcon " "fk.object_id=fkc.constraint_object_id\n" "where fkc.parent_object_id = object_id('t1')\nand " @@ -276,7 +285,7 @@ class OpTest(TestBase): op.drop_column("t1", "c1", schema="xyz", mssql_drop_foreign_key=True) context.assert_contains( "declare @const_name varchar(256)\n" - "select @const_name = [name] from\n" + "select @const_name = QUOTENAME([name]) from\n" "sys.foreign_keys fk join sys.foreign_key_columns fkcon " "fk.object_id=fkc.constraint_object_id\n" "where fkc.parent_object_id = object_id('xyz.t1')\nand " @@ -516,3 +525,51 @@ class OpTest(TestBase): server_default=sd(), existing_server_default=esd(), ) + + +class RoundTripTest(TestBase): + __backend__ = True + __only_on__ = "mssql" + + @fixture + def tables(self, connection): + self.meta = MetaData() + self.tbl_other = Table( + "other", self.meta, Column("oid", Integer, primary_key=True) + ) + self.tbl = Table( + "round_trip_table", + self.meta, + Column("id", Integer, primary_key=True), + Column( + "oid_fk", ForeignKey("other.oid", name="require quote charß!") + ), + Column("with_check", Integer), + CheckConstraint("with_check > 0", name="chéck çst"), + ) + self.meta.create_all(connection) + yield + self.meta.drop_all(connection) + + def test_drop_col_with_fk(self, ops_context, connection, tables): + ops_context.drop_column( + "round_trip_table", "oid_fk", mssql_drop_foreign_key=True + ) + insp = inspect(connection) + eq_(insp.get_foreign_keys("round_trip_table"), []) + + def test_drop_col_with_check(self, ops_context, connection, tables): + ops_context.drop_column( + "round_trip_table", "with_check", mssql_drop_check=True + ) + # No get_check_constraints on mssql + val = connection.execute( + text( + "select name from sys.check_constraints where " + "parent_object_id=object_id('round_trip_table')" + ) + ).scalar() + eq_(val, None) + + # don't know if a default constraint can be explicitly named, but + # the path is the same as the check constraint, so it should be good |