"""Implementation of SpecClient"""
from typing import List
from nisystemlink.clients import core
from nisystemlink.clients.core._uplink._base_client import BaseClient
from nisystemlink.clients.core._uplink._methods import get, post
from uplink import Field, retry
from . import models
from ..core.helpers._partial_success import unwrap_single_item_partial_success
[docs]@retry(
when=retry.when.status(408, 429, 502, 503, 504),
stop=retry.stop.after_attempt(5),
on_exception=retry.CONNECTION_ERROR,
)
class SpecClient(BaseClient):
[docs] def __init__(self, configuration: core.HttpConfiguration | None = 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 Spec Service.
"""
if configuration is None:
configuration = core.HttpConfigurationManager.get_configuration()
super().__init__(configuration, base_path="/nispec/v1/")
[docs] @get("")
def api_info(self) -> models.V1Operations:
"""Get information about available API operations.
Returns:
Information about available API operations.
Raises:
ApiException: if unable to communicate with the DataFrame Service.
"""
...
[docs] @post("specs")
def create_specs(
self, specs: models.CreateSpecificationsRequest
) -> models.CreateSpecificationsPartialSuccess:
"""Creates one or more specifications.
Args:
specs: A list of specifications to create.
Returns:
A list of specs that were successfully created and ones that failed to be created.
Raises:
ApiException: if unable to communicate with the `/nispec` service or if there are
invalid arguments.
"""
...
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.",
)
[docs] @post("delete-specs", args=[Field("ids")])
def delete_specs(
self, ids: List[str]
) -> models.DeleteSpecificationsPartialSuccess | None:
"""Deletes one or more specifications by global id.
Args:
ids: a list of specification ids. Note that these are the global ids and not the
`specId` that is local to a product and workspace.
Returns:
None if all deletes succeed otherwise a list of which ids failed and which succeeded.
Raises:
ApiException: if unable to communicate with the `nispec` service or if there are invalid
arguments.
"""
...
[docs] @post("query-specs")
def query_specs(
self, query: models.QuerySpecificationsRequest
) -> models.PagedSpecifications:
"""Queries for specs that match the filters.
Args:
query: The query contains a product id as well as a filter for specs under that product.
Returns:
A list of specifications that match the filter.
"""
...
[docs] @get("specs/{id}")
def get_spec(self, id: str) -> models.Specification:
"""Retrieves a single spec by id.
Args:
id (str): Unique ID of a specification.
Returns:
The single spec matching `id`
Raises:
ApiException: if unable to communicate with the `nispec` service or if there are invalid
arguments.
"""
...
[docs] @post("update-specs")
def update_specs(
self, specs: models.UpdateSpecificationsRequest
) -> models.UpdateSpecificationsPartialSuccess | None:
"""Updates one or more specifications.
Update requires that the version field matches the version being updated from.
Args:
specs: a list of specifications that are to be updated. Must include the global id and
each spec being updated must match the version currently on the server.
Returns
A list of specs that were successfully updated and a list of ones that were not along
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.",
)