Publishing rendered products to AYON

client/ayon_deadline_cloud/addon.py

@@ -2,19 +2,26 @@
 from __future__ import annotations
 
 import os
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
-from ayon_core.addon import AYONAddon, IPluginPaths
+import click
+from ayon_core.addon import AYONAddon, IPluginPaths, click_wrap
+
+from ayon_deadline_cloud.api.publish import publish_content
 
 from .version import __version__
 
+if TYPE_CHECKING:
+    from logging import Logger
+
 DEADLINE_CLOUD_ADDON_ROOT = os.path.dirname(os.path.abspath(__file__))
 
 
 class DeadlineCloudAddon(AYONAddon, IPluginPaths):
     """Deadline Cloud Addon for AYON."""
     name = "deadline_cloud"
     version = __version__
+    log: Logger
 
     @staticmethod
     def add_implementation_envs(
@@ -32,9 +39,9 @@ def add_implementation_envs(
         # Deadline Cloud configuration file path for the client to use.
         # env["DEADLINE_CONFIG_FILE_PATH"] = ...
 
-    @staticmethod
-    def get_publish_plugin_paths(
-         host_name: Optional[str] = None  # noqa: ARG004
+    def get_publish_plugin_paths(  # noqa: PLR6301
+            self,
+            host_name: str,
     ) -> list[str]:  # ty:ignore[invalid-method-override]
         """Return list of paths to publish plugins.
 
@@ -49,10 +56,10 @@ def get_publish_plugin_paths(
         return [os.path.join(
             DEADLINE_CLOUD_ADDON_ROOT, "plugins", "publish")]
 
-    @staticmethod
-    def get_create_plugin_paths(
-         host_name: Optional[str] = None
-    ) -> list[str]:  # ty:ignore[invalid-method-override]
+    def get_create_plugin_paths(  # noqa: PLR6301
+            self,
+            host_name: str,
+    ) -> list[str]:
         """Return list of paths to creator plugins.
 
         Args:
@@ -67,17 +74,123 @@ def get_create_plugin_paths(
             DEADLINE_CLOUD_ADDON_ROOT,
             "plugins", "create", host_name or "global")]
 
-    @staticmethod
-    def get_launch_hook_paths(app: str) -> list[str]:  # noqa: ARG004
-        """Return list of paths to launch hooks.
+    def _publish(  # noqa: PLR0913, PLR0917
+            self,
+            path: str,
+            folder_path: str,
+            project_name: str,
+            user_name: str,
+            product_base_type: str,
+            task_name: Optional[str],
+            host_name: Optional[str],
+            source_file: Optional[str],
+    ) -> None:
+        """Publish the result of a Deadline Cloud processed job.
 
         Args:
-            app: Name of the host application
-                to get specific hook paths.
+            path: Path to the folder containing the job
+                result to publish.
+            folder_path: Folder path for the context on AYON.
+            project_name: Name of the project associated with the job result.
+            user_name: Name of the user who submitted the job.
+            product_base_type: Base name of the product to publish.
+                Note that currently it is overridden down the line
+                with hardcoded `render` - in the future, product base type
+                should be passed correctly to support other publish
+                types.
+            task_name: Optional name of the task associated with
+                the job result.
+            host_name: Optional name of the host application associated with
+                the job result.
+            source_file: Optional path to a source file related to the job
+                result, which might be used for validation or as part of
+                the publishing process.
 
-        Returns:
-            List of paths to launch hooks.
+        """
+        # TODO(antirotor): Implement this method to trigger publishing of the
+        # result of a Deadline Cloud processed job. This might involve
+        # collecting the output files from the job, validating them, and then
+        # moving them to their final destination or registering them in AYON.
+        self.log.debug(
+            "publish called with arguments: "
+            "folder_path=%s, project_name=%s, "
+            "user_name=%s, product_base_name=%s, task_name=%s, "
+            "host_name=%s, source_file=%s",
+            folder_path, project_name, user_name,
+            product_base_type, task_name, host_name, source_file
+        )
+
+        publish_content(
+            path=path,
+            project_name=project_name,
+            folder_path=folder_path,
+            user_name=user_name,
+            task_name=task_name,
+            host_name=host_name,
+            source_file=source_file,
+        )
+
+    def _cli_main(self) -> None:
+        """Add CLI commands to this addon."""
+
+    def cli(self, addon_click_group: click.Group) -> None:
+        """CLI interface.
+
+        Args:
+            addon_click_group: Click group to add commands to.
 
         """
-        return [os.path.join(
-            DEADLINE_CLOUD_ADDON_ROOT, "hooks")]
+        cli_main = click_wrap.group(
+            self._cli_main,
+            name=self.name,
+            help="Deadline Cloud commands",
+        )
+
+        cli_main.command(
+            self._publish,
+            name="publish",
+            help=(
+                "Publish the result of a Deadline Cloud "
+                "processed job."
+            ),
+        ).option(
+            "-f",
+            "--folder-path",
+            type=click.STRING,
+            required=True,
+        ).option(
+            "-t",
+            "--task-name",
+            type=click.STRING,
+            required=False,
+        ).option(
+            "-p",
+            "--project-name",
+            type=click.STRING,
+            required=True,
+        ).option(
+            "-u",
+            "--user-name",
+            type=click.STRING,
+            required=True,
+        ).option(
+            "--host-name",
+            type=click.STRING,
+            required=False,
+        ).option(
+            "--product-base-type",
+            type=click.STRING,
+            required=True
+        ).option(
+            "-s",
+            "--source-file",
+            type=click.Path(exists=True, file_okay=True, dir_okay=False),
+        )
+
+        cli_main.argument(
+            "path",
+            nargs=1,
+            type=click.Path(exists=True, file_okay=False, dir_okay=True),
+        )
+
+        addon_click_group.add_command(cli_main.to_click_obj())

client/ayon_deadline_cloud/api/datatypes.py

@@ -0,0 +1,35 @@
+"""Data classes."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Union
+
+if TYPE_CHECKING:
+    from ayon_core.pipeline.traits import Representation as TraitRepresentation
+
+
+@dataclass
+class StandardRepresentation:
+    """Representation dataclass."""
+    name: str
+    ext: str
+    files: Union[str, list[str]]
+    stagingDir: str
+
+
+@dataclass
+class InstanceData:
+    """Instance dataclass."""
+    publish: bool
+    active: bool
+    label: str
+    name: str
+    productName: str
+    productBaseType: str
+    family: str
+    families: list[str]
+    folderPath: str
+    task: str
+    variant: str
+    standard_representations: list[StandardRepresentation]
+    trait_representations: list[TraitRepresentation]

client/ayon_deadline_cloud/api/publish.py

@@ -0,0 +1,102 @@
+"""Publishing Deadline Cloud Jobs."""
+from __future__ import annotations
+
+import contextlib
+from typing import Optional
+
+import ayon_api
+import pyblish.api
+import pyblish.util
+from ayon_core.pipeline import install_ayon_plugins
+from ayon_core.pipeline.publish import publish_plugins_discover
+
+
+def publish_content(  # noqa: PLR0913, PLR0917
+        path: str,
+        project_name: str,
+        folder_path: str,
+        user_name: str,
+        task_name: Optional[str] = None,
+        host_name: Optional[str] = None,
+        source_file: Optional[str] = None,
+    ) -> None:
+    """Publish content.
+
+    This function bootstraps pyblish to run, collect the
+    rendered files and publish them using appropriate pipeline.
+
+    Args:
+        path: Path to the folder where the content is to be published.
+        project_name: Name of the project.
+        folder_path: Path to the folder where the content is to be published.
+        user_name: Name of the user who submitted the job.
+        task_name: Name of the task to be published.
+        host_name: Name of the host to be published.
+        source_file: Path to the source file.
+
+    Raises:
+        ValueError: If the folder or task does not exist.
+        RuntimeError: If the publishing process fails.
+
+    """
+    # Make public ayon api behave as other user
+    # - this works only if public ayon api is using service user:
+    # ayon-python-api does not have public api function to find
+    # out if is used service user. So we need to have try-except.
+    con = ayon_api.get_server_api_connection()
+    with contextlib.suppress(ValueError):
+        con.set_default_service_username(user_name)
+
+    # check if folder exists
+    folder_entity = ayon_api.get_folder_by_path(
+        project_name=project_name,
+        folder_path=folder_path,
+    )
+    if not folder_entity:
+        msg = (
+            f"Unable to find folder '{folder_path}' in "
+            f"project '{project_name}'."
+        )
+        raise ValueError(msg)
+
+    # check if task exists
+    if task_name:
+        task_entity = ayon_api.get_task_by_name(
+            project_name=project_name,
+            folder_id=folder_entity["id"],
+            task_name=task_name,
+        )
+        if not task_entity:
+            msg = (
+                f"Unable to find task '{task_name}' in "
+                f"folder '{folder_path}' in project '{project_name}'."
+            )
+            raise ValueError(msg)
+
+    pyblish_context = pyblish.api.Context()
+    pyblish_context.data["hostName"] = "workflow"
+    pyblish_context.data["projectName"] = project_name
+    pyblish_context.data["folderPath"] = folder_path
+    pyblish_context.data["outputPath"] = path
+
+    if task_name:
+        pyblish_context.data["taskName"] = task_name
+
+    if host_name:
+        pyblish_context.data["hostName"] = host_name
+
+    if source_file:
+        pyblish_context.data["sourceFile"] = source_file
+
+    pyblish.api.register_host("shell")
+
+    install_ayon_plugins()
+    discover_result = publish_plugins_discover()
+    publish_plugins = discover_result.plugins
+
+    for result in pyblish.util.publish_iter(
+            context=pyblish_context,
+            plugins=publish_plugins,
+    ):
+        if result["error"]:
+            raise RuntimeError(repr(result))