AY-971 Custom staging dir functionality migration

client/ayon_core/pipeline/__init__.py

@@ -7,6 +7,10 @@
 
 from .anatomy import Anatomy
 
+from .tempdir import get_temp_dir
+
+from .staging_dir import get_staging_dir_info
+
 from .create import (
     BaseCreator,
     Creator,
@@ -117,6 +121,12 @@
     # --- Anatomy ---
     "Anatomy",
 
+    # --- Temp dir ---
+    "get_temp_dir",
+
+    # --- Staging dir ---
+    "get_staging_dir_info",
+
     # --- Create ---
     "BaseCreator",
     "Creator",

client/ayon_core/pipeline/create/creator_plugins.py

@@ -1,19 +1,21 @@
 # -*- coding: utf-8 -*-
+import os
 import copy
 import collections
 from typing import TYPE_CHECKING, Optional, Dict, Any
 
 from abc import ABC, abstractmethod
 
 from ayon_core.settings import get_project_settings
-from ayon_core.lib import Logger
+from ayon_core.lib import Logger, get_version_from_path
 from ayon_core.pipeline.plugin_discover import (
     discover,
     register_plugin,
     register_plugin_path,
     deregister_plugin,
     deregister_plugin_path
 )
+from ayon_core.pipeline import get_staging_dir_info
 
 from .constants import DEFAULT_VARIANT_VALUE
 from .product_name import get_product_name
@@ -831,6 +833,95 @@ def get_pre_create_attr_defs(self):
         """
         return self.pre_create_attr_defs
 
+    def get_staging_dir(self, instance):
+        """Return the staging dir and persistence from instance.
+
+        Args:
+            instance (CreatedInstance): Instance for which should be staging
+                dir gathered.
+
+        Returns:
+            Optional[namedtuple]: Staging dir path and persistence or None
+        """
+        create_ctx = self.create_context
+        product_name = instance.get("productName")
+        product_type = instance.get("productType")
+        folder_path = instance.get("folderPath")
+
+        # this can only work if product name and folder path are available
+        if not product_name or not folder_path:
+            return None
+
+        publish_settings = self.project_settings["core"]["publish"]
+        follow_workfile_version = (
+            publish_settings
+            ["CollectAnatomyInstanceData"]
+            ["follow_workfile_version"]
+        )
+
+        # Gather version number provided from the instance.
+        version = instance.get("version")
+
+        # If follow workfile, gather version from workfile path.
+        if version is None and follow_workfile_version:
+            current_workfile = self.create_context.get_current_workfile_path()
+            workfile_version = get_version_from_path(current_workfile)
+            version = int(workfile_version)
+
+        # Fill-up version with next version available.
+        elif version is None:
+            versions = self.get_next_versions_for_instances(
+                [instance]
+            )
+            version, = tuple(versions.values())
+
+        template_data = {"version": version}
+
+        staging_dir_info = get_staging_dir_info(
+            create_ctx.get_current_project_entity(),
+            create_ctx.get_folder_entity(folder_path),
+            create_ctx.get_task_entity(folder_path, instance.get("task")),
+            product_type,
+            product_name,
+            create_ctx.host_name,
+            anatomy=create_ctx.get_current_project_anatomy(),
+            project_settings=create_ctx.get_current_project_settings(),
+            always_return_path=False,
+            logger=self.log,
+            template_data=template_data,
+        )
+
+        return staging_dir_info or None
+
+    def apply_staging_dir(self, instance):
+        """Apply staging dir with persistence to instance's transient data.
+
+        Method is called on instance creation and on instance update.
+
+        Args:
+            instance (CreatedInstance): Instance for which should be staging
+                dir applied.
+
+        Returns:
+            Optional[str]: Staging dir path or None if not applied.
+        """
+        staging_dir_info = self.get_staging_dir(instance)
+        if staging_dir_info is None:
+            return None
+
+        # path might be already created by get_staging_dir_info
+        staging_dir_path = staging_dir_info.directory
+        os.makedirs(staging_dir_path, exist_ok=True)
+
+        instance.transient_data.update({
+            "stagingDir": staging_dir_path,
+            "stagingDir_persistent": staging_dir_info.persistent,
+        })
+
+        self.log.info(f"Applied staging dir to instance: {staging_dir_path}")
+
+        return staging_dir_path
+
     def _pre_create_attr_defs_changed(self):
         """Called when pre-create attribute definitions change.
 

client/ayon_core/pipeline/publish/constants.py

@@ -8,6 +8,5 @@
 
 DEFAULT_PUBLISH_TEMPLATE = "default"
 DEFAULT_HERO_PUBLISH_TEMPLATE = "default"
-TRANSIENT_DIR_TEMPLATE = "default"
 
 FARM_JOB_ENV_DATA_KEY: str = "farmJobEnv"

client/ayon_core/pipeline/publish/lib.py

@@ -2,7 +2,7 @@
 import sys
 import inspect
 import copy
-import tempfile
+import warnings
 import xml.etree.ElementTree
 from typing import Optional, Union, List
 
@@ -18,15 +18,11 @@
 )
 from ayon_core.settings import get_project_settings
 from ayon_core.addon import AddonsManager
