Comments on (named) constraints

Adds support for comments on named constraints, including `ForeignKeyConstraint`, `PrimaryKeyConstraint`, `CheckConstraint`, `UniqueConstraint`, solving the [Issue 5667](https://github.com/sqlalchemy/sqlalchemy/issues/5667).

Supports only PostgreSQL backend.

### Description

Following the example of [Issue 1546](https://github.com/sqlalchemy/sqlalchemy/issues/1546), supports comments on constraints. Specifically, enables comments on _named_ ones — as I get it, PostgreSQL prohibits comments on unnamed constraints.

Enables setting the comments for named constraints like this:
```
Table(
   'example', metadata,
   Column('id', Integer),
   Column('data', sa.String(30)),
   PrimaryKeyConstraint(
       "id", name="id_pk", comment="id_pk comment"
    ),
   CheckConstraint('id < 100', name="cc1", comment="Id value can't exceed 100"),
   UniqueConstraint(['data'], name="uc1", comment="Must have unique data field"),
)
```

Provides the DDL representation for constraint comments and routines to create and drop them. Class `.Inspector` reflects constraint comments via methods like `get_check_constraints` .
### Checklist
<!-- go over following points. check them with an `x` if they do apply, (they turn into clickable checkboxes once the PR is submitted, so no need to do everything at once)

-->

This pull request is:

- [ ] A documentation / typographical error fix
- [ ] A short code fix
- [x] A new feature implementation
	- Solves the issue 5667.
	- The commit message includes `Fixes: 5667`.
	- Includes tests based on comment reflection.

**Have a nice day!**

Fixes: #5667
Closes: #7742
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/7742
Pull-request-sha: 42a5d3c3e9ccf9a9d5397fd007aeab0854f66130

Change-Id: Ia60f578595afdbd6089541c9a00e37997ef78ad3

12 files changed, 260 insertions, 77 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 502371be9..96a0f8f21 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -3013,6 +3013,7 @@ class MySQLDialect(default.DefaultDialect):
             {"name": spec["name"], "sqltext": spec["sqltext"]}
             for spec in parsed_state.ck_constraints
         ]
+        cks.sort(key=lambda d: d["name"] or "~")  # sort None as last
         return cks if cks else ReflectionDefaults.check_constraints()
 
     @reflection.cache
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 8fc24c933..394c64360 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -2428,6 +2428,36 @@ class PGDDLCompiler(compiler.DDLCompiler):
             create, prefix=prefix, **kw
         )
 
+    def _can_comment_on_constraint(self, ddl_instance):
+        constraint = ddl_instance.element
+        if constraint.name is None:
+            raise exc.CompileError(
+                f"Can't emit COMMENT ON for constraint {constraint!r}: "
+                "it has no name"
+            )
+        if constraint.table is None:
+            raise exc.CompileError(
+                f"Can't emit COMMENT ON for constraint {constraint!r}: "
+                "it has no associated table"
+            )
+
+    def visit_set_constraint_comment(self, create, **kw):
+        self._can_comment_on_constraint(create)
+        return "COMMENT ON CONSTRAINT %s ON %s IS %s" % (
+            self.preparer.format_constraint(create.element),
+            self.preparer.format_table(create.element.table),
+            self.sql_compiler.render_literal_value(
+                create.element.comment, sqltypes.String()
+            ),
+        )
+
+    def visit_drop_constraint_comment(self, drop, **kw):
+        self._can_comment_on_constraint(drop)
+        return "COMMENT ON CONSTRAINT %s ON %s IS NULL" % (
+            self.preparer.format_constraint(drop.element),
+            self.preparer.format_table(drop.element.table),
+        )
+
 
 class PGTypeCompiler(compiler.GenericTypeCompiler):
     def visit_TSVECTOR(self, type_, **kw):
@@ -2873,6 +2903,7 @@ class PGDialect(default.DefaultDialect):
     postfetch_lastrowid = False
 
     supports_comments = True
+    supports_constraint_comments = True
     supports_default_values = True
 
     supports_default_metavalue = True
@@ -3697,6 +3728,12 @@ class PGDialect(default.DefaultDialect):
                 sql.func.generate_subscripts(
                     pg_catalog.pg_constraint.c.conkey, 1
                 ).label("ord"),
