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

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,47 @@
+from typing import Any, Sequence, TypeVar
+
+from nisystemlink.clients import core
+
+_ItemT = TypeVar("_ItemT")
+
+
+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=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.",
+        )

nisystemlink/clients/test_plan/_test_plan_client.py

@@ -8,6 +8,8 @@
 from nisystemlink.clients.test_plan import models
 from uplink import Field, retry
 
+from ..core.helpers._partial_success import unwrap_single_item_partial_success
+
 
 @retry(
     when=retry.when.status(408, 429, 502, 503, 504),
@@ -66,6 +68,32 @@ def create_test_plans(
         """
         ...
 
+    def create_test_plan(
+        self, test_plan: models.CreateTestPlanRequest
+    ) -> models.TestPlan:
+        """Create a single test plan.
+
+        Args:
+            test_plan: The test plan to create.
+
+        Returns:
+            The created test plan.
+
+        Raises:
+            ApiException: if the test plan could not be created or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.create_test_plans([test_plan])
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.created_test_plans,
+            failed=response.failed_test_plans,
+            error=response.error,
+            failure_message="Failed to create test plan.",
+            empty_message="Server returned no created test plans.",
+        )
+
     @post("delete-testplans", args=[Field("ids")])
     def delete_test_plans(self, ids: List[str]) -> None:
         """Delete test plans by IDs.
@@ -106,6 +134,37 @@ def schedule_test_plans(
         """
         ...
 
+    def schedule_test_plan(
+        self,
+        test_plan: models.ScheduleTestPlanRequest,
+        replace: bool | None = None,
+    ) -> models.TestPlan:
+        """Schedule a single test plan.
+
+        Args:
+            test_plan: The test plan schedule request.
+            replace: Whether to replace the existing scheduled test plan.
+
+        Returns:
+            The scheduled test plan.
+
+        Raises:
+            ApiException: if the test plan could not be scheduled or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.schedule_test_plans(
+            models.ScheduleTestPlansRequest(test_plans=[test_plan], replace=replace)
+        )
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.scheduled_test_plans,
+            failed=response.failed_test_plans,
+            error=response.error,
+            failure_message="Failed to schedule test plan.",
+            empty_message="Server returned no scheduled test plans.",
+        )
+
     @post("update-testplans")
     def update_test_plans(
         self, update_request: models.UpdateTestPlansRequest
@@ -120,6 +179,35 @@ def update_test_plans(
         """
         ...
 
+    def update_test_plan(
+        self, test_plan: models.UpdateTestPlanRequest, replace: bool | None = None
+    ) -> models.TestPlan:
+        """Update a single test plan.
+
+        Args:
+            test_plan: The test plan to update.
+            replace: Whether to replace the existing test plan instead of merging updates.
+
+        Returns:
+            The updated test plan.
+
+        Raises:
+            ApiException: if the test plan could not be updated or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.update_test_plans(
+            models.UpdateTestPlansRequest(test_plans=[test_plan], replace=replace)
+        )
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.updated_test_plans,
+            failed=response.failed_test_plans,
+            error=response.error,
+            failure_message="Failed to update test plan.",
+            empty_message="Server returned no updated test plans.",
+        )
+
     @post("testplan-templates", args=[Field("testPlanTemplates")])
     def create_test_plan_templates(
         self, test_plan_templates: List[models.CreateTestPlanTemplateRequest]
@@ -137,6 +225,32 @@ def create_test_plan_templates(
         """
         ...
 
+    def create_test_plan_template(
+        self, test_plan_template: models.CreateTestPlanTemplateRequest
+    ) -> models.TestPlanTemplate:
+        """Creates a single test plan template.
+
+        Args:
+            test_plan_template: The test plan template to create.
+
+        Returns:
+            The created test plan template.
+
+        Raises:
+            ApiException: if the test plan template could not be created or the service returns an
+                unexpected partial-success payload.
+        """
+        response = self.create_test_plan_templates([test_plan_template])
+
+        return unwrap_single_item_partial_success(
+            response=response,
+            items=response.created_test_plan_templates,
+            failed=response.failed_test_plan_templates,
+            error=response.error,
+            failure_message="Failed to create test plan template.",
+            empty_message="Server returned no created test plan templates.",
+        )
+
     @post("query-testplan-templates")
     def query_test_plan_templates(
         self, query_test_plan_templates: models.QueryTestPlanTemplatesRequest