feat: Add client for SystemLink results API

docs/api_reference.rst

@@ -11,6 +11,7 @@ API Reference
    api_reference/tag
    api_reference/product
    api_reference/testmonitor
+   api_reference/result
    api_reference/dataframe
    api_reference/spec
    api_reference/file

docs/api_reference/result.rst

@@ -0,0 +1,23 @@
+.. _api_tag_page:
+
+nisystemlink.clients.result
+======================
+
+.. autoclass:: nisystemlink.clients.result.ResultClient
+   :exclude-members: __init__
+
+   .. automethod:: __init__
+   .. automethod:: create_results
+   .. automethod:: get_results
+   .. automethod:: query_results
+   .. automethod:: query_result_values
+   .. automethod:: update_results
+   .. automethod:: delete_result
+   .. automethod:: delete_results
+
+.. automodule:: nisystemlink.clients.result.models
+   :members:
+   :imported-members:
+
+.. automodule:: nisystemlink.clients.result.utilities
+   :members:

docs/getting_started.rst

@@ -253,3 +253,29 @@ Delete a feed.
 .. literalinclude:: ../examples/feeds/delete_feed.py
    :language: python
    :linenos:
+
+Result API
+-------
+
+Overview
+~~~~~~~~
+
+The :class:`.ResultClient` class is the primary entry point of the Result API.
+
+When constructing a :class:`.ResultClient`, you can pass an
+:class:`.HttpConfiguration` (like one retrieved from the
+:class:`.HttpConfigurationManager`), or let :class:`.ResultClient` use the
+default connection. The default connection depends on your environment.
+
+With a :class:`.ResultClient` object, you can:
+
+* Create, update, query, and delete results
+
+Examples
+~~~~~~~~
+
+Create, query, update, and delete some results
+
+.. literalinclude:: ../examples/result/results.py
+   :language: python
+   :linenos:

examples/result/results.py

@@ -0,0 +1,86 @@
+from nisystemlink.clients.core import HttpConfiguration
+from nisystemlink.clients.result import ResultClient
+from nisystemlink.clients.result.models import (
+    QueryResultsRequest,
+    QueryResultValuesRequest,
+    Result,
+    ResultField,
+    StatusObject,
+    StatusType,
+)
+
+program_name = "Example Name"
+host_name = "Example Host"
+status_type = StatusType.PASSED
+
+
+def create_some_results():
+    """Create two example results on your server."""
+    new_results = [
+        Result(
+            part_number="Example 123 AA",
+            program_name=program_name,
+            host_name=host_name,
+            status=StatusObject(status_type=StatusType.PASSED),
+            keywords=["original keyword"],
+            properties={"original property key": "yes"},
+        ),
+        Result(
+            part_number="Example 123 AA1",
+            program_name=program_name,
+            host_name=host_name,
+            status=StatusObject(status_type=StatusType.CUSTOM, status_name="Custom"),
+            keywords=["original keyword"],
+            properties={"original property key": "original"},
+        ),
+    ]
+    create_response = client.create_results(new_results)
+    return create_response
+
+
+# Setup the server configuration to point to your instance of SystemLink Enterprise
+server_configuration = HttpConfiguration(
+    server_uri="https://yourserver.yourcompany.com",
+    api_key="YourAPIKeyGeneratedFromSystemLink",
+)
+client = ResultClient(configuration=server_configuration)
+
+create_response = create_some_results()
+
+# Get all the results using the continuation token in batches of 100 at a time.
+response = client.get_results_paged(take=100, return_count=True)
+all_results = response.results
+while response.continuation_token:
+    response = client.get_results_paged(
+        take=100, continuation_token=response.continuation_token, return_count=True
+    )
+    all_results.extend(response.results)
+
+# use get for first result created
+created_result = client.get_result(create_response.results[0].id)
+
+# Query results without continuation
+query_request = QueryResultsRequest(
+    filter=f'status.statusType="{status_type}"', return_count=True
+)
+response = client.query_results_paged(query_request)
+
+# Update the first result that you just created and replace the keywords
+updated_result = create_response.results[0]
+updated_result.keywords = ["new keyword"]
+updated_result.properties = {"new property key": "new value"}
+update_response = client.update_results([create_response.results[0]], replace=True)
+
+# Query for just the ids of results that match the family
+values_query = QueryResultValuesRequest(
+    filter=f'programName="{program_name}"', field=ResultField.ID
+)
+values_response = client.query_result_values(query=values_query)
+
+# delete each created result individually by id
+for result in create_response.results:
+    client.delete_result(result.id)
+
+# Create some more and delete them with a single call to delete.
+create_response = create_some_results()
+client.delete_results([result.id for result in create_response.results])

nisystemlink/clients/result/__init__.py

@@ -0,0 +1,3 @@
+from ._result_client import ResultClient
+
+# flake8: noqa

nisystemlink/clients/result/_result_client.py