+                pg_catalog.pg_description.c.description,
+            )
+            .outerjoin(
+                pg_catalog.pg_description,
+                pg_catalog.pg_description.c.objoid
+                == pg_catalog.pg_constraint.c.oid,
             )
             .where(
                 pg_catalog.pg_constraint.c.contype == bindparam("contype"),
@@ -3709,6 +3746,7 @@ class PGDialect(default.DefaultDialect):
             select(
                 con_sq.c.conrelid,
                 con_sq.c.conname,
+                con_sq.c.description,
                 pg_catalog.pg_attribute.c.attname,
             )
             .select_from(pg_catalog.pg_attribute)
@@ -3728,6 +3766,7 @@ class PGDialect(default.DefaultDialect):
                 attr_sq.c.conrelid,
                 sql.func.array_agg(attr_sq.c.attname).label("cols"),
                 attr_sq.c.conname,
+                sql.func.min(attr_sq.c.description).label("description"),
             )
             .group_by(attr_sq.c.conrelid, attr_sq.c.conname)
             .order_by(attr_sq.c.conrelid, attr_sq.c.conname)
@@ -3751,16 +3790,16 @@ class PGDialect(default.DefaultDialect):
             )
 
             result_by_oid = defaultdict(list)
-            for oid, cols, constraint_name in result:
-                result_by_oid[oid].append((cols, constraint_name))
+            for oid, cols, constraint_name, comment in result:
+                result_by_oid[oid].append((cols, constraint_name, comment))
 
             for oid, tablename in batch:
                 for_oid = result_by_oid.get(oid, ())
                 if for_oid:
-                    for cols, constraint in for_oid:
-                        yield tablename, cols, constraint
+                    for cols, constraint, comment in for_oid:
+                        yield tablename, cols, constraint, comment
                 else:
-                    yield tablename, None, None
+                    yield tablename, None, None, None
 
     @reflection.cache
     def get_pk_constraint(self, connection, table_name, schema=None, **kw):
@@ -3790,11 +3829,12 @@ class PGDialect(default.DefaultDialect):
                 {
                     "constrained_columns": [] if cols is None else cols,
                     "name": pk_name,
+                    "comment": comment,
                 }
                 if pk_name is not None
                 else default(),
             )
-            for (table_name, cols, pk_name) in result
+            for table_name, cols, pk_name, comment in result
         )
 
     @reflection.cache
@@ -3836,6 +3876,7 @@ class PGDialect(default.DefaultDialect):
                     else_=None,
                 ),
                 pg_namespace_ref.c.nspname,
+                pg_catalog.pg_description.c.description,
             )
             .select_from(pg_catalog.pg_class)
             .outerjoin(
@@ -3854,6 +3895,11 @@ class PGDialect(default.DefaultDialect):
                 pg_namespace_ref,
                 pg_class_ref.c.relnamespace == pg_namespace_ref.c.oid,
             )
+            .outerjoin(
+                pg_catalog.pg_description,
+                pg_catalog.pg_description.c.objoid
+                == pg_catalog.pg_constraint.c.oid,
+            )
             .order_by(
                 pg_catalog.pg_class.c.relname,
                 pg_catalog.pg_constraint.c.conname,
@@ -3901,7 +3947,7 @@ class PGDialect(default.DefaultDialect):
 
         fkeys = defaultdict(list)
         default = ReflectionDefaults.foreign_keys
-        for table_name, conname, condef, conschema in result:
+        for table_name, conname, condef, conschema, comment in result:
             # ensure that each table has an entry, even if it has
             # no foreign keys
             if conname is None:
@@ -3973,6 +4019,7 @@ class PGDialect(default.DefaultDialect):
                 "referred_table": referred_table,
                 "referred_columns": referred_columns,
                 "options": options,
+                "comment": comment,
             }
             table_fks.append(fkey_d)
         return fkeys.items()
@@ -4242,7 +4289,7 @@ class PGDialect(default.DefaultDialect):
         # each table can have multiple unique constraints
         uniques = defaultdict(list)
         default = ReflectionDefaults.unique_constraints
-        for (table_name, cols, con_name) in result:
+        for table_name, cols, con_name, comment in result:
             # ensure a list is created for each table. leave it empty if
             # the table has no unique cosntraint
             if con_name is None:
@@ -4253,6 +4300,7 @@ class PGDialect(default.DefaultDialect):
                 {
                     "column_names": cols,
                     "name": con_name,
+                    "comment": comment,
                 }
             )
         return uniques.items()