-from ayon_core.pipeline import (
-    tempdir,
-    Anatomy
-)
+from ayon_core.pipeline import get_staging_dir_info
 from ayon_core.pipeline.plugin_discover import DiscoverResult
 from .constants import (
     DEFAULT_PUBLISH_TEMPLATE,
     DEFAULT_HERO_PUBLISH_TEMPLATE,
-    TRANSIENT_DIR_TEMPLATE
 )
 
 
@@ -581,58 +577,6 @@ def context_plugin_should_run(plugin, context):
     return False
 
 
-def get_instance_staging_dir(instance):
-    """Unified way how staging dir is stored and created on instances.
-
-    First check if 'stagingDir' is already set in instance data.
-    In case there already is new tempdir will not be created.
-
-    It also supports `AYON_TMPDIR`, so studio can define own temp
-    shared repository per project or even per more granular context.
-    Template formatting is supported also with optional keys. Folder is
-    created in case it doesn't exists.
-
-    Available anatomy formatting keys:
-        - root[work | <root name key>]
-        - project[name | code]
-
-    Note:
-        Staging dir does not have to be necessarily in tempdir so be careful
-        about its usage.
-
-    Args:
-        instance (pyblish.lib.Instance): Instance for which we want to get
-            staging dir.
-
-    Returns:
-        str: Path to staging dir of instance.
-    """
-    staging_dir = instance.data.get('stagingDir')
-    if staging_dir:
-        return staging_dir
-
-    anatomy = instance.context.data.get("anatomy")
-
-    # get customized tempdir path from `AYON_TMPDIR` env var
-    custom_temp_dir = tempdir.create_custom_tempdir(
-        anatomy.project_name, anatomy)
-
-    if custom_temp_dir:
-        staging_dir = os.path.normpath(
-            tempfile.mkdtemp(
-                prefix="pyblish_tmp_",
-                dir=custom_temp_dir
-            )
-        )
-    else:
-        staging_dir = os.path.normpath(
-            tempfile.mkdtemp(prefix="pyblish_tmp_")
-        )
-    instance.data['stagingDir'] = staging_dir
-
-    return staging_dir
-
-
 def get_publish_repre_path(instance, repre, only_published=False):
     """Get representation path that can be used for integration.
 
@@ -685,6 +629,8 @@ def get_publish_repre_path(instance, repre, only_published=False):
     return None
 
 
+# deprecated: backward compatibility only (2024-09-12)
+# TODO: remove in the future
 def get_custom_staging_dir_info(
     project_name,
     host_name,
@@ -694,67 +640,86 @@ def get_custom_staging_dir_info(
     product_name,
     project_settings=None,
     anatomy=None,
-    log=None
+    log=None,
 ):
-    """Checks profiles if context should use special custom dir as staging.
+    from ayon_core.pipeline.staging_dir import get_staging_dir_config
+    warnings.warn(
+        (
+            "Function 'get_custom_staging_dir_info' in"
+            " 'ayon_core.pipeline.publish' is deprecated. Please use"
+            " 'get_custom_staging_dir_info'"
+            " in 'ayon_core.pipeline.stagingdir'."
+        ),
+        DeprecationWarning,
+    )
+    tr_data = get_staging_dir_config(
+        project_name,
+        task_type,
+        task_name,
+        product_type,
+        product_name,
+        host_name,
+        project_settings=project_settings,
+        anatomy=anatomy,
+        log=log,
+    )
 
