New method get_unique_constraints II-14372 (#52)

lib/sqlalchemy_ingres/_version.py

@@ -1,2 +1,2 @@
-version_tuple = __version_info__ = (0, 0, 7, "dev4")
+version_tuple = __version_info__ = (0, 0, 7, "dev5")
 version = version_string = __version__ = '.'.join(map(str, __version_info__))

lib/sqlalchemy_ingres/base.py

@@ -474,6 +474,37 @@ def get_columns(self, connection, table_name, schema=None, **kw):
         finally:
             if rs:
                 rs.close()
+
+    @reflection.cache
+    def get_unique_constraints(self, connection, table_name, schema=None, **kw):
+        sqltext = """
+            SELECT
+                k.column_name
+            FROM
+                iikeys k,
+                iiconstraints c
+            WHERE
+                k.constraint_name = c.constraint_name
+            AND c.constraint_type = 'U'
+            AND k.table_name = ?"""
+        params = (self.denormalize_name(table_name),)
+
+        if schema:
+            sqltext += """
+                AND k.schema_name = ?"""
+            params = (*params, self.denormalize_name(schema))
+
+        rs = None
+
+        try:
+            rs = connection.exec_driver_sql(sqltext, params)
+
+            cols = [row[0].rstrip() for row in rs.fetchall()]
+            return {"constrained_columns": [] if cols is None else cols, "name": None}
+
+        finally:
+            if rs:
+                rs.close()
 
     @reflection.cache
     def get_primary_keys(self, connection, table_name, schema=None, **kw):