feat: Add link files API route in Asset Management Client (#141)

Co-authored-by: Suganth Sadaiyappan <[email protected]>

Author: SSadaiyappan-NI

docs/api_reference/assetmanagement.rst

@@ -10,6 +10,7 @@ nisystemlink.clients.assetmanagement
    .. automethod:: create_assets
    .. automethod:: query_assets
    .. automethod:: delete_assets
+   .. automethod:: link_files
 
 .. automodule:: nisystemlink.clients.assetmanagement.models
    :members:

docs/getting_started.rst

@@ -336,12 +336,12 @@ default connection. The default connection depends on your environment.
 
 With a :class:`.AssetManagementClient` object, you can:
 
-* Create, delete and get the list of assets.
+* Create, delete, get the list of assets and link files to assets.
 
 Examples
 ~~~~~~~~
 
-create, delete and query asset
+create, delete, query asset and link files to assets.
 
 .. literalinclude:: ../examples/assetmanagement/assets.py
    :language: python

examples/assetmanagement/assets.py

@@ -84,6 +84,13 @@
 )
 client.query_assets(query=query_asset_request)
 
+# Link files to the created asset.
+file_ids = ["sameple-file-id"]
+if created_asset_id:
+    link_files_response = client.link_files(
+        asset_id=created_asset_id, file_ids=file_ids
+    )
+
 # Delete the created asset.
 if created_asset_id is not None:
     client.delete_assets(ids=[created_asset_id])

nisystemlink/clients/assetmanagement/_asset_management_client.py

@@ -6,7 +6,7 @@
 from nisystemlink.clients.core._http_configuration import HttpConfiguration
 from nisystemlink.clients.core._uplink._base_client import BaseClient
 from nisystemlink.clients.core._uplink._methods import post
-from uplink import Field, retry
+from uplink import Field, Path, retry
 
 from . import models
 
@@ -122,3 +122,22 @@ def delete_assets(self, ids: List[str]) -> models.DeleteAssetsResponse:
             ApiException: If unable to communicate with the asset management service or if there are invalid arguments.
         """
         ...
+
+    @post("assets/{assetId}/file", args=[Path("assetId"), Field("fileIds")])
+    def link_files(
+        self, asset_id: str, file_ids: List[str]
+    ) -> Optional[models.LinkFilesPartialSuccessResponse]:
+        """Link files to an asset.
+
+        Args:
+            asset_id: The ID of the asset to which files should be linked.
+            file_ids: The list of file IDs to link.
+
+        Returns:
+            None if all link files succeed, otherwise a response containing the IDs of files that were
+            successfully linked and those that failed.
+
+        Raises:
+            ApiException: If unable to communicate with the asset management service or if there are invalid arguments.
+        """
+        ...

nisystemlink/clients/assetmanagement/models/__init__.py

@@ -20,5 +20,6 @@
     ExternalCalibration,
 )
 from ._asset_types import AssetBusType, AssetDiscoveryType, AssetType
+from ._link_files_partial_success_response import LinkFilesPartialSuccessResponse
 
 # flake8: noqa

nisystemlink/clients/assetmanagement/models/_link_files_partial_success_response.py

@@ -0,0 +1,17 @@
+from typing import List, Optional
+
+from nisystemlink.clients.core._api_error import ApiError
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+
+
+class LinkFilesPartialSuccessResponse(JsonModel):
+    """Model for a Link Files Partial Success Response."""
+
+    succeeded: Optional[List[str]] = None
+    """Gets or sets array of file IDs that were successfully linked to assets."""
+
+    failed: Optional[List[str]] = None
+    """Gets or sets array of file IDs that failed to link to assets."""
+
+    error: Optional[ApiError] = None
+    """Gets or sets error if the link file operation failed."""

tests/integration/assetmanagement/test_asset_management.py

@@ -123,6 +123,25 @@ def test__create_asset__returns_created_asset(
         asset_id = assets[0].id
         assert asset_id is not None
 
+    def test__link_files__link_succeded(
+        self, client: AssetManagementClient, create_asset
+    ):
+        self._create_assets_request[0].model_number = 101
+        create_response = create_asset(self._create_assets_request)
+
+        asset_id = (
+            create_response.assets[0].id
+            if create_response.assets and create_response.assets[0].id
+            else None
+        )
+
+        assert asset_id is not None
+
+        file_ids = ["608a5684800e325b48837c2a"]
+        link_files_response = client.link_files(asset_id=asset_id, file_ids=file_ids)
+
+        assert link_files_response is None
+
     def test__delete_asset__returns_deleted_asset(self, client: AssetManagementClient):
         self._create_assets_request[0].model_number = 102
         create_response: CreateAssetsPartialSuccessResponse = client.create_assets(