reduce usage of legay_io

client/ayon_core/pipeline/context_tools.py

@@ -29,11 +29,11 @@ from .publish.lib import filter_pyblish_plugins
 from .anatomy import Anatomy
 from .template_data import get_template_data_with_names
 from .workfile import (
+    get_workdir,
     get_workfile_template_key,
     get_custom_workfile_template_by_string_context,
 )
 from . import (
-    legacy_io,
     register_loader_plugin_path,
     register_inventory_action_path,
     register_creator_plugin_path,
@@ -116,22 +116,15 @@ def install_host(host):
     # Make sure global AYON connection has set site id and version
     get_ayon_server_api_connection()
 
-    legacy_io.install()
     addons_manager = _get_addons_manager()
 
-    missing = list()
-    for key in ("AVALON_PROJECT", "AVALON_ASSET"):
-        if key not in legacy_io.Session:
-            missing.append(key)
+    project_name = os.getenv("AVALON_PROJECT")
+    if not project_name:
+        raise ValueError(
+            "AVALON_PROJECT is missing in environment variables."
+        )
 
-    assert not missing, (
-        "%s missing from environment, %s" % (
-            ", ".join(missing),
-            json.dumps(legacy_io.Session, indent=4, sort_keys=True)
-        ))
-
-    project_name = legacy_io.Session["AVALON_PROJECT"]
-    log.info("Activating %s.." % project_name)
+    log.info("Activating {}..".format(project_name))
 
     # Optional host install function
     if hasattr(host, "install"):
@@ -158,7 +151,6 @@ def install_host(host):
         print("Registering pyblish target: automated")
         pyblish.api.register_target("automated")
 
-    project_name = os.environ.get("AVALON_PROJECT")
     host_name = os.environ.get("AVALON_APP")
 
     # Give option to handle host installation
@@ -256,8 +248,6 @@ def uninstall_host():
 
     deregister_host()
 
-    legacy_io.uninstall()
-
     log.info("Successfully uninstalled Avalon!")
 
 
@@ -482,13 +472,17 @@ def get_template_data_from_session(session=None, system_settings=None):
         Dict[str, Any]: All available data from session.
     """
 
-    if session is None:
-        session = legacy_io.Session
-
-    project_name = session["AVALON_PROJECT"]
-    asset_name = session["AVALON_ASSET"]
-    task_name = session["AVALON_TASK"]
-    host_name = session["AVALON_APP"]
+    if session is not None:
+        project_name = session["AVALON_PROJECT"]
+        asset_name = session["AVALON_ASSET"]
+        task_name = session["AVALON_TASK"]
+        host_name = session["AVALON_APP"]
+    else:
+        context = get_current_context()
+        project_name = context["project_name"]
+        asset_name = context["asset_name"]
+        task_name = context["task_name"]
+        host_name = get_current_host_name()
 
     return get_template_data_with_names(
         project_name, asset_name, task_name, host_name, system_settings
@@ -529,10 +523,12 @@ def get_workdir_from_session(session=None, template_key=None):
         str: Workdir path.
     """
 
-    if session is None:
-        session = legacy_io.Session
-    project_name = session["AVALON_PROJECT"]
-    host_name = session["AVALON_APP"]
+    if session is not None:
+        project_name = session["AVALON_PROJECT"]
+        host_name = session["AVALON_APP"]
+    else:
+        project_name = get_current_project_name()
+        host_name = get_current_host_name()
     template_data = get_template_data_from_session(session)
 
     if not template_key:
@@ -556,86 +552,39 @@ def get_custom_workfile_template_from_session(
 ):
     """Filter and fill workfile template profiles by current context.
 
-    Current context is defined by `legacy_io.Session`. That's why this
-    function should be used only inside host where context is set and stable.
+    This function cab be used only inside host where context is set.
 
     Args:
-        session (Union[None, Dict[str, str]]): Session from which are taken
+        session (Optional[Dict[str, str]]): Session from which are taken
             data.
-        project_settings(Dict[str, Any]): Template profiles from settings.
+        project_settings(Optional[Dict[str, Any]]): Project settings.
 
     Returns:
         str: Path to template or None if none of profiles match current
             context. (Existence of formatted path is not validated.)
     """
 
-    if session is None:
-        session = legacy_io.Session
+    if session is not None:
+        project_name = session["AVALON_PROJECT"]
+        asset_name = session["AVALON_ASSET"]
+        task_name = session["AVALON_TASK"]
+        host_name = session["AVALON_APP"]
+    else:
+        context = get_current_context()
+        project_name = context["project_name"]
+        asset_name = context["asset_name"]
+        task_name = context["task_name"]
+        host_name = get_current_host_name()
 
     return get_custom_workfile_template_by_string_context(
-        session["AVALON_PROJECT"],
-        session["AVALON_ASSET"],
-        session["AVALON_TASK"],
-        session["AVALON_APP"],
+        project_name,
+        asset_name,
+        task_name,
+        host_name,
         project_settings=project_settings
     )
 
 
-def compute_session_changes(
-    session, asset_doc, task_name, template_key=None
-):
-    """Compute the changes for a session object on task under asset.
-
-    Function does not change the session object, only returns changes.
-
-    Args:
-        session (Dict[str, str]): The initial session to compute changes to.
-            This is required for computing the full Work Directory, as that
-            also depends on the values that haven't changed.
-        asset_doc (Dict[str, Any]): Asset document to switch to.
-        task_name (str): Name of task to switch to.
-        template_key (Union[str, None]): Prepare workfile template key in
-            anatomy templates.
-
-    Returns:
-        Dict[str, str]: Changes in the Session dictionary.
-    """
-
-    # Get asset document and asset
-    if not asset_doc:
-        task_name = None
-        asset_name = None
-    else:
-        asset_name = get_asset_name_identifier(asset_doc)
-
-    # Detect any changes compared session
-    mapping = {
-        "AVALON_ASSET": asset_name,
-        "AVALON_TASK": task_name,
-    }
-    changes = {
-        key: value
-        for key, value in mapping.items()
-        if value != session.get(key)
-    }
-    if not changes:
-        return changes
-
-    # Compute work directory (with the temporary changed session so far)
-    changed_session = session.copy()
-    changed_session.update(changes)
-
-    workdir = None
-    if asset_doc:
-        workdir = get_workdir_from_session(
-            changed_session, template_key
-        )
-
-    changes["AVALON_WORKDIR"] = workdir
-
-    return changes
-
-
 def change_current_context(asset_doc, task_name, template_key=None):
     """Update active Session to a new task work area.
 
@@ -651,32 +600,47 @@ def change_current_context(asset_doc, task_name, template_key=None):
         Dict[str, str]: The changed key, values in the current Session.
     """
 
-    changes = compute_session_changes(
-        legacy_io.Session,
-        asset_doc,
-        task_name,
-        template_key=template_key
-    )
+    project_name = get_current_project_name()
+    workdir = None
+    if asset_doc:
+        project_doc = get_project(project_name)
+        host_name = get_current_host_name()
+        workdir = get_workdir(
+            project_doc,
+            asset_doc,
+            task_name,
+            host_name,
+            template_key=template_key
+        )
+
+    folder_path = get_asset_name_identifier(asset_doc)
+    envs = {
+        "AVALON_PROJECT": project_name,
+        "AVALON_ASSET": folder_path,
+        "AVALON_TASK": task_name,
+        "AVALON_WORKDIR": workdir,
+    }
 
     # Update the Session and environments. Pop from environments all keys with
     # value set to None.
-    for key, value in changes.items():
-        legacy_io.Session[key] = value
+    for key, value in envs.items():
         if value is None:
             os.environ.pop(key, None)
         else:
             os.environ[key] = value
 
-    data = changes.copy()
+    data = envs.copy()
+
     # Convert env keys to human readable keys
-    data["project_name"] = legacy_io.Session["AVALON_PROJECT"]
-    data["asset_name"] = legacy_io.Session["AVALON_ASSET"]
-    data["task_name"] = legacy_io.Session["AVALON_TASK"]
+    data["project_name"] = project_name
+    data["asset_name"] = get_asset_name_identifier(asset_doc)
+    data["task_name"] = task_name
+    data["workdir_path"] = workdir
 
     # Emit session change
     emit_event("taskChanged", data)
 
-    return changes
+    return data
 
 
 def get_process_id():

client/ayon_core/pipeline/create/context.py

@@ -27,7 +27,7 @@ from ayon_core.lib.attribute_definitions import (
     get_default_values,
 )
 from ayon_core.host import IPublishHost, IWorkfileHost
-from ayon_core.pipeline import legacy_io, Anatomy
+from ayon_core.pipeline import Anatomy
 from ayon_core.pipeline.plugin_discover import DiscoverResult
 
 from .creator_plugins import (
@@ -1684,25 +1684,16 @@ class CreateContext:
         if isinstance(self.host, IWorkfileHost):
             workfile_path = self.host.get_current_workfile()
 
-        # --- TODO remove these conditions ---
-        if not project_name:
-            project_name = legacy_io.Session.get("AVALON_PROJECT")
-        if not asset_name:
-            asset_name = legacy_io.Session.get("AVALON_ASSET")
-        if not task_name:
-            task_name = legacy_io.Session.get("AVALON_TASK")
-        # ---
         return project_name, asset_name, task_name, workfile_path
 
     def reset_current_context(self):
         """Refresh current context.
 
         Reset is based on optional host implementation of `get_current_context`
-        function or using `legacy_io.Session`.
+        function.
 
         Some hosts have ability to change context file without using workfiles
-        tool but that change is not propagated to 'legacy_io.Session'
-        nor 'os.environ'.
+        tool but that change is not propagated to 'os.environ'.
 
         Todos:
             UI: Current context should be also checked on save - compare

client/ayon_core/pipeline/create/subset_name.py

@@ -2,7 +2,6 @@ import os
 
 from ayon_core.settings import get_project_settings
 from ayon_core.lib import filter_profiles, prepare_template_data
-from ayon_core.pipeline import legacy_io
 
 from .constants import DEFAULT_SUBSET_TEMPLATE
 
@@ -135,7 +134,7 @@ def get_subset_name(
     family = family.rsplit(".", 1)[-1]
 
     if project_name is None:
-        project_name = legacy_io.Session["AVALON_PROJECT"]
+        project_name = os.environ.get("AVALON_PROJECT")
 
     asset_tasks = asset_doc.get("data", {}).get("tasks") or {}
     task_info = asset_tasks.get(task_name) or {}

client/ayon_core/pipeline/load/plugins.py

@@ -2,10 +2,7 @@ import os
 import logging
 
 from ayon_core.settings import get_system_settings, get_project_settings
-from ayon_core.pipeline import (
-    schema,
-    legacy_io,
-)
+from ayon_core.pipeline import schema
 from ayon_core.pipeline.plugin_discover import (
     discover,
     register_plugin,