-    Args:
-        project_name (str)
-        host_name (str)
-        product_type (str)
-        task_name (str)
-        task_type (str)
-        product_name (str)
-        project_settings(Dict[str, Any]): Prepared project settings.
-        anatomy (Dict[str, Any])
-        log (Logger) (optional)
+    if not tr_data:
+        return None, None
+
+    return tr_data["template"], tr_data["persistence"]
+
+
+def get_instance_staging_dir(instance):
+    """Unified way how staging dir is stored and created on instances.
+
+    First check if 'stagingDir' is already set in instance data.
+    In case there already is new tempdir will not be created.
 
     Returns:
-        (tuple)
-    Raises:
-        ValueError - if misconfigured template should be used
+        str: Path to staging dir
     """
-    settings = project_settings or get_project_settings(project_name)
-    custom_staging_dir_profiles = (settings["core"]
-                                           ["tools"]
-                                           ["publish"]
-                                           ["custom_staging_dir_profiles"])
-    if not custom_staging_dir_profiles:
-        return None, None
+    staging_dir = instance.data.get("stagingDir")
 
-    if not log:
-        log = Logger.get_logger("get_custom_staging_dir_info")
+    if staging_dir:
+        return staging_dir
 
-    filtering_criteria = {
-        "hosts": host_name,
-        "families": product_type,
-        "task_names": task_name,
-        "task_types": task_type,
-        "subsets": product_name
-    }
-    profile = filter_profiles(custom_staging_dir_profiles,
-                              filtering_criteria,
-                              logger=log)
+    anatomy_data = instance.data["anatomyData"]
+    template_data = copy.deepcopy(anatomy_data)
 
-    if not profile or not profile["active"]:
-        return None, None
+    # context data based variables
+    context = instance.context
 
-    if not anatomy:
-        anatomy = Anatomy(project_name)
+    # add current file as workfile name into formatting data
+    current_file = context.data.get("currentFile")
+    if current_file:
+        workfile = os.path.basename(current_file)
+        workfile_name, _ = os.path.splitext(workfile)
+        template_data["workfile_name"] = workfile_name
+
+    staging_dir_info = get_staging_dir_info(
+        context.data["projectEntity"],
+        instance.data.get("folderEntity"),
+        instance.data.get("taskEntity"),
+        instance.data["productType"],
+        instance.data["productName"],
+        context.data["hostName"],
+        anatomy=context.data["anatomy"],
+        project_settings=context.data["project_settings"],
+        template_data=template_data,
+        always_return_path=True,
+    )
 
-    template_name = profile["template_name"] or TRANSIENT_DIR_TEMPLATE
+    staging_dir_path = staging_dir_info.directory
 
-    custom_staging_dir = anatomy.get_template_item(
-        "staging", template_name, "directory", default=None
-    )
-    if custom_staging_dir is None:
-        raise ValueError((
-            "Anatomy of project \"{}\" does not have set"
-            " \"{}\" template key!"
-        ).format(project_name, template_name))
-    is_persistent = profile["custom_staging_dir_persistent"]
+    # path might be already created by get_staging_dir_info
+    os.makedirs(staging_dir_path, exist_ok=True)
+    instance.data.update({
+        "stagingDir": staging_dir_path,
+        "stagingDir_persistent": staging_dir_info.persistent,
+    })
 
-    return custom_staging_dir.template, is_persistent
+    return staging_dir_path
 
 
 def get_published_workfile_instance(context):