Add single-item convenience wrappers for bulk partial-success APIs

examples/assetmanagement/assets.py

@@ -31,61 +31,56 @@
 
 client = AssetManagementClient(configuration=server_configuration)
 
-create_assets_request = [
-    CreateAssetRequest(
-        model_number=4000,
-        model_name="NI PXIe-6368",
-        serial_number="01BB877A",
-        vendor_name="NI",
-        vendor_number=4244,
-        bus_type=AssetBusType.ACCESSORY,
-        name="PCISlot2",
-        asset_type=AssetType.DEVICE_UNDER_TEST,
-        firmware_version="A1",
-        hardware_version="12A",
-        visa_resource_name="vs-3144",
+create_asset_request = CreateAssetRequest(
+    model_number=4000,
+    model_name="NI PXIe-6368",
+    serial_number="01BB877A",
+    vendor_name="NI",
+    vendor_number=4244,
+    bus_type=AssetBusType.ACCESSORY,
+    name="PCISlot2",
+    asset_type=AssetType.DEVICE_UNDER_TEST,
+    firmware_version="A1",
+    hardware_version="12A",
+    visa_resource_name="vs-3144",
+    temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
+    supports_self_calibration=True,
+    supports_external_calibration=True,
+    custom_calibration_interval=24,
+    self_calibration=SelfCalibration(
         temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
-        supports_self_calibration=True,
-        supports_external_calibration=True,
-        custom_calibration_interval=24,
-        self_calibration=SelfCalibration(
-            temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
-            is_limited=False,
-            date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
-        ),
-        is_NI_asset=True,
-        workspace=workspace_id,
-        location=AssetLocationForCreate(
-            state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT)
+        is_limited=False,
+        date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
+    ),
+    is_NI_asset=True,
+    workspace=workspace_id,
+    location=AssetLocationForCreate(
+        state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT)
+    ),
+    external_calibration=ExternalCalibration(
+        temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
+        date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
+        recommended_interval=10,
+        next_recommended_date=datetime(
+            2023, 11, 14, 20, 42, 11, 583000, tzinfo=timezone.utc
         ),
-        external_calibration=ExternalCalibration(
-            temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
-            date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
-            recommended_interval=10,
-            next_recommended_date=datetime(
-                2023, 11, 14, 20, 42, 11, 583000, tzinfo=timezone.utc
-            ),
-            next_custom_due_date=datetime(
-                2024, 11, 14, 20, 42, 11, 583000, tzinfo=timezone.utc
-            ),
-            resolved_due_date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
+        next_custom_due_date=datetime(
+            2024, 11, 14, 20, 42, 11, 583000, tzinfo=timezone.utc
         ),
-        properties={"Key1": "Value1"},
-        keywords=["Keyword1"],
-        discovery_type=AssetDiscoveryType.MANUAL,
-        file_ids=["608a5684800e325b48837c2a"],
-        supports_self_test=True,
-        supports_reset=True,
-        part_number="A1234 B5",
-    )
-]
+        resolved_due_date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
+    ),
+    properties={"Key1": "Value1"},
+    keywords=["Keyword1"],
+    discovery_type=AssetDiscoveryType.MANUAL,
+    file_ids=["608a5684800e325b48837c2a"],
+    supports_self_test=True,
+    supports_reset=True,
+    part_number="A1234 B5",
+)
 
 # Create an asset.