@@ -4339,6 +4387,7 @@ class PGDialect(default.DefaultDialect):
                     ),
                     else_=None,
                 ),
+                pg_catalog.pg_description.c.description,
             )
             .select_from(pg_catalog.pg_class)
             .outerjoin(
@@ -4349,6 +4398,15 @@ class PGDialect(default.DefaultDialect):
                     pg_catalog.pg_constraint.c.contype == "c",
                 ),
             )
+            .outerjoin(
+                pg_catalog.pg_description,
+                pg_catalog.pg_description.c.objoid
+                == pg_catalog.pg_constraint.c.oid,
+            )
+            .order_by(
+                pg_catalog.pg_class.c.relname,
+                pg_catalog.pg_constraint.c.conname,
+            )
             .where(self._pg_class_relkind_condition(relkinds))
         )
         query = self._pg_class_filter_scope_schema(query, schema, scope)
@@ -4369,7 +4427,7 @@ class PGDialect(default.DefaultDialect):
 
         check_constraints = defaultdict(list)
         default = ReflectionDefaults.check_constraints
-        for table_name, check_name, src in result:
+        for table_name, check_name, src, comment in result:
             # only two cases for check_name and src: both null or both defined
             if check_name is None and src is None:
                 check_constraints[(schema, table_name)] = default()
@@ -4391,7 +4449,11 @@ class PGDialect(default.DefaultDialect):
                 sqltext = re.compile(
                     r"^[\s\n]*\((.+)\)[\s\n]*$", flags=re.DOTALL
                 ).sub(r"\1", m.group(1))
-            entry = {"name": check_name, "sqltext": sqltext}
+            entry = {
+                "name": check_name,
+                "sqltext": sqltext,
+                "comment": comment,
+            }
             if m and m.group(2):
                 entry["dialect_options"] = {"not_valid": True}
 
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 22f003e38..35f30566a 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -2535,7 +2535,7 @@ class SQLiteDialect(default.DefaultDialect):
         )
 
         CHECK_PATTERN = r"(?:CONSTRAINT (.+) +)?" r"CHECK *\( *(.+) *\),? *"
-        check_constraints = []
+        cks = []
         # NOTE: we aren't using re.S here because we actually are
         # taking advantage of each CHECK constraint being all on one
         # line in the table definition in order to delineate.  This
@@ -2548,10 +2548,10 @@ class SQLiteDialect(default.DefaultDialect):
             if name:
                 name = re.sub(r'^"|"$', "", name)
 
-            check_constraints.append({"sqltext": match.group(2), "name": name})
-
-        if check_constraints:
-            return check_constraints
+            cks.append({"sqltext": match.group(2), "name": name})
+        cks.sort(key=lambda d: d["name"] or "~")  # sort None as last
+        if cks:
+            return cks
         else:
             return ReflectionDefaults.check_constraints()
 
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 40af06252..5d3ff8bb7 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -113,6 +113,7 @@ class DefaultDialect(Dialect):
     preparer = compiler.IdentifierPreparer
     supports_alter = True
     supports_comments = False
+    supports_constraint_comments = False
     inline_comments = False
     supports_statement_cache = True
 
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 28ed03f99..c33666e4b 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -357,7 +357,21 @@ class ReflectedColumn(TypedDict):
     object"""
 
 
-class ReflectedCheckConstraint(TypedDict):
+class ReflectedConstraint(TypedDict):
+    """Dictionary representing the reflected elements corresponding to
+    :class:`.Constraint`
+
+    A base class for all constraints
+    """
+
+    name: Optional[str]
+    """constraint name"""
+
+    comment: NotRequired[Optional[str]]
+    """comment for the constraint, if present"""
+
+
+class ReflectedCheckConstraint(ReflectedConstraint):
     """Dictionary representing the reflected elements corresponding to
     :class:`.CheckConstraint`.
 
@@ -366,9 +380,6 @@ class ReflectedCheckConstraint(TypedDict):
 
     """
 
-    name: Optional[str]
-    """constraint name"""
-
     sqltext: str
     """the check constraint's SQL expression"""
 
@@ -377,7 +388,7 @@ class ReflectedCheckConstraint(TypedDict):
     object"""
 
 
-class ReflectedUniqueConstraint(TypedDict):
+class ReflectedUniqueConstraint(ReflectedConstraint):
     """Dictionary representing the reflected elements corresponding to
     :class:`.UniqueConstraint`.
 
