From 045fdce481b32a97954fc39312c16e82e008574d Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Fri, 3 Jul 2026 17:49:16 +0200 Subject: [PATCH] added bulk creation of links --- ayon_api/__init__.py | 2 + ayon_api/_api.py | 37 +++++++++++++++- ayon_api/_api_helpers/links.py | 80 +++++++++++++++++++++++++++++++++- ayon_api/typing.py | 10 ++++- 4 files changed, 124 insertions(+), 5 deletions(-) diff --git a/ayon_api/__init__.py b/ayon_api/__init__.py index 2f04c2800..aef55d7e4 100644 --- a/ayon_api/__init__.py +++ b/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", diff --git a/ayon_api/_api.py b/ayon_api/_api.py index b0c6e02dd..08ed230ff 100644 --- a/ayon_api/_api.py +++ b/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[dict[str, Any]], +) -> 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[dict[str, Any]]): 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, diff --git a/ayon_api/_api_helpers/links.py b/ayon_api/_api_helpers/links.py index a50b21817..a83e1f44d 100644 --- a/ayon_api/_api_helpers/links.py +++ b/ayon_api/_api_helpers/links.py @@ -200,7 +200,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 +223,7 @@ def create_link( with the link. Returns: - CreateLinkData: Information about link. + CreateLinkResponseData: Information about link. Raises: HTTPRequestError: Server error happened. @@ -249,6 +249,59 @@ def create_link( response.raise_for_status() return response.data + def create_links( + self, + project_name: str, + links: list[dict[str, Any]], + ) -> 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[dict[str, Any]]): 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 +672,29 @@ 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 missing_keys)) + raise ValueError(f"Missing required keys in link data {mk}") + + link_type_parts = link_data["linkType"].split("|") + if len(link_type_parts) != 3: + raise ValueError( + f"Invalid linkType format: {link_data['linkType']}. " + "Expected format: 'link_type|input_type|output_type'" + ) + def _prepare_link_filters( self, filters: dict[str, Any], diff --git a/ayon_api/typing.py b/ayon_api/typing.py index 37bb92f51..368d522ca 100644 --- a/ayon_api/typing.py +++ b/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: dict[str, Any] | None = None + name: str | None = None + + class AttributeEnumItemDict(TypedDict): value: str | int | float | bool label: str