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

Author: zzzeek

alembic/ddl/mssql.py

@@ -20,6 +20,7 @@
 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
+        )

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
 
 

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``.

tests/test_mssql.py

@@ -533,6 +533,71 @@ def test_alter_column_identity_add_not_supported(self, sd, esd):
             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 @@ def test_issue_1744(self, ops_context, connection, metadata, op):
             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)