I used this script to test batch publishing in PR https://github.com/BigRoy/ayon-core/pull/6 which extends PR https://github.com/ynput/ayon-core/pull/542

batch_publish_houdini.py

"""
Each publish record has only one product type but can include multiple representations.
e.g. a 'model' product type can include ('.abc', '.fbx', '.bgeo') representations. 
which are the same data but saved in different formats.

Some Notes about Houdini dynamic creator:
    Dynamic creator is accessed via code only.

    Dynamic creator computes the representations on instance creation.
    'CreateRuntimeInstance.create' expects some data to exist in order to compute the representations.
    
    Dynamic creator shouldn't trigger any publish plugins. 
    Maybe we can add one validator that checks if files exists on disk or not.
    However, Dynamic instances still triggers instances plugins with families ["*"] and also context plugins.
    
    It works in a similar manner to tray publisher except it runs on data deduced from Houdini nodes instead of file paths. 
    It's the developer responsibility to write code that deduce the data from Houdini nodes.
    This comes from the concept that we want Dynamic creator to be DCC agnostic therefore we can reuse it almost as it is.
    

Random Notes about this script:
    This Script expects all files exist on disk.

    This script can be refactored to be PDG node(s).

    example_representation = {
                "name": ext,
                "ext": ext,
                "files": output,
                "stagingDir": staging_dir,
                "frameStart": frame_start_handle,
                "frameEnd": frame_end_handle,
                "tags": [review],  # render/review
                "preview": True,   # render/review
                "camera_name": camera_name   # render/review
            }


    'output_paths' key is used to evaluate exported files.
        It can be a list of output paths or a dictionary of AOVs.

    For render product: 
        If there's one AOV, it'll be considered beauty.
        If there are more than one AOV, then multipartExr will be set to False.
        'mark_as_reviewable' key when true:
            sets preview to true for beauty aov only.
            adds ["review"] to tags inside representation of beauty aov only.

    Both 'cache_products' and 'render_products' should be deduced from Houdini nodes.
        However, Vanilla Houdini nodes can't fill in these data:
            
            folder path -> we can get it from the current context
            task name -> we can get it from the current context
            product type -> can be a drop down menu in the Houdini parameter.
            variant -> this one can be the node name.
            product name -> can be computed by 'ayon_core.pipeline.create.get_product_name'.
"""
from ayon_core.pipeline import registered_host
from ayon_core.pipeline.create import CreateContext
import pyblish.api
import pyblish.util

# Product to ingest
cache_products = [
    {
        "product_type": "camera",
        "variant": "my_camera",
        "output_paths": ["$HIP/export/camera.abc"],
        "frameStart": 1001,
        "frameEnd": 1001
    },
    {
        "product_type": "pointcache",
        "variant": "my_pointcache",
        "output_paths": [
            # "$HIP/geo/$HIPNAME.geometry.$F.bgeo.sc",  # This doesn't work with 'create_file_list' function
            "$HIP/geo/geometry.$F.bgeo.sc", 
            "$HIP/geo/output.abc"],
        "frameStart": 1001,
        "frameEnd": 1005
    },
    {
        "product_type": "review",
        "variant": "my_review",
        "output_paths": ["$HIP/render/opengl.$F4.exr"],
        "frameStart": 1001,
        "frameEnd": 1005
    }
]

# Render to ingest
# We create an instance for each aov.
render_products = [
    {
        "product_type": "render",
        "variant": "my_render",
        "output_paths": {
            "beauty": "$HIP/render/mantra.$F4.exr",
            "N": "$HIP/render/mantra.N.$F4.exr",
            "Pz": "$HIP/render/mantra.Pz.$F4.exr"
        },
        "frameStart": 1001,
        "frameEnd": 1005,
        "mark_as_reviewable": True
    }
]

host = registered_host()
assert host, "No registered host."
    
create_context = CreateContext(host)

# Deactivate all instances
# TODO: save active value and use it after resetting the context.
for instance in create_context.instances:
    instance["active"] = False
    print(f"- {instance.label} ({instance.product_type})")

create_context.save_changes()

# I'll use the current context for this example.
project_name = host.get_current_project_name()
folder_path = host.get_current_folder_path()
task_name = host.get_current_task_name()

# Use a dedicated Creator class for dynamic instances
creator_identifier = "io.openpype.creators.houdini.batch"
batch_creator = create_context.creators.get(creator_identifier)

# Prepare instance data and prepare representations.
for product in cache_products:
    instance_data = {
        "project": project_name,
        "folderPath": folder_path,
        "task": task_name,
        
        "product_type": product["product_type"],
        "variant": product["variant"],
        "frameStart": product["frameStart"],
        "frameEnd": product["frameEnd"],
        
    }
    pre_create_data = {
        "output_paths": product["output_paths"]
    }

    instance = batch_creator.create(
        product["variant"],  # pass variant as product name. it'll be overridden anyways.
        instance_data,
        pre_create_data
    )


error_format = "Failed {plugin.__name__}: {error} -- {error.traceback}"
pyblish_context = pyblish.api.Context()
pyblish_context.data["create_context"] = create_context
pyblish_plugins = create_context.publish_plugins

for result in pyblish.util.publish_iter(pyblish_context, pyblish_plugins):
    for record in result["records"]:
        print("{}: {}".format(result["plugin"].label, record.msg))

    # Exit as soon as any error occurs.
    if result["error"]:
        error_message = error_format.format(**result)
        print(error_message)
        raise RuntimeError ("Error occurred.")
    
if not list(pyblish_context):
    raise RuntimeError ("No resulting instances in the context. Assuming publish failed.")

published_versions = []
for instance in pyblish_context:
    version_entity = instance.data.get("versionEntity")
    if version_entity:
        print(f"Instance '{instance}' published version: {version_entity}")
        published_versions.append(version_entity)

if not published_versions:
    raise RuntimeError ("Publish failed.")

print("Publish succeeded.")

Some random thoughts I need to expand on these further.

References:

• Old PR with some good comments: ynput/ayon-core#542
• Fabia's demos: ynput/ayon-core#542 (comment)

Step 1: Design a high-level API interface

# The system should end up allowing us to do AYON publishing in a simple way WITH type hinting across the board.
context = ayon.CreateContext()
instance = context.create(
    variant="Main",
    product_type="texture",
    files=["/path/to/texture.exr"],
    traits=[
        FrameRange(start=1001, end=1010, frame_list=[1001, 1005]),
        OCIO(path="ocio.config", colorspace="ACEScg"),
        BurninMetadata(camera_name="XYZ", username="roy"),
        TagsMetadata(),
    ]
)
context.publish()

Expose the API by designing an abstraction over our current system

This allows us to make an MVP and start confirming the API is nice-to-use.
It allows us to play with it and confirm our thoughts.

# Expose an abstraction over 

def create(context, variant, product_type, traits):
   """Make compatible API against current CreateContext API"""

# Backward compatible transition in pyblish

     instance.data["traits"]

    if isinstance(trait, FrameRange):
        trait.

        instance.data["frameStart"] = trait.start

     instance.data["frameStart"] =