@@ -386,9 +397,6 @@ class ReflectedUniqueConstraint(TypedDict):
 
     """
 
-    name: Optional[str]
-    """constraint name"""
-
     column_names: List[str]
     """column names which comprise the constraint"""
 
@@ -400,7 +408,7 @@ class ReflectedUniqueConstraint(TypedDict):
     object"""
 
 
-class ReflectedPrimaryKeyConstraint(TypedDict):
+class ReflectedPrimaryKeyConstraint(ReflectedConstraint):
     """Dictionary representing the reflected elements corresponding to
     :class:`.PrimaryKeyConstraint`.
 
@@ -409,9 +417,6 @@ class ReflectedPrimaryKeyConstraint(TypedDict):
 
     """
 
-    name: Optional[str]
-    """constraint name"""
-
     constrained_columns: List[str]
     """column names which comprise the constraint"""
 
@@ -420,7 +425,7 @@ class ReflectedPrimaryKeyConstraint(TypedDict):
     object"""
 
 
-class ReflectedForeignKeyConstraint(TypedDict):
+class ReflectedForeignKeyConstraint(ReflectedConstraint):
     """Dictionary representing the reflected elements corresponding to
     :class:`.ForeignKeyConstraint`.
 
@@ -429,9 +434,6 @@ class ReflectedForeignKeyConstraint(TypedDict):
 
     """
 
-    name: Optional[str]
-    """constraint name"""
-
     constrained_columns: List[str]
     """local column names which comprise the constraint"""
 
@@ -888,6 +890,12 @@ class Dialect(EventTarget):
     definition of a Table or Column.  If False, this implies that ALTER must
     be used to set table and column comments."""
 
+    supports_constraint_comments: bool
+    """Indicates if the dialect supports comment DDL on constraints.
+
+    .. versionadded: 2.0
+    """
+
     _has_events = False
 
     supports_statement_cache: bool = True
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 655a9f5c1..9bac97db0 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -974,6 +974,9 @@ class Inspector(inspection.Inspectable["Inspector"]):
         * ``name`` -
           optional name of the primary key constraint.
 
+        * ``comment`` -
+          optional comment on the primary key constraint.
+
         :param table_name: string name of the table.  For special quoting,
          use :class:`.quoted_name`.
 
@@ -1073,6 +1076,9 @@ class Inspector(inspection.Inspectable["Inspector"]):
         * ``name`` -
           optional name of the foreign key constraint.
 
+        * ``comment`` -
+          optional comment on the foreign key constraint
+
         :param table_name: string name of the table.  For special quoting,
          use :class:`.quoted_name`.
 
@@ -1273,6 +1279,9 @@ class Inspector(inspection.Inspectable["Inspector"]):
         * ``column_names`` -
           list of column names in order
 
+        * ``comment`` -
+          optional comment on the constraint
+
         :param table_name: string name of the table.  For special quoting,
          use :class:`.quoted_name`.
 
@@ -1462,6 +1471,9 @@ class Inspector(inspection.Inspectable["Inspector"]):
           may or may not be present; a dictionary with additional
           dialect-specific options for this CHECK constraint
 
+        * ``comment`` -
+          optional comment on the constraint
+
           .. versionadded:: 1.3.8
 
         :param table_name: string name of the table.  For special quoting,
@@ -1785,8 +1797,9 @@ class Inspector(inspection.Inspectable["Inspector"]):
                 if pk in cols_by_orig_name and pk not in exclude_columns
             ]
 
-            # update pk constraint name
+            # update pk constraint name and comment
             table.primary_key.name = pk_cons.get("name")
+            table.primary_key.comment = pk_cons.get("comment", None)
 
             # tell the PKConstraint to re-initialize
             # its column collection
@@ -1867,6 +1880,7 @@ class Inspector(inspection.Inspectable["Inspector"]):
                     refspec,
                     conname,
                     link_to_name=True,
+                    comment=fkey_d.get("comment"),
                     **options,
                 )
             )
@@ -1948,6 +1962,7 @@ class Inspector(inspection.Inspectable["Inspector"]):
         for const_d in constraints:
             conname = const_d["name"]
             columns = const_d["column_names"]
+            comment = const_d.get("comment")
             duplicates = const_d.get("duplicates_index")
             if include_columns and not set(columns).issubset(include_columns):
                 continue
