fix: [SG-43950] Support component schemas in schema provisioning

python/tank/flowam/config.json

@@ -1,6 +1,6 @@
 {
   "name": "demo_pipeline",
-  "version": "1.0.2",
+  "version": "1.0.3",
   "description": "Flow Toolkit pipeline schema definitions",
   "schemas": [
     {
@@ -183,6 +183,78 @@
           "data_type": "String"
         }
       ]
+    },
+    {
+      "name": "component.layer",
+      "version": "1.0.0",
+      "display_name": "Layer",
+      "kind": "component",
+      "description": "A named compositional relationship. The target asset contributes compositionally to this asset.",
+      "inherits": [],
+      "properties": [
+        {
+          "name": "name",
+          "data_type": "String"
+        },
+        {
+          "name": "targetAsset",
+          "data_type": "$id:autodesk.me:reference-2.0.0"
+        }
+      ]
+    },
+    {
+      "name": "component.variantSet",
+      "version": "1.0.0",
+      "display_name": "Variant Set",
+      "kind": "component",
+      "description": "The target asset is a selectable realization of this asset for setName=variantName.",
+      "inherits": [],
+      "properties": [
+        {
+          "name": "setName",
+          "data_type": "String"
+        },
+        {
+          "name": "variantName",
+          "data_type": "String"
+        },
+        {
+          "name": "targetAsset",
+          "data_type": "$id:autodesk.me:reference-2.0.0"
+        },
+        {
+          "name": "displayName",
+          "data_type": "String"
+        }
+      ]
+    },
+    {
+      "name": "component.source",
+      "version": "1.0.0",
+      "display_name": "Source",
+      "kind": "component",
+      "description": "The target asset represents provenance or derivation lineage.",
+      "inherits": [],
+      "properties": [
+        {
+          "name": "targetVersion",
+          "data_type": "$id:autodesk.me:reference-2.0.0"
+        }
+      ]
+    },
+    {
+      "name": "component.reference",
+      "version": "1.0.0",
+      "display_name": "Reference",
+      "kind": "component",
+      "description": "The target asset is referenced as a dependency or included asset.",
+      "inherits": [],
+      "properties": [
+        {
+          "name": "targetVersion",
+          "data_type": "$id:autodesk.me:reference-2.0.0"
+        }
+      ]
     }
   ]
 }

python/tank/flowam/create.py

@@ -13,12 +13,15 @@
 import re
 from enum import Enum
 
+from tank_vendor.flow_data_sdk.base import model as medm_model
 from tank_vendor.flow_integration_sdk.exceptions import CreateAssetError, FlowError
 from tank_vendor.flow_integration_sdk.globals import FOLDER_TYPE_ID
 from tank_vendor.flow_integration_sdk.objects import FlowAsset, FlowProject
 from tank_vendor.flow_integration_sdk.publish import (
+    LayerComponentSpec,
     TypeComponentSpec,
     publish_new_asset,
+    publish_new_revision,
 )
 from tank_vendor.flow_integration_sdk.schema import get_schema_id
 from tank_vendor.flow_integration_sdk.utils import get_logger, trace
