# pylint: disable=too-many-lines # coding=utf-8 # -------------------------------------------------------------------------- # Code generated by Microsoft (R) AutoRest Code Generator (autorest: 3.9.2, generator: @autorest/python@5.19.0) # Changes may cause incorrect behavior and will be lost if the code is regenerated. # -------------------------------------------------------------------------- from typing import Any, Callable, Dict, IO, Iterable, Optional, TypeVar, Union, cast, overload from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error from azure.core.paging import ItemPaged from azure.core.pipeline import PipelineResponse from azure.core.pipeline.transport import HttpResponse from azure.core.polling import LROPoller, NoPolling, PollingMethod from azure.core.polling.base_polling import LROBasePolling from azure.core.rest import HttpRequest from azure.core.tracing.decorator import distributed_trace from azure.core.utils import case_insensitive_dict from .. import models as _models from .._serialization import Serializer from .._vendor import _convert_request, _format_url_section T = TypeVar('T') ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]] _SERIALIZER = Serializer() _SERIALIZER.client_side_validation = False def build_list_device_classes_request( instance_id: str, *, filter: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceClasses") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') if filter is not None: _params['filter'] = _SERIALIZER.query("filter", filter, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_device_class_request( device_class_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceClasses/{deviceClassId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_update_device_class_request( device_class_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceClasses/{deviceClassId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if content_type is not None: _headers['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="PATCH", url=_url, params=_params, headers=_headers, **kwargs ) def build_delete_device_class_request( device_class_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceClasses/{deviceClassId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="DELETE", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_installable_updates_for_device_class_request( device_class_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceClasses/{deviceClassId}/installableUpdates") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_devices_request( instance_id: str, *, filter: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/devices") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters if filter is not None: _params['filter'] = _SERIALIZER.query("filter", filter, 'str') _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_import_devices_request( instance_id: str, *, json: Union[str, "_models.ImportType"], **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/devices:import") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if content_type is not None: _headers['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="POST", url=_url, params=_params, headers=_headers, json=json, **kwargs ) def build_get_device_request( device_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/devices/{deviceId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "deviceId": _SERIALIZER.url("device_id", device_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_device_module_request( device_id: str, module_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/devices/{deviceId}/modules/{moduleId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "deviceId": _SERIALIZER.url("device_id", device_id, 'str'), "moduleId": _SERIALIZER.url("module_id", module_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_update_compliance_request( instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/updateCompliance") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_groups_request( instance_id: str, *, order_by: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters if order_by is not None: _params['orderby'] = _SERIALIZER.query("order_by", order_by, 'str') _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_group_request( group_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_delete_group_request( group_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="DELETE", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_update_compliance_for_group_request( group_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/updateCompliance") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_best_updates_for_group_request( group_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/bestUpdates") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_deployments_for_group_request( group_id: str, instance_id: str, *, order_by: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') if order_by is not None: _params['orderby'] = _SERIALIZER.query("order_by", order_by, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_deployment_request( group_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_create_or_update_deployment_request( group_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if content_type is not None: _headers['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="PUT", url=_url, params=_params, headers=_headers, **kwargs ) def build_delete_deployment_request( group_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="DELETE", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_deployment_status_request( group_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}/status") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_device_class_subgroups_for_group_request( group_id: str, instance_id: str, *, filter: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters if filter is not None: _params['filter'] = _SERIALIZER.query("filter", filter, 'str') _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_device_class_subgroup_request( group_id: str, device_class_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_delete_device_class_subgroup_request( group_id: str, device_class_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="DELETE", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_device_class_subgroup_update_compliance_request( group_id: str, device_class_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/updateCompliance") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_best_updates_for_device_class_subgroup_request( group_id: str, device_class_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/bestUpdates") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_deployments_for_device_class_subgroup_request( group_id: str, device_class_id: str, instance_id: str, *, order_by: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') if order_by is not None: _params['orderby'] = _SERIALIZER.query("order_by", order_by, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_deployment_for_device_class_subgroup_request( group_id: str, device_class_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_delete_deployment_for_device_class_subgroup_request( group_id: str, device_class_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="DELETE", url=_url, params=_params, headers=_headers, **kwargs ) def build_stop_deployment_request( group_id: str, device_class_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}:cancel") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="POST", url=_url, params=_params, headers=_headers, **kwargs ) def build_retry_deployment_request( group_id: str, device_class_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}:retry") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="POST", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_device_class_subgroup_deployment_status_request( group_id: str, device_class_id: str, deployment_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}/status") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_device_states_for_device_class_subgroup_deployment_request( group_id: str, device_class_id: str, deployment_id: str, instance_id: str, *, filter: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}/devicestates") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "groupId": _SERIALIZER.url("group_id", group_id, 'str'), "deviceClassId": _SERIALIZER.url("device_class_id", device_class_id, 'str'), "deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters if filter is not None: _params['filter'] = _SERIALIZER.query("filter", filter, 'str') _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_operation_status_request( operation_id: str, instance_id: str, *, if_none_match: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/operations/{operationId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "operationId": _SERIALIZER.url("operation_id", operation_id, 'str', max_length=256, min_length=1), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if if_none_match is not None: _headers['If-None-Match'] = _SERIALIZER.header("if_none_match", if_none_match, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_operation_statuses_request( instance_id: str, *, filter: Optional[str] = None, top: Optional[int] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/operations") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters if filter is not None: _params['filter'] = _SERIALIZER.query("filter", filter, 'str') if top is not None: _params['top'] = _SERIALIZER.query("top", top, 'int') _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_start_log_collection_request( log_collection_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "operationId": _SERIALIZER.url("log_collection_id", log_collection_id, 'str', max_length=256, min_length=1), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if content_type is not None: _headers['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="PUT", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_log_collection_request( log_collection_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "operationId": _SERIALIZER.url("log_collection_id", log_collection_id, 'str', max_length=256, min_length=1), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_log_collections_request( instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceDiagnostics/logCollections") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_log_collection_detailed_status_request( log_collection_id: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}/detailedStatus") # pylint: disable=line-too-long path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), "operationId": _SERIALIZER.url("log_collection_id", log_collection_id, 'str', max_length=256, min_length=1), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_health_of_devices_request( instance_id: str, *, filter: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/management/deviceDiagnostics/deviceHealth") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') _params['filter'] = _SERIALIZER.query("filter", filter, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) class DeviceManagementOperations: # pylint: disable=too-many-public-methods """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~deviceupdateclient.DeviceUpdateClient`'s :attr:`device_management` attribute. """ models = _models def __init__(self, *args, **kwargs): input_args = list(args) self._client = input_args.pop(0) if input_args else kwargs.pop("client") self._config = input_args.pop(0) if input_args else kwargs.pop("config") self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace def list_device_classes( self, filter: Optional[str] = None, **kwargs: Any ) -> Iterable["_models.DeviceClass"]: """Gets a list of all device classes (sets of devices compatible with the same updates based on the model Id and compat properties reported in the Device Update PnP interface in IoT Hub) for all devices connected to Device Update for IoT Hub. :param filter: Restricts the set of device classes returned. You can filter on friendly name. Default value is None. :type filter: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either DeviceClass or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.DeviceClass] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceClassesList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_device_classes_request( instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=self.list_device_classes.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_device_classes_request( instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DeviceClassesList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_device_classes.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceClasses"} # type: ignore @distributed_trace def get_device_class( self, device_class_id: str, **kwargs: Any ) -> _models.DeviceClass: """Gets the properties of a device class. :param device_class_id: Device class identifier. Required. :type device_class_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: DeviceClass or the result of cls(response) :rtype: ~deviceupdateclient.models.DeviceClass :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceClass] request = build_get_device_class_request( device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_device_class.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('DeviceClass', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_device_class.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceClasses/{deviceClassId}"} # type: ignore @overload def update_device_class( self, device_class_id: str, device_class_patch: _models.PatchBody, *, content_type: str = "application/merge-patch+json", **kwargs: Any ) -> _models.DeviceClass: """Update device class details. :param device_class_id: Device class identifier. Required. :type device_class_id: str :param device_class_patch: The device class json merge patch body. Currently only supports patching friendlyName. Required. :type device_class_patch: ~deviceupdateclient.models.PatchBody :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/merge-patch+json". :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: DeviceClass or the result of cls(response) :rtype: ~deviceupdateclient.models.DeviceClass :raises ~azure.core.exceptions.HttpResponseError: """ @overload def update_device_class( self, device_class_id: str, device_class_patch: IO, *, content_type: str = "application/merge-patch+json", **kwargs: Any ) -> _models.DeviceClass: """Update device class details. :param device_class_id: Device class identifier. Required. :type device_class_id: str :param device_class_patch: The device class json merge patch body. Currently only supports patching friendlyName. Required. :type device_class_patch: IO :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/merge-patch+json". :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: DeviceClass or the result of cls(response) :rtype: ~deviceupdateclient.models.DeviceClass :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def update_device_class( self, device_class_id: str, device_class_patch: Union[_models.PatchBody, IO], **kwargs: Any ) -> _models.DeviceClass: """Update device class details. :param device_class_id: Device class identifier. Required. :type device_class_id: str :param device_class_patch: The device class json merge patch body. Currently only supports patching friendlyName. Is either a model type or a IO type. Required. :type device_class_patch: ~deviceupdateclient.models.PatchBody or IO :keyword content_type: Body Parameter content-type. Known values are: 'application/merge-patch+json'. Default value is None. :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: DeviceClass or the result of cls(response) :rtype: ~deviceupdateclient.models.DeviceClass :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceClass] content_type = content_type or "application/merge-patch+json" _json = None _content = None if isinstance(device_class_patch, (IO, bytes)): _content = device_class_patch else: _json = self._serialize.body(device_class_patch, 'PatchBody') request = build_update_device_class_request( device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, content_type=content_type, json=_json, content=_content, template_url=self.update_device_class.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('DeviceClass', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized update_device_class.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceClasses/{deviceClassId}"} # type: ignore @distributed_trace def delete_device_class( # pylint: disable=inconsistent-return-statements self, device_class_id: str, **kwargs: Any ) -> None: """Deletes a device class. Device classes are created automatically when Device Update-enabled devices are connected to the hub but are not automatically cleaned up since they are referenced by DeviceClassSubgroups. If the user has deleted all DeviceClassSubgroups for a device class they can also delete the device class to remove the records from the system and to stop checking the compatibility of this device class with new updates. If a device is ever reconnected for this device class it will be re-created. :param device_class_id: Device class identifier. Required. :type device_class_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: None or the result of cls(response) :rtype: None :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] request = build_delete_device_class_request( device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.delete_device_class.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [204]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) if cls: return cls(pipeline_response, None, {}) delete_device_class.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceClasses/{deviceClassId}"} # type: ignore @distributed_trace def list_installable_updates_for_device_class( self, device_class_id: str, **kwargs: Any ) -> Iterable["_models.UpdateInfo"]: """Gets a list of installable updates for a device class. :param device_class_id: Device class identifier. Required. :type device_class_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either UpdateInfo or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.UpdateInfo] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.UpdateInfoList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_installable_updates_for_device_class_request( device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.list_installable_updates_for_device_class.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_installable_updates_for_device_class_request( device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("UpdateInfoList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_installable_updates_for_device_class.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceClasses/{deviceClassId}/installableUpdates"} # type: ignore @distributed_trace def list_devices( self, filter: Optional[str] = None, **kwargs: Any ) -> Iterable["_models.Device"]: """Gets a list of devices connected to Device Update for IoT Hub. :param filter: Restricts the set of devices returned. You can filter on GroupId, DeviceClassId, or GroupId and DeploymentStatus. Use DeploymentStatus eq null to query for devices with no deployment status (that have never been deployed to). Default value is None. :type filter: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either Device or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.Device] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DevicesList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_devices_request( instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=self.list_devices.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_devices_request( instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DevicesList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_devices.metadata = {'url': "/deviceUpdate/{instanceId}/management/devices"} # type: ignore def _import_devices_initial( # pylint: disable=inconsistent-return-statements self, import_type: Union[str, "_models.ImportType"], **kwargs: Any ) -> None: error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', "application/json")) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] # @digimaun - str -> object _json = self._serialize.body(import_type, 'object') request = build_import_devices_request( instance_id=self._config.instance_id, api_version=api_version, content_type=content_type, json=_json, template_url=self._import_devices_initial.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [202]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers['Operation-Location']=self._deserialize('str', response.headers.get('Operation-Location')) if cls: return cls(pipeline_response, None, response_headers) _import_devices_initial.metadata = {'url': "/deviceUpdate/{instanceId}/management/devices:import"} # type: ignore @distributed_trace def begin_import_devices( self, import_type: Union[str, "_models.ImportType"], **kwargs: Any ) -> LROPoller[None]: """Import existing devices from IoT Hub. This is a long-running-operation; use Operation-Location response header value to check for operation status. :param import_type: The types of devices to import. Known values are: "Devices", "Modules", and "All". Required. :type import_type: str or ~deviceupdateclient.models.ImportType :keyword callable cls: A custom type or function that will be passed the direct response :keyword str continuation_token: A continuation token to restart a poller from a saved state. :keyword polling: By default, your polling method will be LROBasePolling. Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. :paramtype polling: bool or ~azure.core.polling.PollingMethod :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', "application/json")) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] lro_delay = kwargs.pop( 'polling_interval', self._config.polling_interval ) cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] if cont_token is None: raw_result = self._import_devices_initial( # type: ignore import_type=import_type, api_version=api_version, content_type=content_type, cls=lambda x,y,z: x, headers=_headers, params=_params, **kwargs ) kwargs.pop('error_map', None) def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements if cls: return cls(pipeline_response, None, {}) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } if polling is True: polling_method = cast(PollingMethod, LROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs )) # type: PollingMethod elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller.from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output ) return LROPoller(self._client, raw_result, get_long_running_output, polling_method) begin_import_devices.metadata = {'url': "/deviceUpdate/{instanceId}/management/devices:import"} # type: ignore @distributed_trace def get_device( self, device_id: str, **kwargs: Any ) -> _models.Device: """Gets the device properties and latest deployment status for a device connected to Device Update for IoT Hub. :param device_id: Device identifier in Azure IoT Hub. Required. :type device_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Device or the result of cls(response) :rtype: ~deviceupdateclient.models.Device :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.Device] request = build_get_device_request( device_id=device_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_device.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Device', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_device.metadata = {'url': "/deviceUpdate/{instanceId}/management/devices/{deviceId}"} # type: ignore @distributed_trace def get_device_module( self, device_id: str, module_id: str, **kwargs: Any ) -> _models.Device: """Gets the device module properties and latest deployment status for a device module connected to Device Update for IoT Hub. :param device_id: Device identifier in Azure IoT Hub. Required. :type device_id: str :param module_id: Device module identifier in Azure IoT Hub. Required. :type module_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Device or the result of cls(response) :rtype: ~deviceupdateclient.models.Device :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.Device] request = build_get_device_module_request( device_id=device_id, module_id=module_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_device_module.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Device', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_device_module.metadata = {'url': "/deviceUpdate/{instanceId}/management/devices/{deviceId}/modules/{moduleId}"} # type: ignore @distributed_trace def get_update_compliance( self, **kwargs: Any ) -> _models.UpdateCompliance: """Gets the breakdown of how many devices are on their latest update, have new updates available, or are in progress receiving new updates. :keyword callable cls: A custom type or function that will be passed the direct response :return: UpdateCompliance or the result of cls(response) :rtype: ~deviceupdateclient.models.UpdateCompliance :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.UpdateCompliance] request = build_get_update_compliance_request( instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_update_compliance.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('UpdateCompliance', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_update_compliance.metadata = {'url': "/deviceUpdate/{instanceId}/management/updateCompliance"} # type: ignore @distributed_trace def list_groups( self, order_by: Optional[str] = None, **kwargs: Any ) -> Iterable["_models.Group"]: """Gets a list of all device groups. The $default group will always be returned first. :param order_by: Orders the set of groups returned. You can order by groupId, deviceCount, createdDate, subgroupsWithNewUpdatesAvailableCount, subgroupsWithUpdatesInProgressCount, or subgroupsOnLatestUpdateCount. Default value is None. :type order_by: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either Group or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.Group] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.GroupsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_groups_request( instance_id=self._config.instance_id, order_by=order_by, api_version=api_version, template_url=self.list_groups.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_groups_request( instance_id=self._config.instance_id, order_by=order_by, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("GroupsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_groups.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups"} # type: ignore @distributed_trace def get_group( self, group_id: str, **kwargs: Any ) -> _models.Group: """Gets the device group properties. :param group_id: Group identifier. Required. :type group_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Group or the result of cls(response) :rtype: ~deviceupdateclient.models.Group :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.Group] request = build_get_group_request( group_id=group_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_group.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Group', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_group.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}"} # type: ignore @distributed_trace def delete_group( # pylint: disable=inconsistent-return-statements self, group_id: str, **kwargs: Any ) -> None: """Deletes a device group. This group is automatically created when a Device Update-enabled device is connected to the hub and reports its properties. Groups, subgroups, and deployments are not automatically cleaned up but are retained for history purposes. Users can call this method to delete a group if they do not need to retain any of the history of the group and no longer need it. If a device is ever connected again for this group after the group was deleted it will be automatically re-created but there will be no history. :param group_id: Group identifier. Required. :type group_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: None or the result of cls(response) :rtype: None :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] request = build_delete_group_request( group_id=group_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.delete_group.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [204]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) if cls: return cls(pipeline_response, None, {}) delete_group.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}"} # type: ignore @distributed_trace def get_update_compliance_for_group( self, group_id: str, **kwargs: Any ) -> _models.UpdateCompliance: """Get device group update compliance information such as how many devices are on their latest update, how many need new updates, and how many are in progress on receiving a new update. :param group_id: Group identifier. Required. :type group_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: UpdateCompliance or the result of cls(response) :rtype: ~deviceupdateclient.models.UpdateCompliance :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.UpdateCompliance] request = build_get_update_compliance_for_group_request( group_id=group_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_update_compliance_for_group.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('UpdateCompliance', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_update_compliance_for_group.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/updateCompliance"} # type: ignore @distributed_trace def list_best_updates_for_group( self, group_id: str, **kwargs: Any ) -> Iterable["_models.DeviceClassSubgroupUpdatableDevices"]: """Get the best available updates for a device group and a count of how many devices need each update. :param group_id: Group identifier. Required. :type group_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either DeviceClassSubgroupUpdatableDevices or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.DeviceClassSubgroupUpdatableDevices] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceClassSubgroupUpdatableDevicesList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_best_updates_for_group_request( group_id=group_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.list_best_updates_for_group.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_best_updates_for_group_request( group_id=group_id, instance_id=self._config.instance_id, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DeviceClassSubgroupUpdatableDevicesList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_best_updates_for_group.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/bestUpdates"} # type: ignore @distributed_trace def list_deployments_for_group( self, group_id: str, order_by: Optional[str] = None, **kwargs: Any ) -> Iterable["_models.Deployment"]: """Gets a list of deployments for a device group. :param group_id: Group identifier. Required. :type group_id: str :param order_by: Orders the set of deployments returned. You can order by start date. Default value is None. :type order_by: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either Deployment or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.Deployment] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeploymentsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_deployments_for_group_request( group_id=group_id, instance_id=self._config.instance_id, order_by=order_by, api_version=api_version, template_url=self.list_deployments_for_group.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_deployments_for_group_request( group_id=group_id, instance_id=self._config.instance_id, order_by=order_by, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DeploymentsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_deployments_for_group.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments"} # type: ignore @distributed_trace def get_deployment( self, group_id: str, deployment_id: str, **kwargs: Any ) -> _models.Deployment: """Gets the deployment properties. :param group_id: Group identifier. Required. :type group_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Deployment or the result of cls(response) :rtype: ~deviceupdateclient.models.Deployment :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.Deployment] request = build_get_deployment_request( group_id=group_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_deployment.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Deployment', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_deployment.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}"} # type: ignore @overload def create_or_update_deployment( self, group_id: str, deployment_id: str, deployment: _models.Deployment, *, content_type: str = "application/json", **kwargs: Any ) -> _models.Deployment: """Creates or updates a deployment. :param group_id: Group identifier. Required. :type group_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :param deployment: The deployment properties. Required. :type deployment: ~deviceupdateclient.models.Deployment :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Deployment or the result of cls(response) :rtype: ~deviceupdateclient.models.Deployment :raises ~azure.core.exceptions.HttpResponseError: """ @overload def create_or_update_deployment( self, group_id: str, deployment_id: str, deployment: IO, *, content_type: str = "application/json", **kwargs: Any ) -> _models.Deployment: """Creates or updates a deployment. :param group_id: Group identifier. Required. :type group_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :param deployment: The deployment properties. Required. :type deployment: IO :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Deployment or the result of cls(response) :rtype: ~deviceupdateclient.models.Deployment :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def create_or_update_deployment( self, group_id: str, deployment_id: str, deployment: Union[_models.Deployment, IO], **kwargs: Any ) -> _models.Deployment: """Creates or updates a deployment. :param group_id: Group identifier. Required. :type group_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :param deployment: The deployment properties. Is either a model type or a IO type. Required. :type deployment: ~deviceupdateclient.models.Deployment or IO :keyword content_type: Body Parameter content-type. Known values are: 'application/json'. Default value is None. :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Deployment or the result of cls(response) :rtype: ~deviceupdateclient.models.Deployment :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] cls = kwargs.pop('cls', None) # type: ClsType[_models.Deployment] content_type = content_type or "application/json" _json = None _content = None if isinstance(deployment, (IO, bytes)): _content = deployment else: _json = self._serialize.body(deployment, 'Deployment') request = build_create_or_update_deployment_request( group_id=group_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, content_type=content_type, json=_json, content=_content, template_url=self.create_or_update_deployment.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Deployment', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized create_or_update_deployment.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}"} # type: ignore @distributed_trace def delete_deployment( # pylint: disable=inconsistent-return-statements self, group_id: str, deployment_id: str, **kwargs: Any ) -> None: """Deletes a deployment. :param group_id: Group identifier. Required. :type group_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: None or the result of cls(response) :rtype: None :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] request = build_delete_deployment_request( group_id=group_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.delete_deployment.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [204]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) if cls: return cls(pipeline_response, None, {}) delete_deployment.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}"} # type: ignore @distributed_trace def get_deployment_status( self, group_id: str, deployment_id: str, **kwargs: Any ) -> _models.DeploymentStatus: """Gets the status of a deployment including a breakdown of how many devices in the deployment are in progress, completed, or failed. :param group_id: Group identifier. Required. :type group_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: DeploymentStatus or the result of cls(response) :rtype: ~deviceupdateclient.models.DeploymentStatus :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeploymentStatus] request = build_get_deployment_status_request( group_id=group_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_deployment_status.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('DeploymentStatus', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_deployment_status.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}/status"} # type: ignore @distributed_trace def list_device_class_subgroups_for_group( self, group_id: str, filter: Optional[str] = None, **kwargs: Any ) -> Iterable["_models.DeviceClassSubgroup"]: """Get the device class subgroups for the group. A device class subgroup is the set of devices within the group that share the same device class. All devices within the same device class are compatible with the same updates. :param group_id: Group identifier. Required. :type group_id: str :param filter: Restricts the set of device class subgroups returned. You can filter on compat properties by name and value. (i.e. filter=compatProperties/propertyName1 eq 'value1' and compatProperties/propertyName2 eq 'value2'). Default value is None. :type filter: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either DeviceClassSubgroup or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.DeviceClassSubgroup] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceClassSubgroupsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_device_class_subgroups_for_group_request( group_id=group_id, instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=self.list_device_class_subgroups_for_group.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_device_class_subgroups_for_group_request( group_id=group_id, instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DeviceClassSubgroupsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_device_class_subgroups_for_group.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups"} # type: ignore @distributed_trace def get_device_class_subgroup( self, group_id: str, device_class_id: str, **kwargs: Any ) -> _models.DeviceClassSubgroup: """Gets device class subgroup details. A device class subgroup is the set of devices within the group that share the same device class. All devices within the same device class are compatible with the same updates. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: DeviceClassSubgroup or the result of cls(response) :rtype: ~deviceupdateclient.models.DeviceClassSubgroup :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceClassSubgroup] request = build_get_device_class_subgroup_request( group_id=group_id, device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_device_class_subgroup.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('DeviceClassSubgroup', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_device_class_subgroup.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}"} # type: ignore @distributed_trace def delete_device_class_subgroup( # pylint: disable=inconsistent-return-statements self, group_id: str, device_class_id: str, **kwargs: Any ) -> None: """Deletes a device class subgroup. This subgroup is automatically created when a Device Update-enabled device is connected to the hub and reports its properties. Groups, subgroups, and deployments are not automatically cleaned up but are retained for history purposes. Users can call this method to delete a subgroup if they do not need to retain any of the history of the subgroup and no longer need it. If a device is ever connected again for this subgroup after the subgroup was deleted it will be automatically re-created but there will be no history. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: None or the result of cls(response) :rtype: None :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] request = build_delete_device_class_subgroup_request( group_id=group_id, device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.delete_device_class_subgroup.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [204]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) if cls: return cls(pipeline_response, None, {}) delete_device_class_subgroup.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}"} # type: ignore @distributed_trace def get_device_class_subgroup_update_compliance( self, group_id: str, device_class_id: str, **kwargs: Any ) -> _models.UpdateCompliance: """Get device class subgroup update compliance information such as how many devices are on their latest update, how many need new updates, and how many are in progress on receiving a new update. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: UpdateCompliance or the result of cls(response) :rtype: ~deviceupdateclient.models.UpdateCompliance :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.UpdateCompliance] request = build_get_device_class_subgroup_update_compliance_request( group_id=group_id, device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_device_class_subgroup_update_compliance.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('UpdateCompliance', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_device_class_subgroup_update_compliance.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/updateCompliance"} # type: ignore @distributed_trace def get_best_updates_for_device_class_subgroup( self, group_id: str, device_class_id: str, **kwargs: Any ) -> _models.DeviceClassSubgroupUpdatableDevices: """Get the best available update for a device class subgroup and a count of how many devices need this update. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: DeviceClassSubgroupUpdatableDevices or the result of cls(response) :rtype: ~deviceupdateclient.models.DeviceClassSubgroupUpdatableDevices :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceClassSubgroupUpdatableDevices] request = build_get_best_updates_for_device_class_subgroup_request( group_id=group_id, device_class_id=device_class_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_best_updates_for_device_class_subgroup.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('DeviceClassSubgroupUpdatableDevices', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_best_updates_for_device_class_subgroup.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/bestUpdates"} # type: ignore @distributed_trace def list_deployments_for_device_class_subgroup( self, group_id: str, device_class_id: str, order_by: Optional[str] = None, **kwargs: Any ) -> Iterable["_models.Deployment"]: """Gets a list of deployments for a device class subgroup. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :param order_by: Orders the set of deployments returned. You can order by start date. Default value is None. :type order_by: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either Deployment or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.Deployment] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeploymentsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_deployments_for_device_class_subgroup_request( group_id=group_id, device_class_id=device_class_id, instance_id=self._config.instance_id, order_by=order_by, api_version=api_version, template_url=self.list_deployments_for_device_class_subgroup.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_deployments_for_device_class_subgroup_request( group_id=group_id, device_class_id=device_class_id, instance_id=self._config.instance_id, order_by=order_by, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DeploymentsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_deployments_for_device_class_subgroup.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments"} # type: ignore @distributed_trace def get_deployment_for_device_class_subgroup( self, group_id: str, device_class_id: str, deployment_id: str, **kwargs: Any ) -> _models.Deployment: """Gets the deployment properties. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Deployment or the result of cls(response) :rtype: ~deviceupdateclient.models.Deployment :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.Deployment] request = build_get_deployment_for_device_class_subgroup_request( group_id=group_id, device_class_id=device_class_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_deployment_for_device_class_subgroup.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Deployment', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_deployment_for_device_class_subgroup.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}"} # type: ignore @distributed_trace def delete_deployment_for_device_class_subgroup( # pylint: disable=inconsistent-return-statements self, group_id: str, device_class_id: str, deployment_id: str, **kwargs: Any ) -> None: """Deletes a device class subgroup deployment. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: None or the result of cls(response) :rtype: None :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] request = build_delete_deployment_for_device_class_subgroup_request( group_id=group_id, device_class_id=device_class_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.delete_deployment_for_device_class_subgroup.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [204]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) if cls: return cls(pipeline_response, None, {}) delete_deployment_for_device_class_subgroup.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}"} # type: ignore @distributed_trace def stop_deployment( self, group_id: str, device_class_id: str, deployment_id: str, **kwargs: Any ) -> _models.Deployment: """Stops a deployment. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Deployment or the result of cls(response) :rtype: ~deviceupdateclient.models.Deployment :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.Deployment] request = build_stop_deployment_request( group_id=group_id, device_class_id=device_class_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.stop_deployment.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Deployment', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized stop_deployment.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}:cancel"} # type: ignore @distributed_trace def retry_deployment( self, group_id: str, device_class_id: str, deployment_id: str, **kwargs: Any ) -> _models.Deployment: """Retries a deployment with failed devices. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: Deployment or the result of cls(response) :rtype: ~deviceupdateclient.models.Deployment :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.Deployment] request = build_retry_deployment_request( group_id=group_id, device_class_id=device_class_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.retry_deployment.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Deployment', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized retry_deployment.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}:retry"} # type: ignore @distributed_trace def get_device_class_subgroup_deployment_status( self, group_id: str, device_class_id: str, deployment_id: str, **kwargs: Any ) -> _models.DeviceClassSubgroupDeploymentStatus: """Gets the status of a deployment including a breakdown of how many devices in the deployment are in progress, completed, or failed. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: DeviceClassSubgroupDeploymentStatus or the result of cls(response) :rtype: ~deviceupdateclient.models.DeviceClassSubgroupDeploymentStatus :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceClassSubgroupDeploymentStatus] request = build_get_device_class_subgroup_deployment_status_request( group_id=group_id, device_class_id=device_class_id, deployment_id=deployment_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_device_class_subgroup_deployment_status.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('DeviceClassSubgroupDeploymentStatus', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_device_class_subgroup_deployment_status.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}/status"} # type: ignore @distributed_trace def list_device_states_for_device_class_subgroup_deployment( self, group_id: str, device_class_id: str, deployment_id: str, filter: Optional[str] = None, **kwargs: Any ) -> Iterable["_models.DeploymentDeviceState"]: """Gets a list of devices in a deployment along with their state. Useful for getting a list of failed devices. :param group_id: Group identifier. Required. :type group_id: str :param device_class_id: Device class identifier. Required. :type device_class_id: str :param deployment_id: Deployment identifier. Required. :type deployment_id: str :param filter: Restricts the set of deployment device states returned. You can filter on deviceId and moduleId and/or deviceState. Default value is None. :type filter: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either DeploymentDeviceState or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.DeploymentDeviceState] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeploymentDeviceStatesList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_device_states_for_device_class_subgroup_deployment_request( group_id=group_id, device_class_id=device_class_id, deployment_id=deployment_id, instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=self.list_device_states_for_device_class_subgroup_deployment.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_device_states_for_device_class_subgroup_deployment_request( group_id=group_id, device_class_id=device_class_id, deployment_id=deployment_id, instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DeploymentDeviceStatesList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_device_states_for_device_class_subgroup_deployment.metadata = {'url': "/deviceUpdate/{instanceId}/management/groups/{groupId}/deviceClassSubgroups/{deviceClassId}/deployments/{deploymentId}/devicestates"} # type: ignore @distributed_trace def get_operation_status( self, operation_id: str, access_condition: Optional[_models.AccessCondition] = None, **kwargs: Any ) -> _models.DeviceOperation: """Retrieve operation status. :param operation_id: Operation identifier. Required. :type operation_id: str :param access_condition: Parameter group. Default value is None. :type access_condition: ~deviceupdateclient.models.AccessCondition :keyword callable cls: A custom type or function that will be passed the direct response :return: DeviceOperation or the result of cls(response) :rtype: ~deviceupdateclient.models.DeviceOperation :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceOperation] _if_none_match = None if access_condition is not None: _if_none_match = access_condition.if_none_match request = build_get_operation_status_request( operation_id=operation_id, instance_id=self._config.instance_id, if_none_match=_if_none_match, api_version=api_version, template_url=self.get_operation_status.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers['Retry-After']=self._deserialize('str', response.headers.get('Retry-After')) deserialized = self._deserialize('DeviceOperation', pipeline_response) if cls: return cls(pipeline_response, deserialized, response_headers) return deserialized get_operation_status.metadata = {'url': "/deviceUpdate/{instanceId}/management/operations/{operationId}"} # type: ignore @distributed_trace def list_operation_statuses( self, filter: Optional[str] = None, top: Optional[int] = None, **kwargs: Any ) -> Iterable["_models.DeviceOperation"]: """Get a list of all device import operations. Completed operations are kept for 7 days before auto-deleted. :param filter: Restricts the set of operations returned. Only one specific filter is supported: "status eq 'NotStarted' or status eq 'Running'". Default value is None. :type filter: str :param top: Specifies a non-negative integer n that limits the number of items returned from a collection. The service returns the number of available items up to but not greater than the specified value n. Default value is None. :type top: int :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either DeviceOperation or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.DeviceOperation] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceOperationsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_operation_statuses_request( instance_id=self._config.instance_id, filter=filter, top=top, api_version=api_version, template_url=self.list_operation_statuses.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_operation_statuses_request( instance_id=self._config.instance_id, filter=filter, top=top, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DeviceOperationsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_operation_statuses.metadata = {'url': "/deviceUpdate/{instanceId}/management/operations"} # type: ignore @overload def start_log_collection( self, log_collection_id: str, log_collection: _models.LogCollection, *, content_type: str = "application/json", **kwargs: Any ) -> _models.LogCollection: """Start the device diagnostics log collection on specified devices. :param log_collection_id: Log collection identifier. Required. :type log_collection_id: str :param log_collection: The log collection properties. Required. :type log_collection: ~deviceupdateclient.models.LogCollection :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: LogCollection or the result of cls(response) :rtype: ~deviceupdateclient.models.LogCollection :raises ~azure.core.exceptions.HttpResponseError: """ @overload def start_log_collection( self, log_collection_id: str, log_collection: IO, *, content_type: str = "application/json", **kwargs: Any ) -> _models.LogCollection: """Start the device diagnostics log collection on specified devices. :param log_collection_id: Log collection identifier. Required. :type log_collection_id: str :param log_collection: The log collection properties. Required. :type log_collection: IO :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: LogCollection or the result of cls(response) :rtype: ~deviceupdateclient.models.LogCollection :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def start_log_collection( self, log_collection_id: str, log_collection: Union[_models.LogCollection, IO], **kwargs: Any ) -> _models.LogCollection: """Start the device diagnostics log collection on specified devices. :param log_collection_id: Log collection identifier. Required. :type log_collection_id: str :param log_collection: The log collection properties. Is either a model type or a IO type. Required. :type log_collection: ~deviceupdateclient.models.LogCollection or IO :keyword content_type: Body Parameter content-type. Known values are: 'application/json'. Default value is None. :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :return: LogCollection or the result of cls(response) :rtype: ~deviceupdateclient.models.LogCollection :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] cls = kwargs.pop('cls', None) # type: ClsType[_models.LogCollection] content_type = content_type or "application/json" _json = None _content = None if isinstance(log_collection, (IO, bytes)): _content = log_collection else: _json = self._serialize.body(log_collection, 'LogCollection') request = build_start_log_collection_request( log_collection_id=log_collection_id, instance_id=self._config.instance_id, api_version=api_version, content_type=content_type, json=_json, content=_content, template_url=self.start_log_collection.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [201]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('LogCollection', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized start_log_collection.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}"} # type: ignore @distributed_trace def get_log_collection( self, log_collection_id: str, **kwargs: Any ) -> _models.LogCollection: """Get the device diagnostics log collection. :param log_collection_id: Log collection identifier. Required. :type log_collection_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: LogCollection or the result of cls(response) :rtype: ~deviceupdateclient.models.LogCollection :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.LogCollection] request = build_get_log_collection_request( log_collection_id=log_collection_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_log_collection.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('LogCollection', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_log_collection.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}"} # type: ignore @distributed_trace def list_log_collections( self, **kwargs: Any ) -> Iterable["_models.LogCollection"]: """Get all device diagnostics log collections. :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either LogCollection or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.LogCollection] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.LogCollectionList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_log_collections_request( instance_id=self._config.instance_id, api_version=api_version, template_url=self.list_log_collections.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_log_collections_request( instance_id=self._config.instance_id, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("LogCollectionList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_log_collections.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceDiagnostics/logCollections"} # type: ignore @distributed_trace def get_log_collection_detailed_status( self, log_collection_id: str, **kwargs: Any ) -> _models.LogCollectionOperationDetailedStatus: """Get log collection with detailed status. :param log_collection_id: Log collection identifier. Required. :type log_collection_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: LogCollectionOperationDetailedStatus or the result of cls(response) :rtype: ~deviceupdateclient.models.LogCollectionOperationDetailedStatus :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.LogCollectionOperationDetailedStatus] request = build_get_log_collection_detailed_status_request( log_collection_id=log_collection_id, instance_id=self._config.instance_id, api_version=api_version, template_url=self.get_log_collection_detailed_status.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('LogCollectionOperationDetailedStatus', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized get_log_collection_detailed_status.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}/detailedStatus"} # type: ignore @distributed_trace def list_health_of_devices( self, filter: str, **kwargs: Any ) -> Iterable["_models.DeviceHealth"]: """Get list of device health. :param filter: Restricts the set of devices for which device health is returned. You can filter on status, device id and module id. Required. :type filter: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either DeviceHealth or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.DeviceHealth] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.DeviceHealthList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_health_of_devices_request( instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=self.list_health_of_devices.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_health_of_devices_request( instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("DeviceHealthList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_health_of_devices.metadata = {'url': "/deviceUpdate/{instanceId}/management/deviceDiagnostics/deviceHealth"} # type: ignore