@@ -1971,7 +1986,9 @@ class Inspector(inspection.Inspectable["Inspector"]):
                 else:
                     constrained_cols.append(constrained_col)
             table.append_constraint(
-                sa_schema.UniqueConstraint(*constrained_cols, name=conname)
+                sa_schema.UniqueConstraint(
+                    *constrained_cols, name=conname, comment=comment
+                )
             )
 
     def _reflect_check_constraints(
diff --git a/lib/sqlalchemy/schema.py b/lib/sqlalchemy/schema.py
index d16bf36b3..f85ef4032 100644
--- a/lib/sqlalchemy/schema.py
+++ b/lib/sqlalchemy/schema.py
@@ -25,6 +25,7 @@ from .sql.ddl import DDL as DDL
 from .sql.ddl import DDLElement as DDLElement
 from .sql.ddl import DropColumnComment as DropColumnComment
 from .sql.ddl import DropConstraint as DropConstraint
+from .sql.ddl import DropConstraintComment as DropConstraintComment
 from .sql.ddl import DropIndex as DropIndex
 from .sql.ddl import DropSchema as DropSchema
 from .sql.ddl import DropSequence as DropSequence
@@ -33,6 +34,7 @@ from .sql.ddl import DropTableComment as DropTableComment
 from .sql.ddl import ExecutableDDLElement as ExecutableDDLElement
 from .sql.ddl import InvokeDDLBase as InvokeDDLBase
 from .sql.ddl import SetColumnComment as SetColumnComment
+from .sql.ddl import SetConstraintComment as SetConstraintComment
 from .sql.ddl import SetTableComment as SetTableComment
 from .sql.ddl import sort_tables as sort_tables
 from .sql.ddl import (
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 1ad547b79..60ec09771 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -5160,6 +5160,12 @@ class DDLCompiler(Compiled):
             drop.element, use_table=True
         )
 
+    def visit_set_constraint_comment(self, create, **kw):
+        raise exc.UnsupportedCompilationError(self, type(create))
+
+    def visit_drop_constraint_comment(self, drop, **kw):
+        raise exc.UnsupportedCompilationError(self, type(drop))
+
     def get_identity_options(self, identity_options):
         text = []
         if identity_options.increment is not None:
diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index eadaa24d3..08d1072c7 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -793,6 +793,18 @@ class DropColumnComment(_CreateDropBase):
     __visit_name__ = "drop_column_comment"
 
 
+class SetConstraintComment(_CreateDropBase):
+    """Represent a COMMENT ON CONSTRAINT IS statement."""
+
+    __visit_name__ = "set_constraint_comment"
+
+
+class DropConstraintComment(_CreateDropBase):
+    """Represent a COMMENT ON CONSTRAINT IS NULL statement."""
+
+    __visit_name__ = "drop_constraint_comment"
+
+
 class InvokeDDLBase(SchemaVisitor):
     def __init__(self, connection):
         self.connection = connection
@@ -933,6 +945,13 @@ class SchemaGenerator(InvokeDDLBase):
                 if column.comment is not None:
                     SetColumnComment(column)._invoke_with(self.connection)
 
+            if self.dialect.supports_constraint_comments:
+                for constraint in table.constraints:
+                    if constraint.comment is not None:
+                        self.connection.execute(
+                            SetConstraintComment(constraint)
+                        )
+
         table.dispatch.after_create(
             table,
             self.connection,
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 569603d79..4158e514b 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -2392,6 +2392,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
         link_to_name: bool = False,
         match: Optional[str] = None,
         info: Optional[_InfoType] = None,
+        comment: Optional[str] = None,
         _unresolvable: bool = False,
         **dialect_kw: Any,
     ):
@@ -2453,6 +2454,11 @@ class ForeignKey(DialectKWArgs, SchemaItem):
 
             .. versionadded:: 1.0.0
 
+        :param comment: Optional string that will render an SQL comment on
+          foreign key constraint creation.
+
+            .. versionadded:: 2.0
+
         :param \**dialect_kw:  Additional keyword arguments are dialect
             specific, and passed in the form ``<dialectname>_<argname>``.  The
             arguments are ultimately handled by a corresponding
@@ -2499,6 +2505,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
         self.initially = initially
         self.link_to_name = link_to_name
         self.match = match
+        self.comment = comment
         if info:
             self.info = info
         self._unvalidated_dialect_kw = dialect_kw
@@ -2539,6 +2546,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
             initially=self.initially,
             link_to_name=self.link_to_name,
             match=self.match,
+            comment=self.comment,
             **self._unvalidated_dialect_kw,
         )
         return self._schema_item_copy(fk)