@@ -106,7 +109,7 @@ def get_or_create_root_folder(inputs: BaseInputs) -> FlowAsset:
         if not folder:
             logger.info(f'Creating "{SHOT_TYPE}" folder...')
             raw_asset = publish_new_asset(
-                name=ensure_unique_name(SHOT_TYPE, project),
+                name=SHOT_TYPE,
                 parent_id=project.id,
                 description="Folder for Shot assets.",
                 components=[
@@ -119,7 +122,7 @@ def get_or_create_root_folder(inputs: BaseInputs) -> FlowAsset:
         if not folder:
             logger.info(f'Creating "{ASSET_FOLDER}" folder...')
             raw_asset = publish_new_asset(
-                name=ensure_unique_name(ASSET_FOLDER, project),
+                name=ASSET_FOLDER,
                 parent_id=project.id,
                 description="Folder for Asset Build assets.",
                 components=[
@@ -132,7 +135,7 @@ def get_or_create_root_folder(inputs: BaseInputs) -> FlowAsset:
         if not folder:
             logger.info(f'Creating "{GENERIC_FOLDER}" folder...')
             raw_asset = publish_new_asset(
-                name=ensure_unique_name(GENERIC_FOLDER, project),
+                name=GENERIC_FOLDER,
                 parent_id=project.id,
                 description="Folder for Generic assets.",
                 components=[
@@ -153,70 +156,84 @@ def get_or_create_root_folder(inputs: BaseInputs) -> FlowAsset:
 def get_or_create_workfile_parent(
     root_folder: FlowAsset, inputs: BaseInputs
 ) -> FlowAsset:
-    """Determine (and create if necessary) the task-level folder that will be
-    the direct parent of the workfile asset.
+    """Determine (and create if necessary) the parent container of the workfile asset.
 
     Returns:
-        The task-folder :class:`FlowAsset`.
+        The parent asset as :class:`FlowAsset`.
     """
     logger = get_logger(__name__)
 
     sg_entity_type = inputs.sg_entity_type
     sg_entity_name = inputs.sg_entity_name
     sg_pipeline_step = inputs.sg_pipeline_step
-    sg_task_name = inputs.sg_task_name
 
     container = root_folder.find_child(sg_entity_name)
+    container_type = (
+        ASSET_CONTAINER_TYPE if sg_entity_type == ASSET_TYPE else SHOT_CONTAINER_TYPE
+    )
     if not container:
         logger.info(
             f'Creating container asset for "{sg_entity_name}" under '
             f'folder "{root_folder.name}"...'
         )
-        container_type = (
-            ASSET_CONTAINER_TYPE
-            if sg_entity_type == ASSET_TYPE
-            else SHOT_CONTAINER_TYPE
-        )
-        raw_asset = publish_new_asset(
-            name=ensure_unique_name(sg_entity_name, root_folder),
+        medm_asset = publish_new_asset(
+            name=sg_entity_name,
             parent_id=root_folder.id,
             components=[
-                TypeComponentSpec(
-                    type_id=get_schema_id(container_type), name=f"Type {container_type}"
-                )
+                TypeComponentSpec(type_id=get_schema_id(container_type), name=f"Type")
             ],
         )
-        container = FlowAsset(raw_asset)
+        container = FlowAsset(medm_asset)
 
     pipeline_step = container.find_child(sg_pipeline_step)
     if not pipeline_step:
-        logger.info(f'Creating pipeline step asset for "{sg_pipeline_step}"...')
-        raw_asset = publish_new_asset(
-            name=ensure_unique_name(sg_pipeline_step, container),
+        logger.info(f'Creating pipeline step folder for "{sg_pipeline_step}"...')
+        medm_asset = publish_new_asset(
+            name=sg_pipeline_step,
             parent_id=container.id,
-            components=[
-                TypeComponentSpec(
-                    type_id=get_schema_id(PIPELINE_STEP_TYPE),
-                    name=f"Type {PIPELINE_STEP_TYPE}",
-                )
-            ],
+            components=[TypeComponentSpec(type_id=FOLDER_TYPE_ID, name="Type")],
+        )
+        pipeline_step = FlowAsset(medm_asset)
+
+        logger.info(
+            f'Adding layer component for "{sg_pipeline_step}" on '
+            f'container "{container.name}"...'
         )
-        pipeline_step = FlowAsset(raw_asset)
-
-    task_folder = pipeline_step.find_child(sg_task_name)
-    if not task_folder:
-        logger.info(f'Creating task folder asset for "{sg_task_name}"...')
-        raw_asset = publish_new_asset(
-            name=ensure_unique_name(sg_task_name, pipeline_step),
-            parent_id=pipeline_step.id,
-            description=f'Folder for task "{sg_task_name}".',
+        publish_new_revision(
+            asset_id=container.id,
             components=[
-                TypeComponentSpec(type_id=FOLDER_TYPE_ID, name=f"Type {FOLDER_TYPE_ID}")
+                LayerComponentSpec(
+                    layer_name=sg_pipeline_step,
+                    asset_id=pipeline_step.id,
+                    display_name=sg_pipeline_step,
+                )
             ],
+            components_action=medm_model.ListAction.ADD,
         )
-        task_folder = FlowAsset(raw_asset)
+    if inputs.create_mode == CreateMode.GENERIC:
+        # Parent generic assets directly under pipeline step
+        parent = pipeline_step
+    else:
+        # For dcc assets, parent them under a root asset
+        # that will house the dcc asset as one of its "representations"
+        # NOTE: the root asset will be container type for now
+        asset_root = pipeline_step.find_child(sg_entity_name)
+        if not asset_root:
+            logger.info(f'Creating root asset for "{sg_entity_name}"...')
+            medm_asset = publish_new_asset(
+                name=sg_entity_name,
+                parent_id=pipeline_step.id,
+                description=f'Root asset for "{sg_entity_name}".',
+                components=[
+                    TypeComponentSpec(
+                        type_id=get_schema_id(container_type), name=f"Type"
+                    )
+                ],
+            )
+            asset_root = FlowAsset(medm_asset)
+        parent = asset_root
 
-    return task_folder
+    return parent
 
 
 def ensure_unique_name(name: str, parent: FlowAsset | FlowProject) -> str:

python/tank/flowam/utils.py

@@ -32,11 +32,13 @@
 from tank_vendor.flow_integration_sdk.publish import (
     ComponentSpec,
     CommentComponentSpec,
+    ReferenceComponentSpec,
     SourceComponentSpec,
     ThumbnailComponentSpec,
     TypeComponentSpec,
     FileSeqComponentSpec,
 )
+from tank_vendor.flow_integration_sdk.dependency import DependencyData, DepType
 from tank_vendor.flow_integration_sdk.objects import FlowProject
 from tank_vendor.flow_integration_sdk.schema_builder import create_pipeline_schemas
 from tank_vendor.flow_integration_sdk.utils import trace
@@ -216,6 +218,7 @@ def create_components_for_publish(
     thumbnail_path: str = "",
     comment: str = "",
     type_ids: list[str] | None = None,
+    deps: list[DependencyData] | None = None,
 ) -> list[ComponentSpec]:
     """Generate the components relevant to publish a new revision.
 
@@ -228,6 +231,11 @@ def create_components_for_publish(
         type_ids: A list of type ids to be converted into type components.
                   This is only relevant if publishing a new asset direct to remote
                   (i.e. not going through sandbox).
+        deps: Optional list of internal dependencies found in the scene.
+                  These will be recorded as ReferenceComponentSpec entries on the revision.
+
+    Raises:
+        FlowError
     """
     # Source component contains the source file
     components: list[ComponentSpec] = []
@@ -272,6 +280,28 @@ def create_components_for_publish(
         # NOTE: component names must be unique!
         type_comp = TypeComponentSpec(type_id=type_id, name=f"Type {i}")
         components.append(type_comp)
+    # Add reference components for each dependency.
+    # Callers are expected to pass only resolved asset dependencies
+    # (dep_type == ASSET with a valid version_id).
+    if deps:
+        for i, dep in enumerate(deps):
+            if dep.dep_type != DepType.ASSET:
+                raise FlowError(
+                    f"Dependency at index {i} has unexpected type "
+                    f"{dep.dep_type}. Only DepType.ASSET deps should be "
+                    "passed as reference components."
+                )
+            if not dep.version_id:
+                raise FlowError(
+                    f"Dependency at index {i} is missing a version_id. "
+                    "Cannot create a reference component without a target version."
+                )
+            components.append(
+                ReferenceComponentSpec(
+                    name=f"Reference {i}",
+                    version_id=dep.version_id,
+                )
+            )
     return components
 
 

python/tank_vendor/flow_integration_sdk/globals.py

@@ -27,28 +27,38 @@
 from .exceptions import FlowError
 from .utils import get_logger
 
-# Component type ids
+# Schema type ids
 # ------------------
 # Type ids correspond to specific MEDM schemas (and versions).
 # Schemas can be created in a hierarchical fashion using inheritance.
 # The schemas below are official Autodesk supported types that are
 # commonly relevant to asset management.
 
+BASE_COMPONENT_TYPE_ID = "autodesk.me:component-1.0.0"
+BASE_PROPERTY_TYPE_ID = "autodesk.me:property-1.0.0"
 BASE_TYPE_ID = "autodesk.me:type-1.1.0"
 BINARY_TYPE_ID = "autodesk.me:component.binary-1.0.0"
 COMMENT_TYPE_ID = "autodesk.me:component.publishComment-1.0.0"
-# NOTE: This is a temporary schema being annexed for representing derivative source
-#       which should be switched for a dedicated schema later.
-DER_SOURCE_TYPE_ID = "autodesk.me:component.dynamicPlaylistSource-1.0.0"
 FOLDER_TYPE_ID = "autodesk.me:type.folder-1.0.0"
 IMAGE_TYPE_ID = "autodesk.me:component.binary.image-1.0.0"
 
+# Maps schema kind name to its root base type ID
+KIND_BASE_TYPE_ID = {
+    "component": BASE_COMPONENT_TYPE_ID,
+    "property": BASE_PROPERTY_TYPE_ID,
+    "type": BASE_TYPE_ID,
+}
+
 # Component types
 # ---------------
 # Component base type names without full ids.
 # This should be a temporary measure, only necessary while some types
 # are not yet added to the autodesk domain, and must be created per collection.
 FILE_SEQ_TYPE = "type.fileSequence"
+REFERENCE_TYPE = "component.reference"
+DER_SOURCE_TYPE = "component.source"
+LAYER_TYPE = "component.layer"
+VARIANT_SET_TYPE = "component.variantSet"
 
 
 # Component purposes