diff --git a/client/ayon_core/cli.py b/client/ayon_core/cli.py index acc7dfb6d4..b7dad94346 100644 --- a/client/ayon_core/cli.py +++ b/client/ayon_core/cli.py @@ -19,7 +19,6 @@ from ayon_core.lib import ( Logger, ) -from .cli_commands import Commands class AliasedGroup(click.Group): @@ -152,7 +151,103 @@ def publish(ctx, path, targets, gui): Publish collects json from path provided as an argument. """ - Commands.publish(path, targets, gui, ctx.obj["addons_manager"]) + import ayon_api + import pyblish.util + + 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) + + if gui: + from ayon_core.tools.utils.host_tools import show_publish + from ayon_core.tools.utils.lib import qt_app_context + with qt_app_context(): + show_publish() + else: + # 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.command(context_settings={"ignore_unknown_options": True}) diff --git a/client/ayon_core/cli_commands.py b/client/ayon_core/cli_commands.py deleted file mode 100644 index 085dd5bb04..0000000000 --- a/client/ayon_core/cli_commands.py +++ /dev/null @@ -1,137 +0,0 @@ -# -*- coding: utf-8 -*- -"""Implementation of AYON commands.""" -import os -import sys -from typing import Optional, List - -from ayon_core.addon import AddonsManager - - -class Commands: - """Class implementing commands used by AYON. - - Most of its methods are called by :mod:`cli` module. - """ - @staticmethod - def publish( - path: str, - targets: Optional[List[str]] = None, - gui: Optional[bool] = False, - addons_manager: Optional[AddonsManager] = None, - ) -> 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. - gui (Optional[bool]): Show publish UI. - addons_manager (Optional[AddonsManager]): Addons manager instance. - - Raises: - RuntimeError: When there is no path to process. - RuntimeError: When executed with list of JSON paths. - - """ - from ayon_core.lib import Logger - - from ayon_core.addon import AddonsManager - from ayon_core.pipeline import ( - install_ayon_plugins, - get_global_context, - ) - - import ayon_api - import pyblish.util - - # 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() - - 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) - - if gui: - from ayon_core.tools.utils.host_tools import show_publish - from ayon_core.tools.utils.lib import qt_app_context - with qt_app_context(): - show_publish() - else: - # 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.") -