@@ -2857,6 +2865,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
                 deferrable=self.deferrable,
                 initially=self.initially,
                 match=self.match,
+                comment=self.comment,
                 **self._unvalidated_dialect_kw,
             )
             self.constraint._append_element(column, self)
@@ -3594,6 +3603,7 @@ class Constraint(DialectKWArgs, HasConditionalDDL, SchemaItem):
         deferrable: Optional[bool] = None,
         initially: Optional[str] = None,
         info: Optional[_InfoType] = None,
+        comment: Optional[str] = None,
         _create_rule: Optional[Any] = None,
         _type_bound: bool = False,
         **dialect_kw: Any,
@@ -3616,6 +3626,11 @@ class Constraint(DialectKWArgs, HasConditionalDDL, SchemaItem):
 
             .. versionadded:: 1.0.0
 
+        :param comment: Optional string that will render an SQL comment on
+          foreign key constraint creation.
+
+            .. versionadded:: 2.0
+
         :param \**dialect_kw:  Additional keyword arguments are dialect
             specific, and passed in the form ``<dialectname>_<argname>``.  See
             the documentation regarding an individual dialect at
@@ -3639,6 +3654,7 @@ class Constraint(DialectKWArgs, HasConditionalDDL, SchemaItem):
         self._type_bound = _type_bound
         util.set_creation_order(self)
         self._validate_dialect_kwargs(dialect_kw)
+        self.comment = comment
 
     def _should_create_for_compiler(
         self, compiler: DDLCompiler, **kw: Any
@@ -3921,6 +3937,7 @@ class ColumnCollectionConstraint(ColumnCollectionMixin, Constraint):
                 _copy_expression(expr, self.parent, target_table)
                 for expr in self._columns
             ],
+            comment=self.comment,
             **constraint_kwargs,
         )
         return self._schema_item_copy(c)
@@ -4049,6 +4066,7 @@ class CheckConstraint(ColumnCollectionConstraint):
             deferrable=self.deferrable,
             _create_rule=self._create_rule,
             table=target_table,
+            comment=self.comment,
             _autoattach=False,
             _type_bound=self._type_bound,
         )
@@ -4085,6 +4103,7 @@ class ForeignKeyConstraint(ColumnCollectionConstraint):
         match: Optional[str] = None,
         table: Optional[Table] = None,
         info: Optional[_InfoType] = None,
