Update snowdialect.py

Refactored...

Author: sfc-gh-kterada

src/snowflake/sqlalchemy/snowdialect.py

@@ -319,19 +319,12 @@ def get_check_constraints(self, connection, table_name, schema, **kw):
 
     @reflection.cache
     def _get_table_primary_keys(self, connection, schema, table_name, **kw):
-        if table_name in SnowflakeIdentifierPreparer.reserved_words:
-            fully_qualified_path = self._denormalize_quote_join(schema, table_name)
-            result = connection.execute(
-                text(
-                    f"SHOW /* sqlalchemy:_get_table_primary_keys */ PRIMARY KEYS IN TABLE {fully_qualified_path}"
-                )
-            )
-        else:
-            result = connection.execute(
-                text(
-                    f"SHOW /* sqlalchemy:_get_table_primary_keys */ PRIMARY KEYS IN TABLE {schema}.{table_name}"
-                )
+        fully_qualified_path = self._denormalize_quote_join(schema, self.denormalize_name(table_name))
+        result = connection.execute(
+            text(
+                f"SHOW /* sqlalchemy:_get_table_primary_keys */ PRIMARY KEYS IN TABLE {fully_qualified_path}"
             )
+        )
         ans = {}
         for row in result:
             table_name = self.normalize_name(row._mapping["table_name"])
@@ -388,19 +381,12 @@ def get_pk_constraint(self, connection, table_name, schema=None, **kw):
 
     @reflection.cache
     def _get_table_unique_constraints(self, connection, schema, table_name, **kw):
-        if table_name in SnowflakeIdentifierPreparer.reserved_words:
-            fully_qualified_path = self._denormalize_quote_join(schema, table_name)
-            result = connection.execute(
-                text(
-                    f"SHOW /* sqlalchemy:_get_table_unique_constraints */ UNIQUE KEYS IN TABLE {fully_qualified_path}"
-                )
-            )
-        else:
-            result = connection.execute(
-                text(
-                    f"SHOW /* sqlalchemy:_get_table_unique_constraints */ UNIQUE KEYS IN TABLE {schema}.{table_name}"
-                )
+        fully_qualified_path = self._denormalize_quote_join(schema, table_name)
+        result = connection.execute(
+            text(
+                f"SHOW /* sqlalchemy:_get_table_unique_constraints */ UNIQUE KEYS IN TABLE {fully_qualified_path}"
             )
+        )
         unique_constraints = {}
         for row in result:
             name = self.normalize_name(row._mapping["constraint_name"])
@@ -471,19 +457,12 @@ def get_unique_constraints(self, connection, table_name, schema, **kw):
     @reflection.cache
     def _get_table_foreign_keys(self, connection, schema, table_name, **kw):
         _, current_schema = self._current_database_schema(connection, **kw)
-        if table_name in SnowflakeIdentifierPreparer.reserved_words:
-            fully_qualified_path = self._denormalize_quote_join(schema, table_name)
-            result = connection.execute(
-                text(
-                    f"SHOW /* sqlalchemy:_get_table_foreign_keys */ IMPORTED KEYS IN TABLE {fully_qualified_path}"
-                )
-            )
-        else:
-            result = connection.execute(
-                text(
-                    f"SHOW /* sqlalchemy:_get_table_foreign_keys */ IMPORTED KEYS IN TABLE {schema}.{table_name}"
-                )
+        fully_qualified_path = self._denormalize_quote_join(schema, self.denormalize_name(table_name))
+        result = connection.execute(
+            text(
+                f"SHOW /* sqlalchemy:_get_table_foreign_keys */ IMPORTED KEYS IN TABLE {fully_qualified_path}"
             )
+        )
         foreign_key_map = {}
         for row in result:
             name = self.normalize_name(row._mapping["fk_name"])