Links: Add bulk creation of links

ayon_api/__init__.py

@@ -275,6 +275,7 @@
     delete_link_type,
     make_sure_link_type_exists,
     create_link,
+    create_links,
     delete_link,
     get_entities_links,
     get_folders_links,
@@ -594,6 +595,7 @@
     "delete_link_type",
     "make_sure_link_type_exists",
     "create_link",
+    "create_links",
     "delete_link",
     "get_entities_links",
     "get_folders_links",

ayon_api/_api.py

@@ -53,6 +53,7 @@
         BackgroundOperationTask,
         LinkDirection,
         CreateLinkData,
+        CreateLinkResponseData,
         EventFilter,
         EventStatus,
         EnrollEventData,
@@ -7504,7 +7505,7 @@ def create_link(
     output_type: str,
     link_name: Optional[str] = None,
     data: Optional[dict[str, Any]] = None,
-) -> CreateLinkData:
+) -> CreateLinkResponseData:
     """Create link between 2 entities.
 
     Link has a type which must already exists on a project.
@@ -7527,7 +7528,7 @@ def create_link(
             with the link.
 
     Returns:
-        CreateLinkData: Information about link.
+        CreateLinkResponseData: Information about link.
 
     Raises:
         HTTPRequestError: Server error happened.
@@ -7546,6 +7547,38 @@ def create_link(
     )
 
 
+def create_links(
+    project_name: str,
+    links: list[CreateLinkData],
+) -> None:
+    """Create multiple links in a single request.
+
+    Example of link data::
+        [
+            {
+                "input": "59a212c0d2e211eda0e20242ac120001",
+                "output": "59a212c0d2e211eda0e20242ac120002",
+                "linkType": "reference|folder|folder",
+                "name": "my_link",
+                "data": {"key": "value"}
+            }
+        ]
+
+    Args:
+        project_name (str): Project where links are created.
+        links (list[CreateLinkData]): List of link data.
+
+    Raises:
+        ValueError: Link data is invalid.
+
+    """
+    con = get_server_api_connection()
+    return con.create_links(
+        project_name=project_name,
+        links=links,
+    )
+
+
 def delete_link(
     project_name: str,
     link_id: str,

ayon_api/_api_helpers/links.py

@@ -15,7 +15,11 @@
 from .base import BaseServerAPI
 
 if typing.TYPE_CHECKING:
-    from ayon_api.typing import LinkDirection, CreateLinkData
+    from ayon_api.typing import (
+        LinkDirection,
+        CreateLinkData,
+        CreateLinkResponseData,
+    )
 
 
 class LinksAPI(BaseServerAPI):
@@ -200,7 +204,7 @@ def create_link(
         output_type: str,
         link_name: Optional[str] = None,
         data: Optional[dict[str, Any]] = None,
-    ) -> CreateLinkData:
+    ) -> CreateLinkResponseData:
         """Create link between 2 entities.
 
         Link has a type which must already exists on a project.
@@ -223,7 +227,7 @@ def create_link(
                 with the link.
 
         Returns:
-            CreateLinkData: Information about link.
+            CreateLinkResponseData: Information about link.
 
         Raises:
             HTTPRequestError: Server error happened.
@@ -249,6 +253,59 @@ def create_link(
         response.raise_for_status()
         return response.data
 
+    def create_links(
+        self,
+        project_name: str,
+        links: list[CreateLinkData],
+    ) -> None:
+        """Create multiple links in a single request.
+
+        Example of link data::
+            [
+                {
+                    "input": "59a212c0d2e211eda0e20242ac120001",
+                    "output": "59a212c0d2e211eda0e20242ac120002",
+                    "linkType": "reference|folder|folder",
+                    "name": "my_link",
+                    "data": {"key": "value"}
+                }
+            ]
+
+        Args:
+            project_name (str): Project where links are created.
+            links (list[CreateLinkData]): List of link data.
+
+        Raises:
+            ValueError: Link data is invalid.
+
+        """
+        if not links:
+            return
+
+        for link in links:
+            self._validate_link_data(link)
+
+        if self.get_server_version_tuple() < (1, 15, 8):
+            for link in links:
+                link_type, in_type, out_type = link["linkType"].split("|")
+                self.create_link(
+                    project_name,
+                    link_type,
+                    link["input"],
+                    in_type,
+                    link["output"],
+                    out_type,
+                    link_name=link.get("name") or None,
+                    data=link.get("data") or None,
+                )
+            return
+
+        response = self.post(
+            f"projects/{project_name}/links/bulk",
+            links=links
+        )
+        response.raise_for_status()
+
     def delete_link(self, project_name: str, link_id: str) -> None:
         """Remove link by id.
 
@@ -619,6 +676,35 @@ def get_representation_links(
             project_name, [representation_id], link_types, link_direction
         )[representation_id]
 
+    def _validate_link_data(self, link_data: dict[str, Any]) -> None:
+        """Validate link data before sending to server.
+
+        Args:
+            link_data (dict[str, Any]): Link data to validate.
+
+        Raises:
+            ValueError: Link data is invalid.
+
+        """
+        required_keys = {"input", "output", "linkType"}
+        missing_keys = required_keys - link_data.keys()
+        if missing_keys:
+            mk = ", ".join(f"'{key}'" for key in sorted(missing_keys))
+            raise ValueError(f"Missing required keys in link data: {mk}")
+
+        link_type = link_data["linkType"]
+        if not isinstance(link_type, str):
+            raise ValueError(
+                f"Invalid linkType type: {type(link_type)}. Expected 'str'"
+            )
+
+        link_type_parts = link_type.split("|")
+        if len(link_type_parts) != 3:
+            raise ValueError(
+                f"Invalid linkType format: {link_type}. Expected format:"
+                " 'link_type|input_type|output_type'"
+            )
+
     def _prepare_link_filters(
         self,
         filters: dict[str, Any],

ayon_api/typing.py

@@ -127,10 +127,18 @@ class BackgroundOperationTask(TypedDict):
 LinkDirection = Literal["in", "out"]
 
 
-class CreateLinkData(TypedDict):
+class CreateLinkResponseData(TypedDict):
     id: str
 
 
+class CreateLinkData(TypedDict):
+    input: str
+    output: str
+    linkType: str
+    data: NotRequired[dict[str, Any] | None]
+    name: NotRequired[str | None]
+
+
 class AttributeEnumItemDict(TypedDict):
     value: str | int | float | bool
     label: str