+        comment: Optional[str] = None,
         **dialect_kw: Any,
     ) -> None:
         r"""Construct a composite-capable FOREIGN KEY.
@@ -4149,6 +4168,11 @@ class ForeignKeyConstraint(ColumnCollectionConstraint):
 
             .. versionadded:: 1.0.0
 
+        :param comment: Optional string that will render an SQL comment on
+          foreign key constraint creation.
+
+            .. versionadded:: 2.0
+
         :param \**dialect_kw:  Additional keyword arguments are dialect
           specific, and passed in the form ``<dialectname>_<argname>``.  See
           the documentation regarding an individual dialect at
@@ -4164,6 +4188,7 @@ class ForeignKeyConstraint(ColumnCollectionConstraint):
             deferrable=deferrable,
             initially=initially,
             info=info,
+            comment=comment,
             **dialect_kw,
         )
         self.onupdate = onupdate
@@ -4360,6 +4385,7 @@ class ForeignKeyConstraint(ColumnCollectionConstraint):
             initially=self.initially,
             link_to_name=self.link_to_name,
             match=self.match,
+            comment=self.comment,
         )
         for self_fk, other_fk in zip(self.elements, fkc.elements):
             self_fk._schema_item_copy(other_fk)
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 038f6e9bd..c0a2bcf65 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -625,6 +625,13 @@ class SuiteRequirements(Requirements):
 
     @property
     def comment_reflection(self):
+        """Indicates if the database support table comment reflection"""
+        return exclusions.closed()
+
+    @property
+    def constraint_comment_reflection(self):
+        """indicates if the database support constraint on constraints
+        and their reflection"""
         return exclusions.closed()
 
     @property
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 7b8e2aa8b..6c71696a0 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -415,35 +415,29 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
             schema_prefix = ""
 
         if testing.requires.self_referential_foreign_keys.enabled:
-            users = Table(
-                "users",
-                metadata,
-                Column("user_id", sa.INT, primary_key=True),
-                Column("test1", sa.CHAR(5), nullable=False),
-                Column("test2", sa.Float(), nullable=False),
-                Column(
-                    "parent_user_id",
-                    sa.Integer,
-                    sa.ForeignKey(
-                        "%susers.user_id" % schema_prefix, name="user_id_fk"
-                    ),
+            parent_id_args = (
+                ForeignKey(
+                    "%susers.user_id" % schema_prefix, name="user_id_fk"
                 ),
-                sa.CheckConstraint("test2 > 0", name="test2_gt_zero"),
-                schema=schema,
-                test_needs_fk=True,
             )
         else:
-            users = Table(
-                "users",
-                metadata,
-                Column("user_id", sa.INT, primary_key=True),
-                Column("test1", sa.CHAR(5), nullable=False),
-                Column("test2", sa.Float(), nullable=False),
-                Column("parent_user_id", sa.Integer),
-                sa.CheckConstraint("test2 > 0", name="test2_gt_zero"),
-                schema=schema,
-                test_needs_fk=True,
-            )
+            parent_id_args = ()
+        users = Table(
+            "users",
+            metadata,
+            Column("user_id", sa.INT, primary_key=True),
+            Column("test1", sa.CHAR(5), nullable=False),
+            Column("test2", sa.Float(), nullable=False),
+            Column("parent_user_id", sa.Integer, *parent_id_args),
+            sa.CheckConstraint(
+                "test2 > 0",
+                name="zz_test2_gt_zero",
+                comment="users check constraint",
+            ),
+            sa.CheckConstraint("test2 <= 1000"),
+            schema=schema,
+            test_needs_fk=True,
+        )
 
         Table(
             "dingalings",
@@ -452,18 +446,27 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
             Column(
                 "address_id",
                 sa.Integer,
-                sa.ForeignKey(
+                ForeignKey(
                     "%semail_addresses.address_id" % schema_prefix,
-                    name="email_add_id_fg",
+                    name="zz_email_add_id_fg",
+                    comment="di fk comment",
                 ),
             ),
+            Column(
+                "id_user",
+                sa.Integer,
+                ForeignKey("%susers.user_id" % schema_prefix),
+            ),
             Column("data", sa.String(30), unique=True),
             sa.CheckConstraint(
                 "address_id > 0 AND address_id < 1000",
                 name="address_id_gt_zero",
             ),
             sa.UniqueConstraint(
-                "address_id", "dingaling_id", name="zz_dingalings_multiple"
+                "address_id",
+                "dingaling_id",
+                name="zz_dingalings_multiple",
+                comment="di unique comment",
             ),
             schema=schema,
             test_needs_fk=True,
@@ -472,11 +475,11 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
             "email_addresses",
             metadata,
             Column("address_id", sa.Integer),
-            Column(
-                "remote_user_id", sa.Integer, sa.ForeignKey(users.c.user_id)
-            ),
+            Column("remote_user_id", sa.Integer, ForeignKey(users.c.user_id)),
             Column("email_address", sa.String(20), index=True),
-            sa.PrimaryKeyConstraint("address_id", name="email_ad_pk"),
+            sa.PrimaryKeyConstraint(
+                "address_id", name="email_ad_pk", comment="ea pk comment"
+            ),
             schema=schema,
             test_needs_fk=True,
         )
@@ -799,6 +802,7 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
             (schema, "dingalings_v"): [
                 col("dingaling_id", auto="omit", nullable=mock.ANY),
                 col("address_id"),
+                col("id_user"),
                 col("data"),
             ]
         }
@@ -831,6 +835,7 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
             (schema, "dingalings"): [
                 pk("dingaling_id"),
                 col("address_id"),
+                col("id_user"),
                 col("data"),
             ],
             (schema, "email_addresses"): [
@@ -873,8 +878,12 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
         kind=ObjectKind.ANY,
         filter_names=None,
     ):
-        def pk(*cols, name=mock.ANY):
-            return {"constrained_columns": list(cols), "name": name}
+        def pk(*cols, name=mock.ANY, comment=None):
+            return {
+                "constrained_columns": list(cols),
+                "name": name,
+                "comment": comment,
+            }
 
         empty = pk(name=None)
         if testing.requires.materialized_views_reflect_pk.enabled:
@@ -890,7 +899,9 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
         tables = {
             (schema, "users"): pk("user_id"),
             (schema, "dingalings"): pk("dingaling_id"),
-            (schema, "email_addresses"): pk("address_id", name="email_ad_pk"),
+            (schema, "email_addresses"): pk(
+                "address_id", name="email_ad_pk", comment="ea pk comment"
+            ),
             (schema, "comment_test"): pk("id"),
             (schema, "no_constraints"): empty,
             (schema, "local_table"): pk("id"),
@@ -926,7 +937,14 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
                     or config.db.dialect.default_schema_name == other
                 )
 
-        def fk(cols, ref_col, ref_table, ref_schema=schema, name=mock.ANY):
+        def fk(
+            cols,
+            ref_col,
+            ref_table,
+            ref_schema=schema,
+            name=mock.ANY,
+            comment=None,
+        ):
             return {
                 "constrained_columns": cols,
                 "referred_columns": ref_col,
@@ -936,6 +954,7 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
                 if ref_schema is not None
                 else tt(),
                 "referred_table": ref_table,
+                "comment": comment,
             }
 
         materialized = {(schema, "dingalings_v"): []}
@@ -950,12 +969,14 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
                 fk(["parent_user_id"], ["user_id"], "users", name="user_id_fk")
             ],
             (schema, "dingalings"): [
+                fk(["id_user"], ["user_id"], "users"),
                 fk(
                     ["address_id"],
                     ["address_id"],
                     "email_addresses",
-                    name="email_add_id_fg",
-                )
+                    name="zz_email_add_id_fg",
+                    comment="di fk comment",
+                ),
             ],
             (schema, "email_addresses"): [
                 fk(["remote_user_id"], ["user_id"], "users")
@@ -1053,6 +1074,7 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
             ],
             (schema, "dingalings"): [
                 *idx("data", name=mock.ANY, unique=True, duplicates=True),
+                *idx("id_user", name=mock.ANY, fk=True),
                 *idx(
                     "address_id",
                     "dingaling_id",
@@ -1118,13 +1140,16 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
         filter_names=None,
         all_=False,
     ):
-        def uc(*cols, name, duplicates_index=None, is_index=False):
+        def uc(
+            *cols, name, duplicates_index=None, is_index=False, comment=None
+        ):
             req = testing.requires.unique_index_reflect_as_unique_constraints
             if is_index and not req.enabled:
                 return ()
             res = {
                 "column_names": list(cols),
                 "name": name,
+                "comment": comment,
             }
             if duplicates_index:
                 res["duplicates_index"] = duplicates_index
@@ -1154,6 +1179,7 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
                     "dingaling_id",
                     name="zz_dingalings_multiple",
                     duplicates_index="zz_dingalings_multiple",
+                    comment="di unique comment",
                 ),
             ],
             (schema, "email_addresses"): [],
@@ -1196,8 +1222,8 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
                 )
                 return self in res
 
-        def cc(text, name):
-            return {"sqltext": tt(text), "name": name}
+        def cc(text, name, comment=None):
+            return {"sqltext": tt(text), "name": name, "comment": comment}
 
         # print({1: "test2 > (0)::double precision"} == {1: tt("test2 > 0")})
         # assert 0
@@ -1209,7 +1235,14 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
         }
         self._resolve_views(views, materialized)
         tables = {
-            (schema, "users"): [cc("test2 > 0", "test2_gt_zero")],
+            (schema, "users"): [
+                cc("test2 <= 1000", mock.ANY),
+                cc(
+                    "test2 > 0",
+                    "zz_test2_gt_zero",
+                    comment="users check constraint",
+                ),
+            ],
             (schema, "dingalings"): [
                 cc(
                     "address_id > 0 and address_id < 1000",
@@ -1764,6 +1797,7 @@ class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
             dupe = refl.pop("duplicates_index", None)
             if dupe:
                 names_that_duplicate_index.add(dupe)
+            eq_(refl.pop("comment", None), None)
             eq_(orig, refl)
 
         reflected_metadata = MetaData()
@@ -2459,7 +2493,7 @@ class ComponentReflectionTestExtra(fixtures.TestBase):
             "table",
             metadata,
             Column("id", Integer, primary_key=True),
-            Column("x_id", Integer, sa.ForeignKey("x.id", name="xid")),
+            Column("x_id", Integer, ForeignKey("x.id", name="xid")),
             Column("test", String(10)),
             test_needs_fk=True,
         )