From: Mike Bayer Date: Thu, 20 Nov 2025 15:45:37 +0000 (-0500) Subject: implement ColumnComment compiles for SQL Server X-Git-Tag: rel_1_18_0~15 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=4c42276f5a1c5f3c7d32e34fa6e6388f22d9f092;p=thirdparty%2Fsqlalchemy%2Falembic.git implement ColumnComment compiles for SQL Server Implemented DDL for column comment add/update/delete when using the :paramref:`.Operations.alter_column.comment` parameter with :meth:`.Operations.alter_column` on Microsoft SQL Server. Previously, these functions were not implemented for SQL Server and would raise ``UnsupportedCompilationError``. Fixes: #1755 Change-Id: Ia4a0f6cd44c4c4b691383b91ced5a9d898d79e46 --- diff --git a/alembic/ddl/mssql.py b/alembic/ddl/mssql.py index 1d1e3e98..a236e41f 100644 --- a/alembic/ddl/mssql.py +++ b/alembic/ddl/mssql.py @@ -20,6 +20,7 @@ from sqlalchemy.sql.elements import ClauseElement from .base import AddColumn from .base import alter_column from .base import alter_table +from .base import ColumnComment from .base import ColumnDefault from .base import ColumnName from .base import ColumnNullable @@ -433,3 +434,89 @@ def visit_rename_table( format_table_name(compiler, element.table_name, element.schema), format_table_name(compiler, element.new_table_name, None), ) + + +def _add_column_comment( + compiler: MSDDLCompiler, + schema: Optional[str], + tname: str, + cname: str, + comment: str, +) -> str: + schema_name = schema if schema else compiler.dialect.default_schema_name + assert schema_name + return ( + "exec sp_addextendedproperty 'MS_Description', {}, " + "'schema', {}, 'table', {}, 'column', {}".format( + compiler.sql_compiler.render_literal_value( + comment, sqltypes.NVARCHAR() + ), + compiler.preparer.quote_schema(schema_name), + compiler.preparer.quote(tname), + compiler.preparer.quote(cname), + ) + ) + + +def _update_column_comment( + compiler: MSDDLCompiler, + schema: Optional[str], + tname: str, + cname: str, + comment: str, +) -> str: + schema_name = schema if schema else compiler.dialect.default_schema_name + assert schema_name + return ( + "exec sp_updateextendedproperty 'MS_Description', {}, " + "'schema', {}, 'table', {}, 'column', {}".format( + compiler.sql_compiler.render_literal_value( + comment, sqltypes.NVARCHAR() + ), + compiler.preparer.quote_schema(schema_name), + compiler.preparer.quote(tname), + compiler.preparer.quote(cname), + ) + ) + + +def _drop_column_comment( + compiler: MSDDLCompiler, schema: Optional[str], tname: str, cname: str +) -> str: + schema_name = schema if schema else compiler.dialect.default_schema_name + assert schema_name + return ( + "exec sp_dropextendedproperty 'MS_Description', " + "'schema', {}, 'table', {}, 'column', {}".format( + compiler.preparer.quote_schema(schema_name), + compiler.preparer.quote(tname), + compiler.preparer.quote(cname), + ) + ) + + +@compiles(ColumnComment, "mssql") +def visit_column_comment( + element: ColumnComment, compiler: MSDDLCompiler, **kw: Any +) -> str: + if element.comment is not None: + if element.existing_comment is not None: + return _update_column_comment( + compiler, + element.schema, + element.table_name, + element.column_name, + element.comment, + ) + else: + return _add_column_comment( + compiler, + element.schema, + element.table_name, + element.column_name, + element.comment, + ) + else: + return _drop_column_comment( + compiler, element.schema, element.table_name, element.column_name + ) diff --git a/alembic/testing/assertions.py b/alembic/testing/assertions.py index 898fbd16..e76103d5 100644 --- a/alembic/testing/assertions.py +++ b/alembic/testing/assertions.py @@ -132,6 +132,7 @@ def _get_dialect(name): d.implicit_returning = True elif name == "mssql": d.legacy_schema_aliasing = False + d.default_schema_name = "dbo" return d diff --git a/docs/build/unreleased/1755.rst b/docs/build/unreleased/1755.rst new file mode 100644 index 00000000..c6048934 --- /dev/null +++ b/docs/build/unreleased/1755.rst @@ -0,0 +1,9 @@ +.. change:: + :tags: bug, mssql + :tickets: 1755 + + Implemented DDL for column comment add/update/delete when using the + :paramref:`.Operations.alter_column.comment` parameter with + :meth:`.Operations.alter_column` on Microsoft SQL Server. Previously, + these functions were not implemented for SQL Server and would raise + ``UnsupportedCompilationError``. \ No newline at end of file diff --git a/tests/test_mssql.py b/tests/test_mssql.py index 1390f2df..0d80d59e 100644 --- a/tests/test_mssql.py +++ b/tests/test_mssql.py @@ -533,6 +533,71 @@ class OpTest(TestBase): existing_server_default=esd(), ) + def test_alter_column_add_comment(self): + context = op_fixture("mssql") + op.alter_column( + "t1", + "c1", + existing_type=String(50), + comment="c1 comment", + ) + context.assert_contains( + "exec sp_addextendedproperty 'MS_Description', " + "N'c1 comment', 'schema', dbo, 'table', t1, 'column', c1" + ) + + def test_alter_column_update_comment(self): + context = op_fixture("mssql") + op.alter_column( + "t1", + "c1", + existing_type=String(50), + comment="updated comment", + existing_comment="old comment", + ) + context.assert_contains( + "exec sp_updateextendedproperty 'MS_Description', " + "N'updated comment', 'schema', dbo, 'table', t1, 'column', c1" + ) + + def test_alter_column_add_comment_schema(self): + context = op_fixture("mssql") + op.alter_column( + "t1", + "c1", + existing_type=String(50), + comment="c1 comment", + schema="xyz", + ) + context.assert_contains("'schema', xyz, 'table', t1, 'column', c1") + + def test_alter_column_add_comment_quoting(self): + context = op_fixture("mssql") + op.alter_column( + "user", + "theme", + existing_type=String(20), + comment="Column comment with 'quotes'", + existing_nullable=True, + schema="dbo", + ) + context.assert_contains("N'Column comment with ''quotes'''") + context.assert_contains("'table', [user], 'column', theme") + + def test_alter_column_drop_comment(self): + context = op_fixture("mssql") + op.alter_column( + "t1", + "c1", + existing_type=String(50), + comment=None, + existing_comment="existing comment", + ) + context.assert_contains( + "exec sp_dropextendedproperty 'MS_Description', " + "'schema', dbo, 'table', t1, 'column', c1" + ) + class RoundTripTest(TestBase): __backend__ = True @@ -605,3 +670,61 @@ class RoundTripTest(TestBase): ops_context.drop_column( "access", "created_at", mssql_drop_default=True ) + + @testing.variation("op", ["add", "update", "drop"]) + def test_column_comment( + self, ops_context, connection, metadata, op: testing.Variation + ): + """test #1755""" + t = Table( + "t", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(50)), + ) + t.create(connection) + + if op.update or op.drop: + ops_context.alter_column( + "t", + "data", + existing_type=String(50), + comment="initial comment", + ) + + if op.add: + ops_context.alter_column( + "t", "data", existing_type=String(50), comment="data comment" + ) + expected = "data comment" + elif op.update: + ops_context.alter_column( + "t", + "data", + existing_type=String(50), + comment="updated comment", + existing_comment="initial comment", + ) + expected = "updated comment" + elif op.drop: + ops_context.alter_column( + "t", + "data", + existing_type=String(50), + comment=None, + existing_comment="initial comment", + ) + expected = None + else: + op.fail() + + # Verify expected result + result = connection.execute( + text( + "SELECT value FROM sys.extended_properties " + "WHERE major_id = OBJECT_ID('t') " + "AND minor_id = COLUMNPROPERTY(major_id, 'data', 'ColumnId') " + "AND name = 'MS_Description'" + ) + ).scalar() + eq_(result, expected)