fix: Catalog with restricted permissions produces an error during dat…

…abase connection (apache#29257)

.gitignore

@@ -116,3 +116,4 @@ docker/requirements-local.txt
 
 cache/
 docker/*local*
+.temp_cache

superset/commands/database/create.py

@@ -40,6 +40,7 @@
 from superset.commands.database.test_connection import TestConnectionDatabaseCommand
 from superset.daos.database import DatabaseDAO
 from superset.daos.exceptions import DAOCreateFailedError
+from superset.databases.ssh_tunnel.models import SSHTunnel
 from superset.exceptions import SupersetErrorsException
 from superset.extensions import db, event_logger, security_manager
 from superset.models.core import Database
@@ -82,7 +83,7 @@ def run(self) -> Model:
             "{}",
         )
 
-        ssh_tunnel = None
+        ssh_tunnel: Optional[SSHTunnel] = None
 
         try:
             database = self._create_database()
@@ -115,19 +116,11 @@ def run(self) -> Model:
                 catalogs = [None]
 
             for catalog in catalogs:
-                for schema in database.get_all_schema_names(
-                    catalog=catalog,
-                    cache=False,
-                    ssh_tunnel=ssh_tunnel,
-                ):
-                    security_manager.add_permission_view_menu(
-                        "schema_access",
-                        security_manager.get_schema_perm(
-                            database.database_name,
-                            catalog,
-                            schema,
-                        ),
-                    )
+                try:
+                    self.add_schema_permissions(database, catalog, ssh_tunnel)
+                except Exception:  # pylint: disable=broad-except
+                    logger.warning("Error processing catalog '%s'", catalog)
+                    continue
 
         except (
             SSHTunnelInvalidError,
@@ -159,6 +152,26 @@ def run(self) -> Model:
 
         return database
 
+    def add_schema_permissions(
+        self,
+        database: Database,
+        catalog: str,
+        ssh_tunnel: Optional[SSHTunnel],
+    ) -> None:
+        for schema in database.get_all_schema_names(
+            catalog=catalog,
+            cache=False,
+            ssh_tunnel=ssh_tunnel,
+        ):
+            security_manager.add_permission_view_menu(
+                "schema_access",
+                security_manager.get_schema_perm(
+                    database.database_name,
+                    catalog,
+                    schema,
+                ),
+            )
+
     def validate(self) -> None:
         exceptions: list[ValidationError] = []
         sqlalchemy_uri: Optional[str] = self._properties.get("sqlalchemy_uri")