-create_assets_response = client.create_assets(assets=create_assets_request)
-
-created_asset_id = None
-if create_assets_response.assets and len(create_assets_response.assets) > 0:
-    created_asset_id = str(create_assets_response.assets[0].id)
+created_asset = client.create_asset(asset=create_asset_request)
+created_asset_id = str(created_asset.id)
 
 # Query assets using id.
 query_asset_request = QueryAssetsRequest(

nisystemlink/clients/assetmanagement/_asset_management_client.py

@@ -13,6 +13,7 @@
 from uplink import Field, Path, retry
 
 from . import models
+from ..core.helpers._partial_success import unwrap_single_item_partial_success
 
 
 @retry(
@@ -56,6 +57,30 @@ def create_assets(
         """
         ...
 
+    def create_asset(self, asset: models.CreateAssetRequest) -> models.Asset:
+        """Create a single asset.
+
+        Args:
+            asset: The asset to create.
+
+        Returns:
+            The created asset.
+
+        Raises:
+            ApiException: if the asset could not be created or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.create_assets([asset])
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.assets,
+            failed=response.failed,
+            error=response.error,
+            failure_message="Failed to create asset.",
+            empty_message="Server returned no created assets.",
+        )
+
     @post("query-assets")
     def __query_assets(
         self, query: models._QueryAssetsRequest

nisystemlink/clients/core/helpers/_partial_success.py

@@ -0,0 +1,65 @@
+from typing import Any, Sequence, TypeVar
+
+from nisystemlink.clients import core
+
+_ItemT = TypeVar("_ItemT")
+_ONE_OR_MORE_ERRORS_OCCURRED_NAME = "Skyline.OneOrMoreErrorsOccurred"
+_ONE_OR_MORE_ERRORS_OCCURRED_CODE = -251041
+
+
+def _unwrap_single_inner_error(error: core.ApiError | None) -> core.ApiError | None:
+    if error is None:
+        return None
+
+    if len(error.inner_errors) != 1:
+        return error
+
+    if (
+        error.name == _ONE_OR_MORE_ERRORS_OCCURRED_NAME
+        or error.code == _ONE_OR_MORE_ERRORS_OCCURRED_CODE
+    ):
+        return error.inner_errors[0]
+
+    return error
+
+
+def unwrap_single_item_partial_success(
+    *,
+    response: Any | None,
+    items: Sequence[_ItemT] | None,
+    failed: Sequence[Any] | None,
+    error: core.ApiError | None,
+    failure_message: str,
+    empty_message: str,
+) -> _ItemT:
+    """Return the first successful item from a partial-success response.
+
+    Raises:
+        ApiException: if the response reports a failure or contains no successful item.
+    """
+    response_data = (
+        response.model_dump(mode="json", by_alias=True)
+        if response is not None
+        else None
+    )
+
+    if failed or error:
+        raise core.ApiException(
+            failure_message,
+            error=_unwrap_single_inner_error(error),
+            response_data=response_data,
+        )
+
+    if not items:
+        raise core.ApiException(
+            empty_message,
+            response_data=response_data,
+        )
+
+    if len(items) != 1:
+        raise core.ApiException(
+            f"Expected exactly one successful item but received {len(items)}.",
+            response_data=response_data,
+        )
+
+    return items[0]

nisystemlink/clients/product/_product_client.py

@@ -8,6 +8,7 @@
 from uplink import Field, Query, retry, returns
 
 from . import models
+from ..core.helpers._partial_success import unwrap_single_item_partial_success
 
 
 @retry(
@@ -50,6 +51,30 @@ def create_products(
         """
         ...
 
+    def create_product(self, product: models.CreateProductRequest) -> models.Product:
+        """Creates a single product.
+
+        Args:
+            product: The product to create.
+
+        Returns:
+            The created product.
+
+        Raises:
+            ApiException: if the product could not be created or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.create_products([product])
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.products,
+            failed=response.failed,
+            error=response.error,
+            failure_message="Failed to create product.",
+            empty_message="Server returned no created products.",
+        )
+
     @get(
         "products",
         args=[Query("continuationToken"), Query("take"), Query("returnCount")],
@@ -153,6 +178,33 @@ def update_products(
         """
         ...
 
+    def update_product(
+        self, product: models.UpdateProductRequest, replace: bool = False
+    ) -> models.Product:
+        """Updates a single product.
+
+        Args:
+            product: The product to update.
+            replace: Replace the existing fields instead of merging them.
+
+        Returns:
+            The updated product.
+
+        Raises:
+            ApiException: if the product could not be updated or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.update_products([product], replace=replace)
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.products,
+            failed=response.failed,
+            error=response.error,
+            failure_message="Failed to update product.",
+            empty_message="Server returned no updated products.",
+        )
+
     @delete("products/{id}")
     def delete_product(self, id: str) -> None:
         """Deletes a single product by id.

nisystemlink/clients/spec/_spec_client.py

@@ -8,6 +8,7 @@
 from uplink import Field, retry
 
 from . import models
+from ..core.helpers._partial_success import unwrap_single_item_partial_success
 
 
 @retry(
@@ -63,6 +64,32 @@ def create_specs(
         """
         ...
 
+    def create_spec(
+        self, spec: models.CreateSpecificationsRequestObject
+    ) -> models.CreatedSpecification:
+        """Creates a single specification.
+
+        Args:
+            spec: The specification to create.
+
+        Returns:
+            The created specification.
+
+        Raises:
+            ApiException: if the specification could not be created or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.create_specs(models.CreateSpecificationsRequest(specs=[spec]))
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.created_specs,
+            failed=response.failed_specs,
+            error=response.error,
+            failure_message="Failed to create spec.",
+            empty_message="Server returned no created specs.",
+        )
+
     @post("delete-specs", args=[Field("ids")])
     def delete_specs(
         self, ids: List[str]
@@ -129,3 +156,29 @@ def update_specs(
             with error messages for updates that failed.
         """
         ...
+
+    def update_spec(
+        self, spec: models.UpdateSpecificationsRequestObject
+    ) -> models.UpdatedSpecification:
+        """Updates a single specification.
+
+        Args:
+            spec: The specification to update.
+
+        Returns:
+            The updated specification.
+
+        Raises:
+            ApiException: if the specification could not be updated or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.update_specs(models.UpdateSpecificationsRequest(specs=[spec]))
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.updated_specs if response is not None else None,
+            failed=response.failed_specs if response is not None else None,
+            error=response.error if response is not None else None,
+            failure_message="Failed to update spec.",
+            empty_message="Server returned no updated specs.",
+        )