# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
import functools
from json import loads as _loads
from typing import Any, Callable, Dict, Generic, Iterable, List, Optional, TypeVar, Union
import warnings
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 msrest import Serializer
from .._vendor import _format_url_section
T = TypeVar('T')
JSONType = Any
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
_SERIALIZER = Serializer()
_SERIALIZER.client_side_validation = False
def build_device_update_import_update_request_initial(
instance_id: str,
*,
action: str,
json: JSONType = None,
content: Any = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', None) # type: Optional[str]
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['action'] = _SERIALIZER.query("action", action, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
json=json,
content=content,
**kwargs
)
def build_device_update_list_updates_request(
instance_id: str,
*,
search: Optional[str] = None,
filter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
if search is not None:
query_parameters['$search'] = _SERIALIZER.query("search", search, 'str')
if filter is not None:
query_parameters['$filter'] = _SERIALIZER.query("filter", filter, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_get_update_request(
instance_id: str,
provider: str,
name: str,
version: str,
*,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
"provider": _SERIALIZER.url("provider", provider, 'str'),
"name": _SERIALIZER.url("name", name, 'str'),
"version": _SERIALIZER.url("version", version, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if if_none_match is not None:
header_parameters['If-None-Match'] = _SERIALIZER.header("if_none_match", if_none_match, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_delete_update_request_initial(
instance_id: str,
provider: str,
name: str,
version: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
"provider": _SERIALIZER.url("provider", provider, 'str'),
"name": _SERIALIZER.url("name", name, 'str'),
"version": _SERIALIZER.url("version", version, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_list_providers_request(
instance_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/providers'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_list_names_request(
instance_id: str,
provider: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/providers/{provider}/names'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
"provider": _SERIALIZER.url("provider", provider, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_list_versions_request(
instance_id: str,
provider: str,
name: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
"provider": _SERIALIZER.url("provider", provider, 'str'),
"name": _SERIALIZER.url("name", name, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
if filter is not None:
query_parameters['$filter'] = _SERIALIZER.query("filter", filter, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_list_files_request(
instance_id: str,
provider: str,
name: str,
version: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}/files'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
"provider": _SERIALIZER.url("provider", provider, 'str'),
"name": _SERIALIZER.url("name", name, 'str'),
"version": _SERIALIZER.url("version", version, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_get_file_request(
instance_id: str,
provider: str,
name: str,
version: str,
file_id: str,
*,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}/files/{fileId}'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
"provider": _SERIALIZER.url("provider", provider, 'str'),
"name": _SERIALIZER.url("name", name, 'str'),
"version": _SERIALIZER.url("version", version, 'str'),
"fileId": _SERIALIZER.url("file_id", file_id, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if if_none_match is not None:
header_parameters['If-None-Match'] = _SERIALIZER.header("if_none_match", if_none_match, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_list_operations_request(
instance_id: str,
*,
filter: Optional[str] = None,
top: Optional[int] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if filter is not None:
query_parameters['$filter'] = _SERIALIZER.query("filter", filter, 'str')
if top is not None:
query_parameters['$top'] = _SERIALIZER.query("top", top, 'int')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_update_get_operation_request(
instance_id: str,
operation_id: str,
*,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/updates/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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if if_none_match is not None:
header_parameters['If-None-Match'] = _SERIALIZER.header("if_none_match", if_none_match, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_device_classes_request(
instance_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_device_class_request(
instance_id: str,
device_class_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_installable_updates_for_device_class_request(
instance_id: str,
device_class_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/deviceclasses/{deviceClassId}/installableupdates'
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_devices_request(
instance_id: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if filter is not None:
query_parameters['$filter'] = _SERIALIZER.query("filter", filter, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_import_devices_request_initial(
instance_id: str,
*,
action: str,
json: JSONType = None,
content: Any = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', None) # type: Optional[str]
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
query_parameters['action'] = _SERIALIZER.query("action", action, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
json=json,
content=content,
**kwargs
)
def build_device_management_get_device_request(
instance_id: str,
device_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_device_module_request(
instance_id: str,
device_id: str,
module_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_update_compliance_request(
instance_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_device_tags_request(
instance_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/devicetags'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_device_tag_request(
instance_id: str,
tag_name: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/devicetags/{tagName}'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
"tagName": _SERIALIZER.url("tag_name", tag_name, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_groups_request(
instance_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_group_request(
instance_id: str,
group_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_create_or_update_group_request(
instance_id: str,
group_id: str,
*,
json: JSONType = None,
content: Any = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', None) # type: Optional[str]
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="PUT",
url=url,
params=query_parameters,
headers=header_parameters,
json=json,
content=content,
**kwargs
)
def build_device_management_delete_group_request(
instance_id: str,
group_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_group_update_compliance_request(
instance_id: str,
group_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_best_updates_for_group_request(
instance_id: str,
group_id: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if filter is not None:
query_parameters['$filter'] = _SERIALIZER.query("filter", filter, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_deployments_for_group_request(
instance_id: str,
group_id: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if filter is not None:
query_parameters['$filter'] = _SERIALIZER.query("filter", filter, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_deployment_request(
instance_id: str,
group_id: str,
deployment_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_create_or_update_deployment_request(
instance_id: str,
deployment_id: str,
group_id: str,
*,
json: JSONType = None,
content: Any = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', None) # type: Optional[str]
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}'
path_format_arguments = {
"instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True),
"deploymentId": _SERIALIZER.url("deployment_id", deployment_id, 'str'),
"groupId": _SERIALIZER.url("group_id", group_id, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="PUT",
url=url,
params=query_parameters,
headers=header_parameters,
json=json,
content=content,
**kwargs
)
def build_device_management_delete_deployment_request(
instance_id: str,
group_id: str,
deployment_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_deployment_status_request(
instance_id: str,
group_id: str,
deployment_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}/status'
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_deployment_devices_request(
instance_id: str,
group_id: str,
deployment_id: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}/devicestates'
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if filter is not None:
query_parameters['$filter'] = _SERIALIZER.query("filter", filter, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_operation_request(
instance_id: str,
operation_id: str,
*,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if if_none_match is not None:
header_parameters['If-None-Match'] = _SERIALIZER.header("if_none_match", if_none_match, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_operations_request(
instance_id: str,
*,
filter: Optional[str] = None,
top: Optional[int] = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
if filter is not None:
query_parameters['$filter'] = _SERIALIZER.query("filter", filter, 'str')
if top is not None:
query_parameters['$top'] = _SERIALIZER.query("top", top, 'int')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_collect_logs_request(
instance_id: str,
operation_id: str,
*,
json: JSONType = None,
content: Any = None,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', None) # type: Optional[str]
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/deviceDiagnostics/logCollections/{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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="PUT",
url=url,
params=query_parameters,
headers=header_parameters,
json=json,
content=content,
**kwargs
)
def build_device_management_get_log_collection_operation_request(
instance_id: str,
operation_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/deviceDiagnostics/logCollections/{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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_list_log_collection_operations_request(
instance_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_get_log_collection_operation_detailed_status_request(
instance_id: str,
operation_id: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
url = '/deviceupdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}/detailedStatus'
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_stop_deployment_request(
instance_id: str,
group_id: str,
deployment_id: str,
*,
action: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['action'] = _SERIALIZER.query("action", action, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
def build_device_management_retry_deployment_request(
instance_id: str,
group_id: str,
deployment_id: str,
*,
action: str,
**kwargs: Any
) -> HttpRequest:
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
accept = "application/json"
# Construct URL
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
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['action'] = _SERIALIZER.query("action", action, 'str')
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="POST",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
[docs]class DeviceUpdateOperations(object):
"""DeviceUpdateOperations operations.
You should not instantiate this class directly. Instead, you should create a Client instance that
instantiates it for you and attaches it as an attribute.
:param client: Client for service requests.
:param config: Configuration of service client.
:param serializer: An object model serializer.
:param deserializer: An object model deserializer.
"""
def __init__(self, client, config, serializer, deserializer):
self._client = client
self._serialize = serializer
self._deserialize = deserializer
self._config = config
def _import_update_initial(
self,
update_to_import: List[JSONType],
*,
action: str,
**kwargs: Any
) -> JSONType:
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', "application/json") # type: Optional[str]
_json = update_to_import
request = build_device_update_import_update_request_initial(
instance_id=self._config.instance_id,
api_version=api_version,
content_type=content_type,
action=action,
json=_json,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
response_headers = {}
response_headers['Operation-Location']=self._deserialize('str', response.headers.get('Operation-Location'))
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, response_headers)
return deserialized
_import_update_initial.metadata = {'url': '/deviceupdate/{instanceId}/updates'} # type: ignore
[docs] @distributed_trace
def begin_import_update(
self,
update_to_import: List[JSONType],
*,
action: str,
**kwargs: Any
) -> LROPoller[JSONType]:
"""Import new update version.
:param update_to_import: The update to be imported.
:type update_to_import: list[JSONType]
:keyword action: Import update action. "import"
:paramtype action: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
: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 JSON object
:rtype: ~azure.core.polling.LROPoller[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# JSON input template you can fill out and use as your body input.
update_to_import = [
{
"files": [
{
"filename": "str", # Required. Update file name as specified inside import manifest.
"url": "str" # Required. Azure Blob location from which the update file can be downloaded by Device Update for IoT Hub. This is typically a read-only SAS-protected blob URL with an expiration set to at least 4 hours.
}
],
"friendlyName": "str", # Optional. Friendly update name.
"importManifest": {
"hashes": {
"str": "str" # Required. A JSON object containing the hash(es) of the file. At least SHA256 hash is required. This object can be thought of as a set of key-value pairs where the key is the hash algorithm, and the value is the hash of the file calculated using that algorithm.
},
"sizeInBytes": 0.0, # Required. File size in number of bytes.
"url": "str" # Required. Azure Blob location from which the import manifest can be downloaded by Device Update for IoT Hub. This is typically a read-only SAS-protected blob URL with an expiration set to at least 4 hours.
}
}
]
# response body for status code(s): 202
response.json() == {
"compatibility": [
{
"str": "str" # Required. List of update compatibility information.
}
],
"createdDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the update was created.
"description": "str", # Optional. Update description specified by creator.
"etag": "str", # Optional. Update ETag.
"friendlyName": "str", # Optional. Friendly update name specified by importer.
"importedDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the update was imported.
"installedCriteria": "str", # Optional. String interpreted by Device Update client to determine if the update is installed on the device. Deprecated in latest import manifest schema.
"instructions": {
"steps": [
{
"description": "str", # Optional. Step description.
"files": [
"str" # Optional. Collection of file names to be passed to handler during execution. Required if step type is inline.
],
"handler": "str", # Optional. Identity of handler that will execute this step. Required if step type is inline.
"handlerProperties": {}, # Optional. Parameters to be passed to handler during execution.
"type": "inline", # Optional. Default value is "inline". Step type. Possible values include: "Inline", "Reference". Default value: "inline".
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
},
"isDeployable": True, # Optional. Default value is True. Whether the update can be deployed to a device on its own.
"manifestVersion": "str", # Required. Schema version of manifest used to import the update.
"referencedBy": [
{
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
],
"scanResult": "str", # Optional. Update aggregate scan result (calculated from payload file scan results).
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"updateType": "str" # Optional. Update type. Deprecated in latest import manifest schema.
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', "application/json") # type: Optional[str]
polling = kwargs.pop('polling', True) # type: Union[bool, azure.core.polling.PollingMethod]
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
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_update_initial(
update_to_import=update_to_import,
action=action,
api_version=api_version,
content_type=content_type,
cls=lambda x,y,z: x,
**kwargs
)
kwargs.pop('error_map', None)
def get_long_running_output(pipeline_response):
response_headers = {}
response = pipeline_response.http_response
response_headers['Operation-Location']=self._deserialize('str', response.headers.get('Operation-Location'))
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, response_headers)
return deserialized
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
if polling is True: polling_method = LROBasePolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
elif polling is False: polling_method = 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
)
else:
return LROPoller(self._client, raw_result, get_long_running_output, polling_method)
begin_import_update.metadata = {'url': '/deviceupdate/{instanceId}/updates'} # type: ignore
[docs] @distributed_trace
def list_updates(
self,
*,
search: Optional[str] = None,
filter: Optional[str] = None,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get a list of all updates that have been imported to Device Update for IoT Hub.
:keyword search: Request updates matching a free-text search expression.
:paramtype search: str
:keyword filter: Filter updates by its properties.
:paramtype filter: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"compatibility": [
{
"str": "str" # Required. List of update compatibility information.
}
],
"createdDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the update was created.
"description": "str", # Optional. Update description specified by creator.
"etag": "str", # Optional. Update ETag.
"friendlyName": "str", # Optional. Friendly update name specified by importer.
"importedDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the update was imported.
"installedCriteria": "str", # Optional. String interpreted by Device Update client to determine if the update is installed on the device. Deprecated in latest import manifest schema.
"instructions": {
"steps": [
{
"description": "str", # Optional. Step description.
"files": [
"str" # Optional. Collection of file names to be passed to handler during execution. Required if step type is inline.
],
"handler": "str", # Optional. Identity of handler that will execute this step. Required if step type is inline.
"handlerProperties": {}, # Optional. Parameters to be passed to handler during execution.
"type": "inline", # Optional. Default value is "inline". Step type. Possible values include: "Inline", "Reference". Default value: "inline".
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
},
"isDeployable": True, # Optional. Default value is True. Whether the update can be deployed to a device on its own.
"manifestVersion": "str", # Required. Schema version of manifest used to import the update.
"referencedBy": [
{
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
],
"scanResult": "str", # Optional. Update aggregate scan result (calculated from payload file scan results).
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"updateType": "str" # Optional. Update type. Deprecated in latest import manifest schema.
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_update_list_updates_request(
instance_id=self._config.instance_id,
api_version=api_version,
search=search,
filter=filter,
)
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)
else:
request = build_device_update_list_updates_request(
instance_id=self._config.instance_id,
api_version=api_version,
search=search,
filter=filter,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_updates.metadata = {'url': '/deviceupdate/{instanceId}/updates'} # type: ignore
[docs] @distributed_trace
def get_update(
self,
provider: str,
name: str,
version: str,
*,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> Optional[JSONType]:
"""Get a specific update version.
:param provider: Update provider.
:type provider: str
:param name: Update name.
:type name: str
:param version: Update version.
:type version: str
:keyword if_none_match: Defines the If-None-Match condition. The operation will be performed
only if the ETag on the server does not match this value.
:paramtype if_none_match: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType or None
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"compatibility": [
{
"str": "str" # Required. List of update compatibility information.
}
],
"createdDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the update was created.
"description": "str", # Optional. Update description specified by creator.
"etag": "str", # Optional. Update ETag.
"friendlyName": "str", # Optional. Friendly update name specified by importer.
"importedDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the update was imported.
"installedCriteria": "str", # Optional. String interpreted by Device Update client to determine if the update is installed on the device. Deprecated in latest import manifest schema.
"instructions": {
"steps": [
{
"description": "str", # Optional. Step description.
"files": [
"str" # Optional. Collection of file names to be passed to handler during execution. Required if step type is inline.
],
"handler": "str", # Optional. Identity of handler that will execute this step. Required if step type is inline.
"handlerProperties": {}, # Optional. Parameters to be passed to handler during execution.
"type": "inline", # Optional. Default value is "inline". Step type. Possible values include: "Inline", "Reference". Default value: "inline".
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
},
"isDeployable": True, # Optional. Default value is True. Whether the update can be deployed to a device on its own.
"manifestVersion": "str", # Required. Schema version of manifest used to import the update.
"referencedBy": [
{
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
],
"scanResult": "str", # Optional. Update aggregate scan result (calculated from payload file scan results).
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"updateType": "str" # Optional. Update type. Deprecated in latest import manifest schema.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[Optional[JSONType]]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_update_get_update_request(
instance_id=self._config.instance_id,
provider=provider,
name=name,
version=version,
api_version=api_version,
if_none_match=if_none_match,
)
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)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 304]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
deserialized = None
if response.status_code == 200:
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_update.metadata = {'url': '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}'} # type: ignore
def _delete_update_initial(
self,
provider: str,
name: str,
version: str,
**kwargs: Any
) -> None:
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_update_delete_update_request_initial(
instance_id=self._config.instance_id,
provider=provider,
name=name,
version=version,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
response_headers = {}
response_headers['Operation-Location']=self._deserialize('str', response.headers.get('Operation-Location'))
if cls:
return cls(pipeline_response, None, response_headers)
_delete_update_initial.metadata = {'url': '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}'} # type: ignore
[docs] @distributed_trace
def begin_delete_update(
self,
provider: str,
name: str,
version: str,
**kwargs: Any
) -> LROPoller[None]:
"""Delete a specific update version.
:param provider: Update provider.
:type provider: str
:param name: Update name.
:type name: str
:param version: Update version.
:type version: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
: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 None
:rtype: ~azure.core.polling.LROPoller[None]
:raises: ~azure.core.exceptions.HttpResponseError
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
polling = kwargs.pop('polling', True) # type: Union[bool, azure.core.polling.PollingMethod]
cls = kwargs.pop('cls', None) # type: ClsType[None]
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._delete_update_initial(
provider=provider,
name=name,
version=version,
api_version=api_version,
cls=lambda x,y,z: x,
**kwargs
)
kwargs.pop('error_map', None)
def get_long_running_output(pipeline_response):
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 = LROBasePolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
elif polling is False: polling_method = 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
)
else:
return LROPoller(self._client, raw_result, get_long_running_output, polling_method)
begin_delete_update.metadata = {'url': '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}'} # type: ignore
[docs] @distributed_trace
def list_providers(
self,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get a list of all update providers that have been imported to Device Update for IoT Hub.
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
"str" # Required. The collection of pageable items.
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_update_list_providers_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
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)
else:
request = build_device_update_list_providers_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_providers.metadata = {'url': '/deviceupdate/{instanceId}/updates/providers'} # type: ignore
[docs] @distributed_trace
def list_names(
self,
provider: str,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get a list of all update names that match the specified provider.
:param provider: Update provider.
:type provider: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
"str" # Required. The collection of pageable items.
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_update_list_names_request(
instance_id=self._config.instance_id,
provider=provider,
api_version=api_version,
)
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)
else:
request = build_device_update_list_names_request(
instance_id=self._config.instance_id,
provider=provider,
api_version=api_version,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_names.metadata = {'url': '/deviceupdate/{instanceId}/updates/providers/{provider}/names'} # type: ignore
[docs] @distributed_trace
def list_versions(
self,
provider: str,
name: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get a list of all update versions that match the specified provider and name.
:param provider: Update provider.
:type provider: str
:param name: Update name.
:type name: str
:keyword filter: Filter updates by its properties.
:paramtype filter: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
"str" # Required. The collection of pageable items.
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_update_list_versions_request(
instance_id=self._config.instance_id,
provider=provider,
name=name,
api_version=api_version,
filter=filter,
)
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)
else:
request = build_device_update_list_versions_request(
instance_id=self._config.instance_id,
provider=provider,
name=name,
api_version=api_version,
filter=filter,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_versions.metadata = {'url': '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions'} # type: ignore
[docs] @distributed_trace
def list_files(
self,
provider: str,
name: str,
version: str,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get a list of all update file identifiers for the specified version.
:param provider: Update provider.
:type provider: str
:param name: Update name.
:type name: str
:param version: Update version.
:type version: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
"str" # Required. The collection of pageable items.
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_update_list_files_request(
instance_id=self._config.instance_id,
provider=provider,
name=name,
version=version,
api_version=api_version,
)
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)
else:
request = build_device_update_list_files_request(
instance_id=self._config.instance_id,
provider=provider,
name=name,
version=version,
api_version=api_version,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_files.metadata = {'url': '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}/files'} # type: ignore
[docs] @distributed_trace
def get_file(
self,
provider: str,
name: str,
version: str,
file_id: str,
*,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> Optional[JSONType]:
"""Get a specific update file from the version.
:param provider: Update provider.
:type provider: str
:param name: Update name.
:type name: str
:param version: Update version.
:type version: str
:param file_id: File identifier.
:type file_id: str
:keyword if_none_match: Defines the If-None-Match condition. The operation will be performed
only if the ETag on the server does not match this value.
:paramtype if_none_match: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType or None
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"etag": "str", # Optional. File ETag.
"fileId": "str", # Required. File identity, generated by server at import time.
"fileName": "str", # Required. File name.
"hashes": {
"str": "str" # Required. Mapping of hashing algorithm to base64 encoded hash values.
},
"mimeType": "str", # Optional. File MIME type.
"scanDetails": "str", # Optional. Anti-malware scan details.
"scanResult": "str", # Optional. Anti-malware scan result.
"sizeInBytes": 0.0 # Required. File size in number of bytes.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[Optional[JSONType]]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_update_get_file_request(
instance_id=self._config.instance_id,
provider=provider,
name=name,
version=version,
file_id=file_id,
api_version=api_version,
if_none_match=if_none_match,
)
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)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 304]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
deserialized = None
if response.status_code == 200:
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_file.metadata = {'url': '/deviceupdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}/files/{fileId}'} # type: ignore
[docs] @distributed_trace
def list_operations(
self,
*,
filter: Optional[str] = None,
top: Optional[int] = None,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get a list of all import update operations. Completed operations are kept for 7 days before
auto-deleted. Delete operations are not returned by this API version.
:keyword filter: Restricts the set of operations returned. Only one specific filter is
supported: "status eq 'NotStarted' or status eq 'Running'".
:paramtype filter: str
:keyword 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.
:paramtype top: int
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"createdDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the operation was created.
"error": {
"code": "str", # Required. Server defined error code.
"details": [
...
],
"innererror": {
"code": "str", # Required. A more specific error code than what was provided by the containing error.
"errorDetail": "str", # Optional. The internal error or exception message.
"innerError": ...,
"message": "str" # Optional. A human-readable representation of the error.
},
"message": "str", # Required. A human-readable representation of the error.
"occurredDateTime": "2020-02-20 00:00:00", # Optional. Date and time in UTC when the error occurred.
"target": "str" # Optional. The target of the error.
},
"etag": "str", # Optional. Operation ETag.
"lastActionDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the operation status was last updated.
"operationId": "str", # Required. Operation Id.
"resourceLocation": "str", # Optional. Location of the imported update when operation is successful.
"status": "str", # Required. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
"traceId": "str", # Optional. Operation correlation identity that can used by Microsoft Support for troubleshooting.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_update_list_operations_request(
instance_id=self._config.instance_id,
api_version=api_version,
filter=filter,
top=top,
)
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)
else:
request = build_device_update_list_operations_request(
instance_id=self._config.instance_id,
api_version=api_version,
filter=filter,
top=top,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_operations.metadata = {'url': '/deviceupdate/{instanceId}/updates/operations'} # type: ignore
[docs] @distributed_trace
def get_operation(
self,
operation_id: str,
*,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> Optional[JSONType]:
"""Retrieve operation status.
:param operation_id: Operation identifier.
:type operation_id: str
:keyword if_none_match: Defines the If-None-Match condition. The operation will be performed
only if the ETag on the server does not match this value.
:paramtype if_none_match: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType or None
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"createdDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the operation was created.
"error": {
"code": "str", # Required. Server defined error code.
"details": [
...
],
"innererror": {
"code": "str", # Required. A more specific error code than what was provided by the containing error.
"errorDetail": "str", # Optional. The internal error or exception message.
"innerError": ...,
"message": "str" # Optional. A human-readable representation of the error.
},
"message": "str", # Required. A human-readable representation of the error.
"occurredDateTime": "2020-02-20 00:00:00", # Optional. Date and time in UTC when the error occurred.
"target": "str" # Optional. The target of the error.
},
"etag": "str", # Optional. Operation ETag.
"lastActionDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the operation status was last updated.
"operationId": "str", # Required. Operation Id.
"resourceLocation": "str", # Optional. Location of the imported update when operation is successful.
"status": "str", # Required. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
"traceId": "str", # Optional. Operation correlation identity that can used by Microsoft Support for troubleshooting.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[Optional[JSONType]]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_update_get_operation_request(
instance_id=self._config.instance_id,
operation_id=operation_id,
api_version=api_version,
if_none_match=if_none_match,
)
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)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 304]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
deserialized = None
response_headers = {}
if response.status_code == 200:
response_headers['Retry-After']=self._deserialize('str', response.headers.get('Retry-After'))
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, response_headers)
return deserialized
get_operation.metadata = {'url': '/deviceupdate/{instanceId}/updates/operations/{operationId}'} # type: ignore
[docs]class DeviceManagementOperations(object):
"""DeviceManagementOperations operations.
You should not instantiate this class directly. Instead, you should create a Client instance that
instantiates it for you and attaches it as an attribute.
:param client: Client for service requests.
:param config: Configuration of service client.
:param serializer: An object model serializer.
:param deserializer: An object model deserializer.
"""
def __init__(self, client, config, serializer, deserializer):
self._client = client
self._serialize = serializer
self._deserialize = deserializer
self._config = config
[docs] @distributed_trace
def list_device_classes(
self,
**kwargs: Any
) -> Iterable[JSONType]:
"""Gets a list of all device classes (unique combinations of device manufacturer and model) for
all devices connected to Device Update for IoT Hub.
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"bestCompatibleUpdateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"compatProperties": {
"str": "str" # Required. The compat properties of the device class. This object can be thought of as a set of key-value pairs where the key is the name of the compatibility property and the value is the value of the compatibility property. There will always be at least 1 compat property.
},
"deviceClassId": "str" # Required. The device class identifier.
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_device_classes_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
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)
else:
request = build_device_management_list_device_classes_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_device_classes.metadata = {'url': '/deviceupdate/{instanceId}/management/deviceclasses'} # type: ignore
[docs] @distributed_trace
def get_device_class(
self,
device_class_id: str,
**kwargs: Any
) -> JSONType:
"""Gets the properties of a device class.
:param device_class_id: Device class identifier.
:type device_class_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"bestCompatibleUpdateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"compatProperties": {
"str": "str" # Required. The compat properties of the device class. This object can be thought of as a set of key-value pairs where the key is the name of the compatibility property and the value is the value of the compatibility property. There will always be at least 1 compat property.
},
"deviceClassId": "str" # Required. The device class identifier.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_device_class_request(
instance_id=self._config.instance_id,
device_class_id=device_class_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_device_class.metadata = {'url': '/deviceupdate/{instanceId}/management/deviceclasses/{deviceClassId}'} # type: ignore
[docs] @distributed_trace
def list_installable_updates_for_device_class(
self,
device_class_id: str,
**kwargs: Any
) -> Iterable[JSONType]:
"""Gets a list of installable updates for a device class.
:param device_class_id: Device class identifier.
:type device_class_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_installable_updates_for_device_class_request(
instance_id=self._config.instance_id,
device_class_id=device_class_id,
api_version=api_version,
)
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)
else:
request = build_device_management_list_installable_updates_for_device_class_request(
instance_id=self._config.instance_id,
device_class_id=device_class_id,
api_version=api_version,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
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
[docs] @distributed_trace
def list_devices(
self,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> Iterable[JSONType]:
"""Gets a list of devices connected to Device Update for IoT Hub.
:keyword filter: Restricts the set of devices returned. You can filter on device GroupId or
DeviceClassId.
:paramtype filter: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"deploymentStatus": "str", # Optional. State of the device in its last deployment. Possible values include: "Succeeded", "InProgress", "Failed", "Canceled", "Incompatible".
"deviceClassId": "str", # Required. Device class identity.
"deviceId": "str", # Required. Device identity.
"groupId": "str", # Optional. Device group identity.
"installedUpdateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"lastAttemptedUpdateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"lastDeploymentId": "str", # Optional. The deployment identifier for the last deployment to the device.
"lastInstallResult": {
"extendedResultCode": 0, # Required. Install extended result code.
"resultCode": 0, # Required. Install result code.
"resultDetails": "str", # Optional. A string containing further details about the install result.
"stepResults": [
{
"description": "str", # Optional. Step description. It might be null for update steps.
"extendedResultCode": 0, # Required. Install extended result code.
"resultCode": 0, # Required. Install result code.
"resultDetails": "str", # Optional. A string containing further details about the install result.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
},
"manufacturer": "str", # Required. Device manufacturer.
"model": "str", # Required. Device model.
"moduleId": "str", # Optional. Device module identity.
"onLatestUpdate": bool # Required. Boolean flag indicating whether the latest update is installed on the device.
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_devices_request(
instance_id=self._config.instance_id,
api_version=api_version,
filter=filter,
)
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)
else:
request = build_device_management_list_devices_request(
instance_id=self._config.instance_id,
api_version=api_version,
filter=filter,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_devices.metadata = {'url': '/deviceupdate/{instanceId}/management/devices'} # type: ignore
def _import_devices_initial(
self,
import_type: str,
*,
action: str,
**kwargs: Any
) -> None:
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', "application/json") # type: Optional[str]
_json = import_type
request = build_device_management_import_devices_request_initial(
instance_id=self._config.instance_id,
api_version=api_version,
content_type=content_type,
action=action,
json=_json,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
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'} # type: ignore
[docs] @distributed_trace
def begin_import_devices(
self,
import_type: str,
*,
action: str,
**kwargs: Any
) -> LROPoller[None]:
"""Import existing devices from IoT Hub.
:param import_type: The types of devices to import. Possible values are: "Devices", "Modules",
and "All".
:type import_type: str
:keyword action: Devices action. "import"
:paramtype action: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
: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 None
:rtype: ~azure.core.polling.LROPoller[None]
:raises: ~azure.core.exceptions.HttpResponseError
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', "application/json") # type: Optional[str]
polling = kwargs.pop('polling', True) # type: Union[bool, azure.core.polling.PollingMethod]
cls = kwargs.pop('cls', None) # type: ClsType[None]
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(
import_type=import_type,
action=action,
api_version=api_version,
content_type=content_type,
cls=lambda x,y,z: x,
**kwargs
)
kwargs.pop('error_map', None)
def get_long_running_output(pipeline_response):
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 = LROBasePolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
elif polling is False: polling_method = 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
)
else:
return LROPoller(self._client, raw_result, get_long_running_output, polling_method)
begin_import_devices.metadata = {'url': '/deviceupdate/{instanceId}/management/devices'} # type: ignore
[docs] @distributed_trace
def get_device(
self,
device_id: str,
**kwargs: Any
) -> JSONType:
"""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.
:type device_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"deploymentStatus": "str", # Optional. State of the device in its last deployment. Possible values include: "Succeeded", "InProgress", "Failed", "Canceled", "Incompatible".
"deviceClassId": "str", # Required. Device class identity.
"deviceId": "str", # Required. Device identity.
"groupId": "str", # Optional. Device group identity.
"installedUpdateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"lastAttemptedUpdateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"lastDeploymentId": "str", # Optional. The deployment identifier for the last deployment to the device.
"lastInstallResult": {
"extendedResultCode": 0, # Required. Install extended result code.
"resultCode": 0, # Required. Install result code.
"resultDetails": "str", # Optional. A string containing further details about the install result.
"stepResults": [
{
"description": "str", # Optional. Step description. It might be null for update steps.
"extendedResultCode": 0, # Required. Install extended result code.
"resultCode": 0, # Required. Install result code.
"resultDetails": "str", # Optional. A string containing further details about the install result.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
},
"manufacturer": "str", # Required. Device manufacturer.
"model": "str", # Required. Device model.
"moduleId": "str", # Optional. Device module identity.
"onLatestUpdate": bool # Required. Boolean flag indicating whether the latest update is installed on the device.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_device_request(
instance_id=self._config.instance_id,
device_id=device_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_device.metadata = {'url': '/deviceupdate/{instanceId}/management/devices/{deviceId}'} # type: ignore
[docs] @distributed_trace
def get_device_module(
self,
device_id: str,
module_id: str,
**kwargs: Any
) -> JSONType:
"""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.
:type device_id: str
:param module_id: Device module identifier in Azure IoT Hub.
:type module_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"deploymentStatus": "str", # Optional. State of the device in its last deployment. Possible values include: "Succeeded", "InProgress", "Failed", "Canceled", "Incompatible".
"deviceClassId": "str", # Required. Device class identity.
"deviceId": "str", # Required. Device identity.
"groupId": "str", # Optional. Device group identity.
"installedUpdateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"lastAttemptedUpdateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
},
"lastDeploymentId": "str", # Optional. The deployment identifier for the last deployment to the device.
"lastInstallResult": {
"extendedResultCode": 0, # Required. Install extended result code.
"resultCode": 0, # Required. Install result code.
"resultDetails": "str", # Optional. A string containing further details about the install result.
"stepResults": [
{
"description": "str", # Optional. Step description. It might be null for update steps.
"extendedResultCode": 0, # Required. Install extended result code.
"resultCode": 0, # Required. Install result code.
"resultDetails": "str", # Optional. A string containing further details about the install result.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
},
"manufacturer": "str", # Required. Device manufacturer.
"model": "str", # Required. Device model.
"moduleId": "str", # Optional. Device module identity.
"onLatestUpdate": bool # Required. Boolean flag indicating whether the latest update is installed on the device.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_device_module_request(
instance_id=self._config.instance_id,
device_id=device_id,
module_id=module_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_device_module.metadata = {'url': '/deviceupdate/{instanceId}/management/devices/{deviceId}/modules/{moduleId}'} # type: ignore
[docs] @distributed_trace
def get_update_compliance(
self,
**kwargs: Any
) -> JSONType:
"""Gets the breakdown of how many devices are on their latest update, have new updates available,
or are in progress receiving new updates.
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"newUpdatesAvailableDeviceCount": 0, # Required. Number of devices with a newer update available.
"onLatestUpdateDeviceCount": 0, # Required. Number of devices on the latest update.
"totalDeviceCount": 0, # Required. Total number of devices.
"updatesInProgressDeviceCount": 0 # Required. Number of devices with update in-progress.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_update_compliance_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_update_compliance.metadata = {'url': '/deviceupdate/{instanceId}/management/updatecompliance'} # type: ignore
list_device_tags.metadata = {'url': '/deviceupdate/{instanceId}/management/devicetags'} # type: ignore
[docs] @distributed_trace
def get_device_tag(
self,
tag_name: str,
**kwargs: Any
) -> JSONType:
"""Gets a count of how many devices have a device tag.
:param tag_name: Tag name.
:type tag_name: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"deviceCount": 0, # Required. Number of devices with this tag.
"tagName": "str" # Required. Tag name.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_device_tag_request(
instance_id=self._config.instance_id,
tag_name=tag_name,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_device_tag.metadata = {'url': '/deviceupdate/{instanceId}/management/devicetags/{tagName}'} # type: ignore
[docs] @distributed_trace
def list_groups(
self,
**kwargs: Any
) -> Iterable[JSONType]:
"""Gets a list of all device groups.
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"createdDateTime": "str", # Required. Date and time when the update was created.
"deploymentId": "str", # Optional. The deployment Id for the group.
"deviceClassId": "str", # Optional. The device class Id for the group.
"deviceCount": 0, # Optional. The number of devices in the group.
"groupId": "str", # Required. Group identity.
"groupType": "str", # Required. Group type. Possible values include: "DeviceClassIdAndIoTHubTag", "InvalidDeviceClassIdAndIoTHubTag", "DefaultDeviceClassId".
"tags": [
"str" # Required. A set of tags. IoT Hub tags.
]
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_groups_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
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)
else:
request = build_device_management_list_groups_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_groups.metadata = {'url': '/deviceupdate/{instanceId}/management/groups'} # type: ignore
[docs] @distributed_trace
def get_group(
self,
group_id: str,
**kwargs: Any
) -> JSONType:
"""Gets the properties of a group.
:param group_id: Group identity.
:type group_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"createdDateTime": "str", # Required. Date and time when the update was created.
"deploymentId": "str", # Optional. The deployment Id for the group.
"deviceClassId": "str", # Optional. The device class Id for the group.
"deviceCount": 0, # Optional. The number of devices in the group.
"groupId": "str", # Required. Group identity.
"groupType": "str", # Required. Group type. Possible values include: "DeviceClassIdAndIoTHubTag", "InvalidDeviceClassIdAndIoTHubTag", "DefaultDeviceClassId".
"tags": [
"str" # Required. A set of tags. IoT Hub tags.
]
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_group_request(
instance_id=self._config.instance_id,
group_id=group_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_group.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}'} # type: ignore
[docs] @distributed_trace
def create_or_update_group(
self,
group_id: str,
group: JSONType,
**kwargs: Any
) -> JSONType:
"""Create or update a device group.
:param group_id: Group identity.
:type group_id: str
:param group: The group properties.
:type group: JSONType
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# JSON input template you can fill out and use as your body input.
group = {
"createdDateTime": "str", # Required. Date and time when the update was created.
"deploymentId": "str", # Optional. The deployment Id for the group.
"deviceClassId": "str", # Optional. The device class Id for the group.
"deviceCount": 0, # Optional. The number of devices in the group.
"groupId": "str", # Required. Group identity.
"groupType": "str", # Required. Group type. Possible values include: "DeviceClassIdAndIoTHubTag", "InvalidDeviceClassIdAndIoTHubTag", "DefaultDeviceClassId".
"tags": [
"str" # Required. A set of tags. IoT Hub tags.
]
}
# response body for status code(s): 200
response.json() == {
"createdDateTime": "str", # Required. Date and time when the update was created.
"deploymentId": "str", # Optional. The deployment Id for the group.
"deviceClassId": "str", # Optional. The device class Id for the group.
"deviceCount": 0, # Optional. The number of devices in the group.
"groupId": "str", # Required. Group identity.
"groupType": "str", # Required. Group type. Possible values include: "DeviceClassIdAndIoTHubTag", "InvalidDeviceClassIdAndIoTHubTag", "DefaultDeviceClassId".
"tags": [
"str" # Required. A set of tags. IoT Hub tags.
]
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', "application/json") # type: Optional[str]
_json = group
request = build_device_management_create_or_update_group_request(
instance_id=self._config.instance_id,
group_id=group_id,
api_version=api_version,
content_type=content_type,
json=_json,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
create_or_update_group.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}'} # type: ignore
[docs] @distributed_trace
def delete_group(
self,
group_id: str,
**kwargs: Any
) -> None:
"""Deletes a device group.
:param group_id: Group identity.
:type group_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_delete_group_request(
instance_id=self._config.instance_id,
group_id=group_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
delete_group.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}'} # type: ignore
[docs] @distributed_trace
def get_group_update_compliance(
self,
group_id: str,
**kwargs: Any
) -> JSONType:
"""Get 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 identity.
:type group_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"newUpdatesAvailableDeviceCount": 0, # Required. Number of devices with a newer update available.
"onLatestUpdateDeviceCount": 0, # Required. Number of devices on the latest update.
"totalDeviceCount": 0, # Required. Total number of devices.
"updatesInProgressDeviceCount": 0 # Required. Number of devices with update in-progress.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_group_update_compliance_request(
instance_id=self._config.instance_id,
group_id=group_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_group_update_compliance.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/updateCompliance'} # type: ignore
[docs] @distributed_trace
def list_best_updates_for_group(
self,
group_id: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get the best available updates for a group and a count of how many devices need each update.
:param group_id: Group identity.
:type group_id: str
:keyword filter: Restricts the set of bestUpdates returned. You can filter on update Provider,
Name and Version property.
:paramtype filter: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"deviceCount": 0, # Required. Total number of devices for which the update is applicable.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_best_updates_for_group_request(
instance_id=self._config.instance_id,
group_id=group_id,
api_version=api_version,
filter=filter,
)
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)
else:
request = build_device_management_list_best_updates_for_group_request(
instance_id=self._config.instance_id,
group_id=group_id,
api_version=api_version,
filter=filter,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_best_updates_for_group.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/bestUpdates'} # type: ignore
[docs] @distributed_trace
def list_deployments_for_group(
self,
group_id: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> Iterable[JSONType]:
"""Gets a list of deployments for a group.
:param group_id: Group identity.
:type group_id: str
:keyword filter: Restricts the set of deployments returned. You can filter on update Provider,
Name and Version property.
:paramtype filter: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"deploymentId": "str", # Required. The deployment identifier.
"groupId": "str", # Required. The group identity.
"isCanceled": bool, # Optional. Boolean flag indicating whether the deployment was canceled.
"isRetried": bool, # Optional. Boolean flag indicating whether the deployment has been retried.
"startDateTime": "2020-02-20 00:00:00", # Required. The deployment start datetime.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_deployments_for_group_request(
instance_id=self._config.instance_id,
group_id=group_id,
api_version=api_version,
filter=filter,
)
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)
else:
request = build_device_management_list_deployments_for_group_request(
instance_id=self._config.instance_id,
group_id=group_id,
api_version=api_version,
filter=filter,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_deployments_for_group.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments'} # type: ignore
[docs] @distributed_trace
def get_deployment(
self,
group_id: str,
deployment_id: str,
**kwargs: Any
) -> JSONType:
"""Gets the properties of a deployment.
:param group_id: Group identity.
:type group_id: str
:param deployment_id: Deployment identifier.
:type deployment_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"deploymentId": "str", # Required. The deployment identifier.
"groupId": "str", # Required. The group identity.
"isCanceled": bool, # Optional. Boolean flag indicating whether the deployment was canceled.
"isRetried": bool, # Optional. Boolean flag indicating whether the deployment has been retried.
"startDateTime": "2020-02-20 00:00:00", # Required. The deployment start datetime.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_deployment_request(
instance_id=self._config.instance_id,
group_id=group_id,
deployment_id=deployment_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_deployment.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}'} # type: ignore
[docs] @distributed_trace
def create_or_update_deployment(
self,
deployment_id: str,
group_id: str,
deployment: JSONType,
**kwargs: Any
) -> JSONType:
"""Creates or updates a deployment.
:param deployment_id: Deployment identifier.
:type deployment_id: str
:param group_id: Group identity.
:type group_id: str
:param deployment: The deployment properties.
:type deployment: JSONType
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# JSON input template you can fill out and use as your body input.
deployment = {
"deploymentId": "str", # Required. The deployment identifier.
"groupId": "str", # Required. The group identity.
"isCanceled": bool, # Optional. Boolean flag indicating whether the deployment was canceled.
"isRetried": bool, # Optional. Boolean flag indicating whether the deployment has been retried.
"startDateTime": "2020-02-20 00:00:00", # Required. The deployment start datetime.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
# response body for status code(s): 200
response.json() == {
"deploymentId": "str", # Required. The deployment identifier.
"groupId": "str", # Required. The group identity.
"isCanceled": bool, # Optional. Boolean flag indicating whether the deployment was canceled.
"isRetried": bool, # Optional. Boolean flag indicating whether the deployment has been retried.
"startDateTime": "2020-02-20 00:00:00", # Required. The deployment start datetime.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', "application/json") # type: Optional[str]
_json = deployment
request = build_device_management_create_or_update_deployment_request(
instance_id=self._config.instance_id,
deployment_id=deployment_id,
group_id=group_id,
api_version=api_version,
content_type=content_type,
json=_json,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
create_or_update_deployment.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}'} # type: ignore
[docs] @distributed_trace
def delete_deployment(
self,
group_id: str,
deployment_id: str,
**kwargs: Any
) -> None:
"""Deletes a deployment.
:param group_id: Group identity.
:type group_id: str
:param deployment_id: Deployment identifier.
:type deployment_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: None
:rtype: None
:raises: ~azure.core.exceptions.HttpResponseError
"""
cls = kwargs.pop('cls', None) # type: ClsType[None]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_delete_deployment_request(
instance_id=self._config.instance_id,
group_id=group_id,
deployment_id=deployment_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if cls:
return cls(pipeline_response, None, {})
delete_deployment.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}'} # type: ignore
[docs] @distributed_trace
def get_deployment_status(
self,
group_id: str,
deployment_id: str,
**kwargs: Any
) -> JSONType:
"""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 identity.
:type group_id: str
:param deployment_id: Deployment identifier.
:type deployment_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"deploymentState": "str", # Required. The state of the deployment. Possible values include: "Active", "Inactive", "Canceled".
"devicesCanceledCount": 0, # Optional. The number of devices which have had their deployment canceled.
"devicesCompletedFailedCount": 0, # Optional. The number of devices that have completed deployment with a failure.
"devicesCompletedSucceededCount": 0, # Optional. The number of devices which have successfully completed deployment.
"devicesInProgressCount": 0, # Optional. The number of devices that are currently in deployment.
"totalDevices": 0 # Optional. The total number of devices in the deployment.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_deployment_status_request(
instance_id=self._config.instance_id,
group_id=group_id,
deployment_id=deployment_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_deployment_status.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}/status'} # type: ignore
[docs] @distributed_trace
def list_deployment_devices(
self,
group_id: str,
deployment_id: str,
*,
filter: Optional[str] = None,
**kwargs: Any
) -> Iterable[JSONType]:
"""Gets a list of devices in a deployment along with their state. Useful for getting a list of
failed devices.
:param group_id: Group identity.
:type group_id: str
:param deployment_id: Deployment identifier.
:type deployment_id: str
:keyword filter: Restricts the set of deployment device states returned. You can filter on
deviceId and moduleId and/or deviceState.
:paramtype filter: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"deviceId": "str", # Required. Device identity.
"deviceState": "str", # Required. Deployment device state. Possible values include: "Succeeded", "InProgress", "Failed", "Canceled", "Incompatible".
"moduleId": "str", # Optional. Device module identity.
"movedOnToNewDeployment": bool, # Required. Boolean flag indicating whether this device is in a newer deployment and can no longer retry this deployment.
"retryCount": 0 # Required. The number of times this deployment has been retried on this device.
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_deployment_devices_request(
instance_id=self._config.instance_id,
group_id=group_id,
deployment_id=deployment_id,
api_version=api_version,
filter=filter,
)
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)
else:
request = build_device_management_list_deployment_devices_request(
instance_id=self._config.instance_id,
group_id=group_id,
deployment_id=deployment_id,
api_version=api_version,
filter=filter,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_deployment_devices.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}/devicestates'} # type: ignore
[docs] @distributed_trace
def get_operation(
self,
operation_id: str,
*,
if_none_match: Optional[str] = None,
**kwargs: Any
) -> Optional[JSONType]:
"""Retrieve operation status.
:param operation_id: Operation identifier.
:type operation_id: str
:keyword if_none_match: Defines the If-None-Match condition. The operation will be performed
only if the ETag on the server does not match this value.
:paramtype if_none_match: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType or None
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"createdDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the operation was created.
"error": {
"code": "str", # Required. Server defined error code.
"details": [
...
],
"innererror": {
"code": "str", # Required. A more specific error code than what was provided by the containing error.
"errorDetail": "str", # Optional. The internal error or exception message.
"innerError": ...,
"message": "str" # Optional. A human-readable representation of the error.
},
"message": "str", # Required. A human-readable representation of the error.
"occurredDateTime": "2020-02-20 00:00:00", # Optional. Date and time in UTC when the error occurred.
"target": "str" # Optional. The target of the error.
},
"etag": "str", # Optional. Operation ETag.
"lastActionDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the operation status was last updated.
"operationId": "str", # Required. Operation Id.
"status": "str", # Required. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
"traceId": "str" # Optional. Operation correlation identity that can used by Microsoft Support for troubleshooting.
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[Optional[JSONType]]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_operation_request(
instance_id=self._config.instance_id,
operation_id=operation_id,
api_version=api_version,
if_none_match=if_none_match,
)
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)
pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200, 304]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response)
deserialized = None
response_headers = {}
if response.status_code == 200:
response_headers['Retry-After']=self._deserialize('str', response.headers.get('Retry-After'))
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, response_headers)
return deserialized
get_operation.metadata = {'url': '/deviceupdate/{instanceId}/management/operations/{operationId}'} # type: ignore
[docs] @distributed_trace
def list_operations(
self,
*,
filter: Optional[str] = None,
top: Optional[int] = None,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get a list of all device import operations. Completed operations are kept for 7 days before
auto-deleted.
:keyword filter: Restricts the set of operations returned. Only one specific filter is
supported: "status eq 'NotStarted' or status eq 'Running'".
:paramtype filter: str
:keyword 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.
:paramtype top: int
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"createdDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the operation was created.
"error": {
"code": "str", # Required. Server defined error code.
"details": [
...
],
"innererror": {
"code": "str", # Required. A more specific error code than what was provided by the containing error.
"errorDetail": "str", # Optional. The internal error or exception message.
"innerError": ...,
"message": "str" # Optional. A human-readable representation of the error.
},
"message": "str", # Required. A human-readable representation of the error.
"occurredDateTime": "2020-02-20 00:00:00", # Optional. Date and time in UTC when the error occurred.
"target": "str" # Optional. The target of the error.
},
"etag": "str", # Optional. Operation ETag.
"lastActionDateTime": "2020-02-20 00:00:00", # Required. Date and time in UTC when the operation status was last updated.
"operationId": "str", # Required. Operation Id.
"status": "str", # Required. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
"traceId": "str" # Optional. Operation correlation identity that can used by Microsoft Support for troubleshooting.
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_operations_request(
instance_id=self._config.instance_id,
api_version=api_version,
filter=filter,
top=top,
)
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)
else:
request = build_device_management_list_operations_request(
instance_id=self._config.instance_id,
api_version=api_version,
filter=filter,
top=top,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_operations.metadata = {'url': '/deviceupdate/{instanceId}/management/operations'} # type: ignore
[docs] @distributed_trace
def collect_logs(
self,
operation_id: str,
log_collection_request: JSONType,
**kwargs: Any
) -> JSONType:
"""Start the device diagnostics log collection operation on specified devices.
:param operation_id: Operation identifier.
:type operation_id: str
:param log_collection_request: The deployment properties.
:type log_collection_request: JSONType
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# JSON input template you can fill out and use as your body input.
log_collection_request = {
"createdDateTime": "str", # Optional. The timestamp when the operation was created.
"description": "str", # Optional. Description of the diagnostics operation.
"deviceList": [
{
"deviceId": "str", # Required. Device Id.
"moduleId": "str" # Optional. Module Id.
}
],
"lastActionDateTime": "str", # Optional. A timestamp for when the current state was entered.
"operationId": "str", # Optional. The diagnostics operation id.
"status": "str" # Optional. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
}
# response body for status code(s): 201
response.json() == {
"createdDateTime": "str", # Optional. The timestamp when the operation was created.
"description": "str", # Optional. Description of the diagnostics operation.
"deviceList": [
{
"deviceId": "str", # Required. Device Id.
"moduleId": "str" # Optional. Module Id.
}
],
"lastActionDateTime": "str", # Optional. A timestamp for when the current state was entered.
"operationId": "str", # Optional. The diagnostics operation id.
"status": "str" # Optional. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
content_type = kwargs.pop('content_type', "application/json") # type: Optional[str]
_json = log_collection_request
request = build_device_management_collect_logs_request(
instance_id=self._config.instance_id,
operation_id=operation_id,
api_version=api_version,
content_type=content_type,
json=_json,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
collect_logs.metadata = {'url': '/deviceupdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}'} # type: ignore
[docs] @distributed_trace
def get_log_collection_operation(
self,
operation_id: str,
**kwargs: Any
) -> JSONType:
"""Get the device diagnostics log collection operation.
:param operation_id: Operation identifier.
:type operation_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"createdDateTime": "str", # Optional. The timestamp when the operation was created.
"description": "str", # Optional. Description of the diagnostics operation.
"deviceList": [
{
"deviceId": "str", # Required. Device Id.
"moduleId": "str" # Optional. Module Id.
}
],
"lastActionDateTime": "str", # Optional. A timestamp for when the current state was entered.
"operationId": "str", # Optional. The diagnostics operation id.
"status": "str" # Optional. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_log_collection_operation_request(
instance_id=self._config.instance_id,
operation_id=operation_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_log_collection_operation.metadata = {'url': '/deviceupdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}'} # type: ignore
[docs] @distributed_trace
def list_log_collection_operations(
self,
**kwargs: Any
) -> Iterable[JSONType]:
"""Get all device diagnostics log collection operations.
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: An iterator like instance of JSON object
:rtype: ~azure.core.paging.ItemPaged[JSONType]
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"nextLink": "str", # Optional. The link to the next page of items.
"value": [
{
"createdDateTime": "str", # Optional. The timestamp when the operation was created.
"description": "str", # Optional. Description of the diagnostics operation.
"deviceList": [
{
"deviceId": "str", # Required. Device Id.
"moduleId": "str" # Optional. Module Id.
}
],
"lastActionDateTime": "str", # Optional. A timestamp for when the current state was entered.
"operationId": "str", # Optional. The diagnostics operation id.
"status": "str" # Optional. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
}
]
}
"""
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
def prepare_request(next_link=None):
if not next_link:
request = build_device_management_list_log_collection_operations_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
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)
else:
request = build_device_management_list_log_collection_operations_request(
instance_id=self._config.instance_id,
api_version=api_version,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True),
}
request.url = self._client.format_url(next_link, **path_format_arguments)
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 = _loads(pipeline_response.http_response.body())
list_of_elem = deserialized["value"]
if cls:
list_of_elem = cls(list_of_elem)
return deserialized.get("nextLink", None), iter(list_of_elem)
def get_next(next_link=None):
request = prepare_request(next_link)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
return pipeline_response
return ItemPaged(
get_next, extract_data
)
list_log_collection_operations.metadata = {'url': '/deviceupdate/{instanceId}/management/deviceDiagnostics/logCollections'} # type: ignore
[docs] @distributed_trace
def get_log_collection_operation_detailed_status(
self,
operation_id: str,
**kwargs: Any
) -> JSONType:
"""Get device diagnostics log collection operation with detailed status.
:param operation_id: Operation identifier.
:type operation_id: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"createdDateTime": "str", # Optional. The timestamp when the operation was created.
"description": "str", # Optional. Device diagnostics operation description.
"deviceStatus": [
{
"deviceId": "str", # Required. Device id.
"extendedResultCode": "str", # Optional. Log upload extended result code.
"logLocation": "str", # Optional. Log upload location.
"moduleId": "str", # Optional. Module id.
"resultCode": "str", # Optional. Log upload result code.
"status": "str" # Required. Log upload status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
}
],
"lastActionDateTime": "str", # Optional. A timestamp for when the current state was entered.
"operationId": "str", # Optional. The device diagnostics operation id.
"status": "str" # Optional. Operation status. Possible values include: "Undefined", "NotStarted", "Running", "Succeeded", "Failed".
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_get_log_collection_operation_detailed_status_request(
instance_id=self._config.instance_id,
operation_id=operation_id,
api_version=api_version,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
get_log_collection_operation_detailed_status.metadata = {'url': '/deviceupdate/{instanceId}/management/deviceDiagnostics/logCollections/{operationId}/detailedStatus'} # type: ignore
[docs] @distributed_trace
def stop_deployment(
self,
group_id: str,
deployment_id: str,
*,
action: str,
**kwargs: Any
) -> JSONType:
"""Stops a deployment.
:param group_id: Group identity.
:type group_id: str
:param deployment_id: Deployment identifier.
:type deployment_id: str
:keyword action: Cancel deployment action. "cancel"
:paramtype action: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"deploymentId": "str", # Required. The deployment identifier.
"groupId": "str", # Required. The group identity.
"isCanceled": bool, # Optional. Boolean flag indicating whether the deployment was canceled.
"isRetried": bool, # Optional. Boolean flag indicating whether the deployment has been retried.
"startDateTime": "2020-02-20 00:00:00", # Required. The deployment start datetime.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_stop_deployment_request(
instance_id=self._config.instance_id,
group_id=group_id,
deployment_id=deployment_id,
api_version=api_version,
action=action,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
stop_deployment.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}'} # type: ignore
[docs] @distributed_trace
def retry_deployment(
self,
group_id: str,
deployment_id: str,
*,
action: str,
**kwargs: Any
) -> JSONType:
"""Retries a deployment with failed devices.
:param group_id: Group identity.
:type group_id: str
:param deployment_id: Deployment identifier.
:type deployment_id: str
:keyword action: Retry deployment action. "retry"
:paramtype action: str
:keyword api_version: Api Version. The default value is "2021-06-01-preview". Note that
overriding this default value may result in unsupported behavior.
:paramtype api_version: str
:return: JSON object
:rtype: JSONType
:raises: ~azure.core.exceptions.HttpResponseError
Example:
.. code-block:: python
# response body for status code(s): 200
response.json() == {
"deploymentId": "str", # Required. The deployment identifier.
"groupId": "str", # Required. The group identity.
"isCanceled": bool, # Optional. Boolean flag indicating whether the deployment was canceled.
"isRetried": bool, # Optional. Boolean flag indicating whether the deployment has been retried.
"startDateTime": "2020-02-20 00:00:00", # Required. The deployment start datetime.
"updateId": {
"name": "str", # Required. Update name.
"provider": "str", # Required. Update provider.
"version": "str" # Required. Update version.
}
}
"""
cls = kwargs.pop('cls', None) # type: ClsType[JSONType]
error_map = {
401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
}
error_map.update(kwargs.pop('error_map', {}))
api_version = kwargs.pop('api_version', "2021-06-01-preview") # type: str
request = build_device_management_retry_deployment_request(
instance_id=self._config.instance_id,
group_id=group_id,
deployment_id=deployment_id,
api_version=api_version,
action=action,
)
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)
pipeline_response = self._client._pipeline.run(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)
raise HttpResponseError(response=response)
if response.content:
deserialized = response.json()
else:
deserialized = None
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
retry_deployment.metadata = {'url': '/deviceupdate/{instanceId}/management/groups/{groupId}/deployments/{deploymentId}'} # type: ignore