fix: dataset after insert when db relation does not exist

superset/security/manager.py

@@ -1101,7 +1101,9 @@ def _update_vm_database_access(
             .where(view_menu_table.c.id == db_pvm.view_menu_id)
             .values(name=new_view_menu_name)
         )
-        new_db_view_menu = self.find_view_menu(new_view_menu_name)
+        new_db_view_menu = self._find_view_menu_on_sqla_event(
+            connection, new_view_menu_name
+        )
 
         self.on_view_menu_after_update(mapper, connection, new_db_view_menu)
         return new_db_view_menu
@@ -1155,7 +1157,9 @@ def _update_vm_datasources_access(  # pylint: disable=too-many-locals
                 .values(name=new_dataset_vm_name)
             )
             # After update refresh
-            new_dataset_view_menu = self.find_view_menu(new_dataset_vm_name)
+            new_dataset_view_menu = self._find_view_menu_on_sqla_event(
+                connection, new_dataset_vm_name
+            )
 
             # Update dataset (SqlaTable perm field)
             connection.execute(
@@ -1194,11 +1198,21 @@ def dataset_after_insert(
         :param target: The changed dataset object
         :return:
         """
+        from superset.models.core import (  # pylint: disable=import-outside-toplevel
+            Database,
+        )
+
         try:
             dataset_perm = target.get_perm()
+            database = target.database
         except DatasetInvalidPermissionEvaluationException:
-            logger.warning("Dataset has no database refusing to set permission")
-            return
+            logger.warning(
+                "Dataset has no database will retry with database_id to set permission"
+            )
+            database = self.get_session.query(Database).get(target.database_id)
+            dataset_perm = self.get_dataset_perm(
+                target.id, target.table_name, database.database_name
+            )
         dataset_table = target.__table__
 
         self._insert_pvm_on_sqla_event(
@@ -1214,7 +1228,7 @@ def dataset_after_insert(
 
         if target.schema:
             dataset_schema_perm = self.get_schema_perm(
-                target.database.database_name, target.schema
+                database.database_name, target.schema
             )
             self._insert_pvm_on_sqla_event(
                 mapper, connection, "schema_access", dataset_schema_perm
@@ -1484,12 +1498,23 @@ def _delete_pvm_on_sqla_event(  # pylint: disable=too-many-arguments
     def _find_permission_on_sqla_event(
         self, connection: Connection, name: str
     ) -> Permission:
+        """
+        Find a FAB Permission using a SQLA connection.
+
+        A session.query may not return the latest results on newly created/updated
+        objects/rows using connection. On this case we should use a connection also
+
+        :param connection: SQLAlchemy connection
+        :param name: The permission name (it's unique)
+        :return: Permission
+        """
         permission_table = self.permission_model.__table__  # pylint: disable=no-member
 
         permission_ = connection.execute(
             permission_table.select().where(permission_table.c.name == name)
         ).fetchone()
         permission = Permission()
+        # ensures this object is never persisted
         permission.metadata = None
         permission.id = permission_.id
         permission.name = permission_.name
@@ -1498,12 +1523,23 @@ def _find_permission_on_sqla_event(
     def _find_view_menu_on_sqla_event(
         self, connection: Connection, name: str
     ) -> ViewMenu:
+        """
+        Find a FAB ViewMenu using a SQLA connection.
+
+        A session.query may not return the latest results on newly created/updated
+        objects/rows using connection. On this case we should use a connection also
+
+        :param connection: SQLAlchemy connection
+        :param name: The ViewMenu name (it's unique)
+        :return: ViewMenu
+        """
         view_menu_table = self.viewmenu_model.__table__  # pylint: disable=no-member
 
         view_menu_ = connection.execute(
             view_menu_table.select().where(view_menu_table.c.name == name)
         ).fetchone()
         view_menu = ViewMenu()
+        # ensures this object is never persisted
         view_menu.metadata = None
         view_menu.id = view_menu_.id
         view_menu.name = view_menu_.name

tests/integration_tests/security_tests.py

@@ -248,8 +248,8 @@ def test_after_insert_dataset_table_none(self):
         stored_table = (
             session.query(SqlaTable).filter_by(table_name="tmp_perm_table").one()
         )
-        # Assert no permission is created
-        self.assertIsNone(
+        # Assert permission is created
+        self.assertIsNotNone(
             security_manager.find_permission_view_menu(
                 "datasource_access", stored_table.perm
             )