added helper object to handle selection in launcher

client/ayon_core/pipeline/actions.py

@@ -1,4 +1,7 @@
 import logging
+
+import ayon_api
+
 from ayon_core.pipeline.plugin_discover import (
     discover,
     register_plugin,
@@ -10,6 +13,224 @@ from ayon_core.pipeline.plugin_discover import (
 from .load.utils import get_representation_path_from_context
 
 
+class LauncherActionSelection:
+    """Object helper to pass selection to actions.
+
+    Object support backwards compatibility for 'session' from OpenPype where
+    environment variable keys were used to define selection.
+
+    Args:
+        project_name (str): Selected project name.
+        folder_id (str): Selected folder id.
+        task_id (str): Selected task id.
+        folder_path (Optional[str]): Selected folder path.
+        task_name (Optional[str]): Selected task name.
+        project_entity (Optional[dict[str, Any]]): Project entity.
+        folder_entity (Optional[dict[str, Any]]): Folder entity.
+        task_entity (Optional[dict[str, Any]]): Task entity.
+
+    """
+    def __init__(
+        self,
+        project_name,
+        folder_id,
+        task_id,
+        folder_path=None,
+        task_name=None,
+        project_entity=None,
+        folder_entity=None,
+        task_entity=None
+    ):
+        self._project_name = project_name
+        self._folder_id = folder_id
+        self._task_id = task_id
+
+        self._folder_path = folder_path
+        self._task_name = task_name
+
+        self._project_entity = project_entity
+        self._folder_entity = folder_entity
+        self._task_entity = task_entity
+
+    def __getitem__(self, key):
+        if key in {"project_name", "AYON_PROJECT_NAME", "AVALON_PROJECT"}:
+            return self.project_name
+        if key == {"folder_path", "AYON_FOLDER_PATH", "AVALON_ASSET"}:
+            return self.folder_path
+        if key == {"task_name", "AYON_TASK_NAME", "AVALON_TASK"}:
+            return self.task_name
+        if key == "folder_id":
+            return self.folder_id
+        if key == "task_id":
+            return self.task_id
+        if key == "project_entity":
+            return self.project_entity
+        if key == "folder_entity":
+            return self.folder_entity
+        if key == "task_entity":
+            return self.task_entity
+        raise KeyError(f"Key: {key} not found")
+
+    def __contains__(self, key):
+        # Fake missing keys check for backwards compatibility
+        if key in {
+            "AYON_PROJECT_NAME",
+            "AVALON_PROJECT",
+            "project_entity",
+        }:
+            return self._project_name is not None
+        if key in {
+            "AYON_FOLDER_PATH",
+            "folder_id",
+            "folder_path",
+            "folder_entity",
+            "AVALON_ASSET",
+        }:
+            return self._folder_id is not None
+        if key in {
+            "AYON_TASK_NAME",
+            "task_id",
+            "task_name",
+            "task_entity",
+            "AVALON_TASK",
+        }:
+            return self._task_id is not None
+        return False
+
+    def get(self, key, default=None):
+        try:
+            return self[key]
+        except KeyError:
+            return default
+
+    def get_project_name(self):
+        """Selected project name.
+
+        Returns:
+            Union[str, None]: Selected project name.
+
+        """
+        return self._project_name
+
+    def get_folder_id(self):
+        """Selected folder id.
+
+        Returns:
+            Union[str, None]: Selected folder id.
+
+        """
+        return self._folder_id
+
+    def get_folder_path(self):
+        """Selected folder path.
+
+        Returns:
+            Union[str, None]: Selected folder path.
+
+        """
+        if self._folder_id is None:
+            return None
+        if self._folder_path is None:
+            self._folder_path = self.folder_entity["path"]
+        return self._folder_path
+
+    def get_task_id(self):
+        """Selected task id.
+
+        Returns:
+            Union[str, None]: Selected task id.
+
+        """
+        return self._task_id
+
+    def get_task_name(self):
+        """Selected task name.
+
+        Returns:
+            Union[str, None]: Selected task name.
+
+        """
+        if self._task_id is None:
+            return None
+        if self._task_name is None:
+            self._task_name = self.task_entity["name"]
+        return self._task_name
+
+    def get_project_entity(self):
+        """Project entity for the selection.
+
+        Returns:
+            Union[dict[str, Any], None]: Project entity.
+
+        """
+        if self._project_name is None:
+            return None
+        if self._project_entity is None:
+            self._project_entity = ayon_api.get_project(self._project_name)
+        return self._project_entity
+
+    def get_folder_entity(self):
+        """Folder entity for the selection.
+
+        Returns:
+            Union[dict[str, Any], None]: Folder entity.
+
+        """
+        if self._project_name is None or self._folder_id is None:
+            return None
+        if self._folder_entity is None:
+            self._folder_entity = ayon_api.get_folder_by_id(
+                self._project_name, self._folder_id
+            )
+        return self._folder_entity
+
+    def get_task_entity(self):
+        """Task entity for the selection.
+
+        Returns:
+            Union[dict[str, Any], None]: Task entity.
+
+        """
+        if (
+            self._project_name is None
+            or self._task_id is None
+        ):
+            return None
+        if self._task_entity is None:
+            self._task_entity = ayon_api.get_task_by_id(
+                self._project_name, self._task_id
+            )
+        return self._task_entity
+
+    @property
+    def is_project_selected(self):
+        """Return whether a project is selected.
+
+        Returns:
+            bool: Whether a project is selected.
+
+        """
+        return self._project_name is not None
+
+    @property
+    def is_folder_selected(self):
+        return self._folder_id is not None
+
+    @property
+    def is_task_selected(self):
+        return self._task_id is not None
+
+    project_name = property(get_project_name)
+    folder_id = property(get_folder_id)
+    task_id = property(get_task_id)
+    folder_path = property(get_folder_path)
+    task_name = property(get_task_name)
+
+    project_entity = property(get_project_entity)
+    folder_entity = property(get_folder_entity)
+    task_entity = property(get_task_entity)
+
+
 class LauncherAction(object):
     """A custom action available"""
     name = None
@@ -21,17 +242,23 @@ class LauncherAction(object):
     log = logging.getLogger("LauncherAction")
     log.propagate = True
 
-    def is_compatible(self, session):
+    def is_compatible(self, selection):
         """Return whether the class is compatible with the Session.
 
         Args:
-            session (dict[str, Union[str, None]]): Session data with
-                AYON_PROJECT_NAME, AYON_FOLDER_PATH and AYON_TASK_NAME.
-        """
+            selection (LauncherActionSelection): Data with selection.
 
+        """
         return True
 
-    def process(self, session, **kwargs):
+    def process(self, selection, **kwargs):
+        """Process the action.
+
+        Args:
+            selection (LauncherActionSelection): Data with selection.
+            **kwargs: Additional arguments.
+
+        """
         pass
 
 

client/ayon_core/tools/launcher/models/actions.py

@@ -5,6 +5,7 @@ from ayon_core.lib import Logger, AYONSettingsRegistry
 from ayon_core.pipeline.actions import (
     discover_launcher_actions,
     LauncherAction,
+    LauncherActionSelection,
 )
 from ayon_core.pipeline.workfile import should_use_last_workfile_on_launch
 
@@ -69,11 +70,6 @@ class ApplicationAction(LauncherAction):
     project_entities = {}
 
     _log = None
-    required_session_keys = (
-        "AYON_PROJECT_NAME",
-        "AYON_FOLDER_PATH",
-        "AYON_TASK_NAME"
-    )
 
     @property
     def log(self):
@@ -81,18 +77,16 @@ class ApplicationAction(LauncherAction):
             self._log = Logger.get_logger(self.__class__.__name__)
         return self._log
 
-    def is_compatible(self, session):
-        for key in self.required_session_keys:
-            if not session.get(key):
-                return False
+    def is_compatible(self, selection):
+        if not selection.is_task_selected:
+            return False
 
-        project_name = session["AYON_PROJECT_NAME"]
-        project_entity = self.project_entities[project_name]
+        project_entity = self.project_entities[selection.project_name]
         apps = project_entity["attrib"].get("applications")
         if not apps or self.application.full_name not in apps:
             return False
 
-        project_settings = self.project_settings[project_name]
+        project_settings = self.project_settings[selection.project_name]
         only_available = project_settings["applications"]["only_available"]
         if only_available and not self.application.find_executable():
             return False
@@ -112,7 +106,7 @@ class ApplicationAction(LauncherAction):
             dialog.setDetailedText(details)
         dialog.exec_()
 
-    def process(self, session, **kwargs):
+    def process(self, selection, **kwargs):
         """Process the full Application action"""
 
         from ayon_core.lib import (
@@ -120,14 +114,11 @@ class ApplicationAction(LauncherAction):
             ApplicationLaunchFailed,
         )
 
-        project_name = session["AYON_PROJECT_NAME"]
-        folder_path = session["AYON_FOLDER_PATH"]
-        task_name = session["AYON_TASK_NAME"]
         try:
             self.application.launch(
-                project_name=project_name,
-                folder_path=folder_path,
-                task_name=task_name,
+                project_name=selection.project_name,
+                folder_path=selection.folder_path,
+                task_name=selection.task_name,
                 **self.data
             )
 
@@ -335,11 +326,11 @@ class ActionsModel:
         """
         not_open_workfile_actions = self._get_no_last_workfile_for_context(
             project_name, folder_id, task_id)
-        session = self._prepare_session(project_name, folder_id, task_id)
+        selection = self._prepare_selection(project_name, folder_id, task_id)
         output = []
         action_items = self._get_action_items(project_name)
         for identifier, action in self._get_action_objects().items():
-            if not action.is_compatible(session):
+            if not action.is_compatible(selection):
                 continue
 
             action_item = action_items[identifier]
@@ -374,7 +365,7 @@ class ActionsModel:
         )
 
     def trigger_action(self, project_name, folder_id, task_id, identifier):
-        session = self._prepare_session(project_name, folder_id, task_id)
+        selection = self._prepare_selection(project_name, folder_id, task_id)
         failed = False
         error_message = None
         action_label = identifier
@@ -403,7 +394,7 @@ class ActionsModel:
                 )
                 action.data["start_last_workfile"] = start_last_workfile
 
-            action.process(session)
+            action.process(selection)
         except Exception as exc:
             self.log.warning("Action trigger failed.", exc_info=True)
             failed = True
@@ -440,29 +431,8 @@ class ActionsModel:
             .get(task_id, {})
         )
 
-    def _prepare_session(self, project_name, folder_id, task_id):
-        folder_path = None
-        if folder_id:
-            folder = self._controller.get_folder_entity(
-                project_name, folder_id)
-            if folder:
-                folder_path = folder["path"]
-
-        task_name = None
-        if task_id:
-            task = self._controller.get_task_entity(project_name, task_id)
-            if task:
-                task_name = task["name"]
-
-        return {
-            "AYON_PROJECT_NAME": project_name,
-            "AYON_FOLDER_PATH": folder_path,
-            "AYON_TASK_NAME": task_name,
-            # Deprecated - kept for backwards compatibility
-            "AVALON_PROJECT": project_name,
-            "AVALON_ASSET": folder_path,
-            "AVALON_TASK": task_name,
-        }
+    def _prepare_selection(self, project_name, folder_id, task_id):
+        return LauncherActionSelection(project_name, folder_id, task_id)
 
     def _get_discovered_action_classes(self):
         if self._discovered_actions is None: