From f82c420fe499bfd168d0ee3e773bf2dc064c17ca Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Thu, 25 Jul 2024 15:54:42 +0200 Subject: [PATCH] create function for cli in publish --- client/ayon_core/cli.py | 92 +------------- client/ayon_core/pipeline/publish/__init__.py | 4 + client/ayon_core/pipeline/publish/lib.py | 114 +++++++++++++++++- 3 files changed, 119 insertions(+), 91 deletions(-) diff --git a/client/ayon_core/cli.py b/client/ayon_core/cli.py index c1b5e5d5fc..db6674d88f 100644 --- a/client/ayon_core/cli.py +++ b/client/ayon_core/cli.py @@ -149,97 +149,9 @@ def publish(ctx, path, targets): Publish collects json from path provided as an argument. """ - import ayon_api - import pyblish.util + from ayon_core.pipeline.publish import main_cli_publish - from ayon_core.pipeline import ( - install_ayon_plugins, - get_global_context, - ) - - # Register target and host - if not isinstance(path, str): - raise RuntimeError("Path to JSON must be a string.") - - # Fix older jobs - for src_key, dst_key in ( - ("AVALON_PROJECT", "AYON_PROJECT_NAME"), - ("AVALON_ASSET", "AYON_FOLDER_PATH"), - ("AVALON_TASK", "AYON_TASK_NAME"), - ("AVALON_WORKDIR", "AYON_WORKDIR"), - ("AVALON_APP_NAME", "AYON_APP_NAME"), - ("AVALON_APP", "AYON_HOST_NAME"), - ): - if src_key in os.environ and dst_key not in os.environ: - os.environ[dst_key] = os.environ[src_key] - # Remove old keys, so we're sure they're not used - os.environ.pop(src_key, None) - - log = Logger.get_logger("CLI-publish") - - # Make public ayon api behave as other user - # - this works only if public ayon api is using service user - username = os.environ.get("AYON_USERNAME") - if username: - # NOTE: ayon-python-api does not have public api function to find - # out if is used service user. So we need to have try > except - # block. - con = ayon_api.get_server_api_connection() - try: - con.set_default_service_username(username) - except ValueError: - pass - - install_ayon_plugins() - - addons_manager = ctx.obj["addons_manager"] - - # TODO validate if this has to happen - # - it should happen during 'install_ayon_plugins' - publish_paths = addons_manager.collect_plugin_paths()["publish"] - for plugin_path in publish_paths: - pyblish.api.register_plugin_path(plugin_path) - - applications_addon = addons_manager.get_enabled_addon("applications") - if applications_addon is not None: - context = get_global_context() - env = applications_addon.get_farm_publish_environment_variables( - context["project_name"], - context["folder_path"], - context["task_name"], - ) - os.environ.update(env) - - pyblish.api.register_host("shell") - - if targets: - for target in targets: - print(f"setting target: {target}") - pyblish.api.register_target(target) - else: - pyblish.api.register_target("farm") - - os.environ["AYON_PUBLISH_DATA"] = path - os.environ["HEADLESS_PUBLISH"] = 'true' # to use in app lib - - log.info("Running publish ...") - - plugins = pyblish.api.discover() - print("Using plugins:") - for plugin in plugins: - print(plugin) - - # Error exit as soon as any error occurs. - error_format = ("Failed {plugin.__name__}: " - "{error} -- {error.traceback}") - - for result in pyblish.util.publish_iter(): - if result["error"]: - log.error(error_format.format(**result)) - # uninstall() - sys.exit(1) - - log.info("Publish finished.") + main_cli_publish(path, targets, ctx.obj["addons_manager"]) @main_cli.command(context_settings={"ignore_unknown_options": True}) diff --git a/client/ayon_core/pipeline/publish/__init__.py b/client/ayon_core/pipeline/publish/__init__.py index d507972664..ab19b6e360 100644 --- a/client/ayon_core/pipeline/publish/__init__.py +++ b/client/ayon_core/pipeline/publish/__init__.py @@ -42,6 +42,8 @@ from .lib import ( get_plugin_settings, get_publish_instance_label, get_publish_instance_families, + + main_cli_publish, ) from .abstract_expected_files import ExpectedFiles @@ -92,6 +94,8 @@ __all__ = ( "get_publish_instance_label", "get_publish_instance_families", + "main_cli_publish", + "ExpectedFiles", "RenderInstance", diff --git a/client/ayon_core/pipeline/publish/lib.py b/client/ayon_core/pipeline/publish/lib.py index c4e7b2a42c..8b82622e4c 100644 --- a/client/ayon_core/pipeline/publish/lib.py +++ b/client/ayon_core/pipeline/publish/lib.py @@ -4,8 +4,9 @@ import inspect import copy import tempfile import xml.etree.ElementTree -from typing import Optional, Union +from typing import Optional, Union, List +import ayon_api import pyblish.util import pyblish.plugin import pyblish.api @@ -16,6 +17,7 @@ from ayon_core.lib import ( filter_profiles, ) from ayon_core.settings import get_project_settings +from ayon_core.addon import AddonsManager from ayon_core.pipeline import ( tempdir, Anatomy @@ -978,3 +980,113 @@ def get_instance_expected_output_path( path_template_obj = anatomy.get_template_item("publish", "default")["path"] template_filled = path_template_obj.format_strict(template_data) return os.path.normpath(template_filled) + + +def main_cli_publish( + path: str, + targets: Optional[List[str]] = None, + addons_manager: Optional[AddonsManager] = None, +): + """Start headless publishing. + + Publish use json from passed path argument. + + Args: + path (str): Path to JSON. + targets (Optional[List[str]]): List of pyblish targets. + addons_manager (Optional[AddonsManager]): Addons manager instance. + + Raises: + RuntimeError: When there is no path to process or when executed with + list of JSON paths. + + """ + from ayon_core.pipeline import ( + install_ayon_plugins, + get_global_context, + ) + + # Register target and host + if not isinstance(path, str): + raise RuntimeError("Path to JSON must be a string.") + + # Fix older jobs + for src_key, dst_key in ( + ("AVALON_PROJECT", "AYON_PROJECT_NAME"), + ("AVALON_ASSET", "AYON_FOLDER_PATH"), + ("AVALON_TASK", "AYON_TASK_NAME"), + ("AVALON_WORKDIR", "AYON_WORKDIR"), + ("AVALON_APP_NAME", "AYON_APP_NAME"), + ("AVALON_APP", "AYON_HOST_NAME"), + ): + if src_key in os.environ and dst_key not in os.environ: + os.environ[dst_key] = os.environ[src_key] + # Remove old keys, so we're sure they're not used + os.environ.pop(src_key, None) + + log = Logger.get_logger("CLI-publish") + + # Make public ayon api behave as other user + # - this works only if public ayon api is using service user + username = os.environ.get("AYON_USERNAME") + if username: + # NOTE: ayon-python-api does not have public api function to find + # out if is used service user. So we need to have try > except + # block. + con = ayon_api.get_server_api_connection() + try: + con.set_default_service_username(username) + except ValueError: + pass + + install_ayon_plugins() + + if addons_manager is None: + addons_manager = AddonsManager() + + # TODO validate if this has to happen + # - it should happen during 'install_ayon_plugins' + publish_paths = addons_manager.collect_plugin_paths()["publish"] + for plugin_path in publish_paths: + pyblish.api.register_plugin_path(plugin_path) + + applications_addon = addons_manager.get_enabled_addon("applications") + if applications_addon is not None: + context = get_global_context() + env = applications_addon.get_farm_publish_environment_variables( + context["project_name"], + context["folder_path"], + context["task_name"], + ) + os.environ.update(env) + + pyblish.api.register_host("shell") + + if targets: + for target in targets: + print(f"setting target: {target}") + pyblish.api.register_target(target) + else: + pyblish.api.register_target("farm") + + os.environ["AYON_PUBLISH_DATA"] = path + os.environ["HEADLESS_PUBLISH"] = 'true' # to use in app lib + + log.info("Running publish ...") + + plugins = pyblish.api.discover() + print("Using plugins:") + for plugin in plugins: + print(plugin) + + # Error exit as soon as any error occurs. + error_format = ("Failed {plugin.__name__}: " + "{error} -- {error.traceback}") + + for result in pyblish.util.publish_iter(): + if result["error"]: + log.error(error_format.format(**result)) + # uninstall() + sys.exit(1) + + log.info("Publish finished.")