feat(embedded+async queries): support async queries to work with embedded guest user (#26332)

superset/async_events/async_query_manager.py

@@ -191,9 +191,14 @@ class AsyncQueryManager:
         force: Optional[bool] = False,
         user_id: Optional[int] = None,
     ) -> dict[str, Any]:
+        # pylint: disable=import-outside-toplevel
+        from superset import security_manager
+
         job_metadata = self.init_job(channel_id, user_id)
         self._load_explore_json_into_cache_job.delay(
-            job_metadata,
+            {**job_metadata, "guest_token": guest_user.guest_token}
+            if (guest_user := security_manager.get_current_guest_user_if_guest())
+            else job_metadata,
             form_data,
             response_type,
             force,
@@ -201,10 +206,25 @@ class AsyncQueryManager:
         return job_metadata
 
     def submit_chart_data_job(
-        self, channel_id: str, form_data: dict[str, Any], user_id: Optional[int]
+        self,
+        channel_id: str,
+        form_data: dict[str, Any],
+        user_id: Optional[int] = None,
     ) -> dict[str, Any]:
+        # pylint: disable=import-outside-toplevel
+        from superset import security_manager
+
+        # if it's guest user, we want to pass the guest token to the celery task
+        # chart data cache key is calculated based on the current user
+        # this way we can keep the cache key consistent between sync and async command
+        # so that it can be looked up consistently
         job_metadata = self.init_job(channel_id, user_id)
-        self._load_chart_data_into_cache_job.delay(job_metadata, form_data)
+        self._load_chart_data_into_cache_job.delay(
+            {**job_metadata, "guest_token": guest_user.guest_token}
+            if (guest_user := security_manager.get_current_guest_user_if_guest())
+            else job_metadata,
+            form_data,
+        )
         return job_metadata
 
     def read_events(