fix(embedded+async queries): support async queries to work with embedded guest user

superset/async_events/async_query_manager.py

@@ -191,20 +191,40 @@
         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)
+        if guest_user := security_manager.get_current_guest_user_if_guest():
+            job_metadata["guest_token"] = guest_user.guest_token
         self._load_explore_json_into_cache_job.delay(
             job_metadata,
             form_data,
             response_type,
             force,
         )
+        # clean up guest token so that it doesn't get exposed upper level
+        if "guest_token" in job_metadata:
+            del job_metadata["guest_token"]
         return job_metadata
 
     def submit_chart_data_job(
         self, channel_id: str, form_data: dict[str, Any], user_id: Optional[int]
     ) -> 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)
+        if guest_user := security_manager.get_current_guest_user_if_guest():
+            job_metadata["guest_token"] = guest_user.guest_token
         self._load_chart_data_into_cache_job.delay(job_metadata, form_data)
+        # clean up guest token so that it doesn't get exposed upper level
+        if "guest_token" in job_metadata:
+            del job_metadata["guest_token"]
         return job_metadata
 
     def read_events(

superset/common/query_context_processor.py

@@ -600,7 +600,15 @@ def get_payload(
             set_and_log_cache(
                 cache_manager.cache,
                 cache_key,
-                {"data": self._query_context.cache_values},
+                {
+                    "data": {
+                        # setting form_data into query context cache value as well
+                        # so that it can be used to reconstruct form_data field
+                        # for query context object when reading from cache
+                        "form_data": self._query_context.form_data,
+                        **self._query_context.cache_values,
+                    },
+                },
                 self.get_cache_timeout(),
             )
             return_value["cache_key"] = cache_key  # type: ignore

superset/tasks/async_queries.py

@@ -66,10 +66,16 @@
     # pylint: disable=import-outside-toplevel
     from superset.commands.chart.data.get_data_command import ChartDataCommand
 
-    user = (
-        security_manager.get_user_by_id(job_metadata.get("user_id"))
-        or security_manager.get_anonymous_user()
-    )
+    if user_id := job_metadata.get("user_id"):
+        # logged in user
+        user = security_manager.get_user_by_id(user_id)
+    elif guest_token := job_metadata.get("guest_token"):
+        # embedded guest user
+        user = security_manager.get_guest_user_from_token(guest_token)
+        del job_metadata["guest_token"]
+    else:
+        # default to anonymous user if no user is found
+        user = security_manager.get_anonymous_user()
 
     with override_user(user, force=False):
         try:
@@ -106,10 +112,16 @@
 ) -> None:
     cache_key_prefix = "ejr-"  # ejr: explore_json request
 
-    user = (
-        security_manager.get_user_by_id(job_metadata.get("user_id"))
-        or security_manager.get_anonymous_user()
-    )
+    if user_id := job_metadata.get("user_id"):
+        # logged in user
+        user = security_manager.get_user_by_id(user_id)
+    elif guest_token := job_metadata.get("guest_token"):
+        # embedded guest user
+        user = security_manager.get_guest_user_from_token(guest_token)
+        del job_metadata["guest_token"]
+    else:
+        # default to anonymous user if no user is found
+        user = security_manager.get_anonymous_user()
 
     with override_user(user, force=False):
         try:
@@ -140,7 +152,13 @@
                 "response_type": response_type,
             }
             cache_key = generate_cache_key(cache_value, cache_key_prefix)
-            set_and_log_cache(cache_manager.cache, cache_key, cache_value)
+            cache_instance = cache_manager.cache
+            cache_timeout = (
+                cache_instance.cache.default_timeout if cache_instance.cache else None
+            )
+            set_and_log_cache(
+                cache_instance, cache_key, cache_value, cache_timeout=cache_timeout
+            )
             result_url = f"/superset/explore_json/data/{cache_key}"
             async_query_manager.update_job(
                 job_metadata,