@@ -0,0 +1,214 @@
+"""Implementations of the Result Client."""
+
+from typing import List, Optional
+
+from nisystemlink.clients import core
+from nisystemlink.clients.core._uplink._base_client import BaseClient
+from nisystemlink.clients.core._uplink._methods import delete, get, post
+from nisystemlink.clients.result.models import (
+    Result,
+    UpdateResultRequest,
+)
+from uplink import Field, Query, retry, returns
+
+from . import models
+
+
+@retry(when=retry.when.status([429, 503, 504]), stop=retry.stop.after_attempt(5))
+class ResultClient(BaseClient):
+
+    def __init__(self, configuration: Optional[core.HttpConfiguration] = None):
+        """Initialize an instance.
+
+        Args:
+            configuration: Defines the web server to connect to and information about
+                how to connect. If not provided, the
+                :class:`HttpConfigurationManager <nisystemlink.clients.core.HttpConfigurationManager>`
+                is used to obtain the configuration.
+
+        Raises:
+            ApiException: if unable to communicate with the Result Service.
+        """
+        if configuration is None:
+            configuration = core.HttpConfigurationManager.get_configuration()
+        super().__init__(configuration, base_path="/nitestmonitor/v2/")
+
+    @post("results", args=[Field("results")])
+    def create_results(self, results: List[Result]) -> models.ResultsPartialSuccess:
+        """Creates one or more results and returns errors for failed creations.
+
+        Args:
+            results: A list of results to attempt to create.
+
+        Returns: A list of created results, results that failed to create, and errors for
+            failures.
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` service of provided invalid
+                arguments.
+        """
+        ...
+
+    @get(
+        "results",
+        args=[Query("continuationToken"), Query("take"), Query("returnCount")],
+    )
+    def get_results_paged(
+        self,
+        continuation_token: Optional[str] = None,
+        take: Optional[int] = None,
+        return_count: Optional[bool] = None,
+    ) -> models.PagedResults:
+        """Reads a list of results.
+
+        Args:
+            continuation_token: The token used to paginate results.
+            take: The number of results to get in this request.
+            return_count: Whether or not to return the total number of results available.
+
+        Returns:
+            A list of results.
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` Service
+                or provided an invalid argument.
+        """
+        ...
+
+    @get("results/{id}")
+    def get_result(self, id: str) -> models.Result:
+        """Retrieves a single result by id.
+
+        Args:
+            id (str): Unique ID of a result.
+
+        Returns:
+            The single result matching `id`
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` Service
+                or provided an invalid argument.
+        """
+        ...
+
+    @post("query-results")
+    def query_results_paged(
+        self, query: models.QueryResultsRequest
+    ) -> models.PagedResults:
+        """Queries for results that match the filter.
+
+        Args:
+            query : The query contains a DynamicLINQ query string in addition to other details
+                about how to filter and return the list of results.
+
+        Returns:
+            A paged list of results with a continuation token to get the next page.
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` Service or provided invalid
+                arguments.
+        """
+        ...
+
+    @returns.json  # type: ignore
+    @post("query-result-values")
+    def query_result_values(self, query: models.QueryResultValuesRequest) -> List[str]:
+        """Queries for results that match the query and returns a list of the requested field.
+
+        Args:
+            query : The query for the fields.
+
+        Returns:
+            A list of the values of the queried field.
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` Service or provided
+                invalid arguments.
+        """
+        ...
+
+    @post("update-results", args=[Field("results"), Field("replace")])
+    def __update_results(
+        self, results: List[UpdateResultRequest], replace: bool = False
+    ) -> models.ResultsPartialSuccess:
+        """Updates a list of results with optional field replacement.
+
+        Args:
+            `results`: A list of results to update. Results are matched for update by id.
+            `replace`: Replace the existing fields instead of merging them. Defaults to `False`.
+                If this is `True`, then `keywords` and `properties` for the result will be
+                    replaced by what is in the `results` provided in this request.
+                If this is `False`, then the `keywords` and `properties` in this request will
+                    merge with what is already present in the server resource.
+
+        Returns: A list of updates results, results that failed to update, and errors for
+            failures.
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` Service
+                or provided an invalid argument.
+        """
+        ...
+
+    def update_results(
+        self,
+        results: List[UpdateResultRequest],
+        replace: bool = False,
+        allow_workspace_update: bool = False,
+    ) -> models.ResultsPartialSuccess:
+        """Updates a list of results with optional field replacement.
+
+        Args:
+            `results`: A list of results to update. Results are matched for update by id.
+            `replace`: Replace the existing fields instead of merging them. Defaults to `False`.
+                If this is `True`, then `keywords` and `properties` for the result will be
+                    replaced by what is in the `results` provided in this request.
+                If this is `False`, then the `keywords` and `properties` in this request will
+                    merge with what is already present in the server resource.
+            `allow_workspace_update`: If this is set to `False`, the `workspace` field will be set to None
+                before updating.
+
+        Returns: A list of updates results, results that failed to update, and errors for
+            failures.
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` Service
+                or provided an invalid argument.
+        """
+        if not allow_workspace_update:
+            for result in results:
+                result.workspace = None
+
+        return self.__update_results(results, replace)
+
+    @delete("results/{id}")
+    def delete_result(self, id: str) -> None:
+        """Deletes a single result by id.
+
+        Args:
+            id (str): Unique ID of a result.
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` Service
+                or provided an invalid argument.
+        """
+        ...
+
+    @post("delete-results", args=[Field("ids")])
+    def delete_results(
+        self, ids: List[str]
+    ) -> Optional[models.DeleteResultsPartialSuccess]:
+        """Deletes multiple results.
+
+        Args:
+            ids (List[str]): List of unique IDs of results.
+
+        Returns:
+            A partial success if any results failed to delete, or None if all
+            results were deleted successfully.
+
+        Raises:
+            ApiException: if unable to communicate with the ``/nitestmonitor`` Service
+                or provided an invalid argument.
+        """
+        ...

nisystemlink/clients/result/models/__init__.py

@@ -0,0 +1,12 @@
+from ._result import Result, StatusObject, StatusType
+from ._results_partial_success import ResultsPartialSuccess
+from ._delete_results_partial_success import DeleteResultsPartialSuccess
+from ._paged_results import PagedResults
+from ._update_result_request import UpdateResultRequest
+from ._query_results_request import (
+    QueryResultsRequest,
+    ResultField,
+    QueryResultValuesRequest,
+)
+
+# flake8: noqa