mirror of
https://github.com/ynput/ayon-core.git
synced 2025-12-24 21:04:40 +01:00
create function for cli in publish
This commit is contained in:
Jakub Trllo
parent
856a30cd5a
commit
f82c420fe4
3 changed files with 119 additions and 91 deletions
|
|
@ -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})
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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.")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue