# pylint: disable=too-many-lines # coding=utf-8 # -------------------------------------------------------------------------- # Code generated by Microsoft (R) AutoRest Code Generator (autorest: 3.9.2, generator: @autorest/python@5.19.0) # Changes may cause incorrect behavior and will be lost if the code is regenerated. # -------------------------------------------------------------------------- from typing import Any, Callable, Dict, IO, Iterable, List, Optional, TypeVar, Union, cast, overload from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error from azure.core.paging import ItemPaged from azure.core.pipeline import PipelineResponse from azure.core.pipeline.transport import HttpResponse from azure.core.polling import LROPoller, NoPolling, PollingMethod from azure.core.polling.base_polling import LROBasePolling from azure.core.rest import HttpRequest from azure.core.tracing.decorator import distributed_trace from azure.core.utils import case_insensitive_dict from .. import models as _models from .._serialization import Serializer from .._vendor import _convert_request, _format_url_section T = TypeVar('T') ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]] _SERIALIZER = Serializer() _SERIALIZER.client_side_validation = False def build_list_updates_request( instance_id: str, *, search: Optional[str] = None, filter: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') if search is not None: _params['search'] = _SERIALIZER.query("search", search, 'str') if filter is not None: _params['filter'] = _SERIALIZER.query("filter", filter, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_import_update_request( instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/updates:import") path_format_arguments = { "instanceId": _SERIALIZER.url("instance_id", instance_id, 'str', skip_quote=True), } _url = _format_url_section(_url, **path_format_arguments) # Construct parameters _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if content_type is not None: _headers['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="POST", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_update_request( provider: str, name: str, version: str, instance_id: str, *, if_none_match: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}") # pylint: disable=line-too-long 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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if if_none_match is not None: _headers['If-None-Match'] = _SERIALIZER.header("if_none_match", if_none_match, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_delete_update_request( provider: str, name: str, version: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}") # pylint: disable=line-too-long 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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="DELETE", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_providers_request( instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_names_request( provider: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_versions_request( provider: str, name: str, instance_id: str, *, filter: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') if filter is not None: _params['filter'] = _SERIALIZER.query("filter", filter, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_files_request( provider: str, name: str, version: str, instance_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}/files") # pylint: disable=line-too-long 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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_file_request( provider: str, name: str, version: str, file_id: str, instance_id: str, *, if_none_match: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}/files/{fileId}") # pylint: disable=line-too-long 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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if if_none_match is not None: _headers['If-None-Match'] = _SERIALIZER.header("if_none_match", if_none_match, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_list_operation_statuses_request( instance_id: str, *, filter: Optional[str] = None, top: Optional[int] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/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 if filter is not None: _params['filter'] = _SERIALIZER.query("filter", filter, 'str') if top is not None: _params['top'] = _SERIALIZER.query("top", top, 'int') _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) def build_get_operation_status_request( operation_id: str, instance_id: str, *, if_none_match: Optional[str] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', "2022-10-01")) # type: str accept = _headers.pop('Accept', "application/json") # Construct URL _url = kwargs.pop("template_url", "/deviceUpdate/{instanceId}/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 _params['api-version'] = _SERIALIZER.query("api_version", api_version, 'str') # Construct headers if if_none_match is not None: _headers['If-None-Match'] = _SERIALIZER.header("if_none_match", if_none_match, 'str') _headers['Accept'] = _SERIALIZER.header("accept", accept, 'str') return HttpRequest( method="GET", url=_url, params=_params, headers=_headers, **kwargs ) class DeviceUpdateOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~deviceupdateclient.DeviceUpdateClient`'s :attr:`device_update` attribute. """ models = _models def __init__(self, *args, **kwargs): input_args = list(args) self._client = input_args.pop(0) if input_args else kwargs.pop("client") self._config = input_args.pop(0) if input_args else kwargs.pop("config") self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace def list_updates( self, search: Optional[str] = None, filter: Optional[str] = None, **kwargs: Any ) -> Iterable["_models.Update"]: """Get a list of all updates that have been imported to Device Update for IoT Hub. :param search: Request updates matching a free-text search expression. Default value is None. :type search: str :param filter: Optional to filter updates by isDeployable property. Default value is None. :type filter: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either Update or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.Update] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.UpdateList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_updates_request( instance_id=self._config.instance_id, search=search, filter=filter, api_version=api_version, template_url=self.list_updates.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_updates_request( instance_id=self._config.instance_id, search=search, filter=filter, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("UpdateList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_updates.metadata = {'url': "/deviceUpdate/{instanceId}/updates"} # type: ignore def _import_update_initial( # pylint: disable=inconsistent-return-statements self, update_to_import: Union[List[_models.ImportUpdateInputItem], IO], **kwargs: Any ) -> None: error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] cls = kwargs.pop('cls', None) # type: ClsType[None] content_type = content_type or "application/json" _json = None _content = None if isinstance(update_to_import, (IO, bytes)): _content = update_to_import else: _json = self._serialize.body(update_to_import, '[ImportUpdateInputItem]') request = build_import_update_request( instance_id=self._config.instance_id, api_version=api_version, content_type=content_type, json=_json, content=_content, template_url=self._import_update_initial.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [202]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers['Operation-Location']=self._deserialize('str', response.headers.get('Operation-Location')) if cls: return cls(pipeline_response, None, response_headers) _import_update_initial.metadata = {'url': "/deviceUpdate/{instanceId}/updates:import"} # type: ignore @overload def begin_import_update( self, update_to_import: List[_models.ImportUpdateInputItem], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[None]: """Import new update version. This is a long-running-operation; use Operation-Location response header value to check for operation status. :param update_to_import: The update to be imported (see schema https://json.schemastore.org/azure-deviceupdate-import-manifest-5.0.json for details). Required. :type update_to_import: list[~deviceupdateclient.models.ImportUpdateInputItem] :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :keyword str continuation_token: A continuation token to restart a poller from a saved state. :keyword polling: By default, your polling method will be LROBasePolling. Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. :paramtype polling: bool or ~azure.core.polling.PollingMethod :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_import_update( self, update_to_import: IO, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[None]: """Import new update version. This is a long-running-operation; use Operation-Location response header value to check for operation status. :param update_to_import: The update to be imported (see schema https://json.schemastore.org/azure-deviceupdate-import-manifest-5.0.json for details). Required. :type update_to_import: IO :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :keyword str continuation_token: A continuation token to restart a poller from a saved state. :keyword polling: By default, your polling method will be LROBasePolling. Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. :paramtype polling: bool or ~azure.core.polling.PollingMethod :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_import_update( self, update_to_import: Union[List[_models.ImportUpdateInputItem], IO], **kwargs: Any ) -> LROPoller[None]: """Import new update version. This is a long-running-operation; use Operation-Location response header value to check for operation status. :param update_to_import: The update to be imported (see schema https://json.schemastore.org/azure-deviceupdate-import-manifest-5.0.json for details). Is either a list type or a IO type. Required. :type update_to_import: list[~deviceupdateclient.models.ImportUpdateInputItem] or IO :keyword content_type: Body Parameter content-type. Known values are: 'application/json'. Default value is None. :paramtype content_type: str :keyword callable cls: A custom type or function that will be passed the direct response :keyword str continuation_token: A continuation token to restart a poller from a saved state. :keyword polling: By default, your polling method will be LROBasePolling. Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. :paramtype polling: bool or ~azure.core.polling.PollingMethod :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str content_type = kwargs.pop('content_type', _headers.pop('Content-Type', None)) # type: Optional[str] cls = kwargs.pop('cls', None) # type: ClsType[None] polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] lro_delay = kwargs.pop( 'polling_interval', self._config.polling_interval ) cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] if cont_token is None: raw_result = self._import_update_initial( # type: ignore update_to_import=update_to_import, api_version=api_version, content_type=content_type, cls=lambda x,y,z: x, headers=_headers, params=_params, **kwargs ) kwargs.pop('error_map', None) def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements if cls: return cls(pipeline_response, None, {}) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } if polling is True: polling_method = cast(PollingMethod, LROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs )) # type: PollingMethod elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller.from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output ) return LROPoller(self._client, raw_result, get_long_running_output, polling_method) begin_import_update.metadata = {'url': "/deviceUpdate/{instanceId}/updates:import"} # type: ignore @distributed_trace def get_update( self, provider: str, name: str, version: str, access_condition: Optional[_models.AccessCondition] = None, **kwargs: Any ) -> _models.Update: """Get a specific update version. :param provider: Update provider. Required. :type provider: str :param name: Update name. Required. :type name: str :param version: Update version. Required. :type version: str :param access_condition: Parameter group. Default value is None. :type access_condition: ~deviceupdateclient.models.AccessCondition :keyword callable cls: A custom type or function that will be passed the direct response :return: Update or the result of cls(response) :rtype: ~deviceupdateclient.models.Update :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.Update] _if_none_match = None if access_condition is not None: _if_none_match = access_condition.if_none_match request = build_get_update_request( provider=provider, name=name, version=version, instance_id=self._config.instance_id, if_none_match=_if_none_match, api_version=api_version, template_url=self.get_update.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('Update', pipeline_response) 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( # pylint: disable=inconsistent-return-statements self, provider: str, name: str, version: str, **kwargs: Any ) -> None: error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] request = build_delete_update_request( provider=provider, name=name, version=version, instance_id=self._config.instance_id, api_version=api_version, template_url=self._delete_update_initial.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [202]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers['Operation-Location']=self._deserialize('str', response.headers.get('Operation-Location')) if cls: return cls(pipeline_response, None, response_headers) _delete_update_initial.metadata = {'url': "/deviceUpdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}"} # type: ignore @distributed_trace def begin_delete_update( self, provider: str, name: str, version: str, **kwargs: Any ) -> LROPoller[None]: """Delete a specific update version. This is a long-running-operation; use Operation-Location response header value to check for operation status. :param provider: Update provider. Required. :type provider: str :param name: Update name. Required. :type name: str :param version: Update version. Required. :type version: str :keyword callable cls: A custom type or function that will be passed the direct response :keyword str continuation_token: A continuation token to restart a poller from a saved state. :keyword polling: By default, your polling method will be LROBasePolling. Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. :paramtype polling: bool or ~azure.core.polling.PollingMethod :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[None] polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] lro_delay = kwargs.pop( 'polling_interval', self._config.polling_interval ) cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] if cont_token is None: raw_result = self._delete_update_initial( # type: ignore provider=provider, name=name, version=version, api_version=api_version, cls=lambda x,y,z: x, headers=_headers, params=_params, **kwargs ) kwargs.pop('error_map', None) def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements if cls: return cls(pipeline_response, None, {}) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } if polling is True: polling_method = cast(PollingMethod, LROBasePolling( lro_delay, path_format_arguments=path_format_arguments, **kwargs )) # type: PollingMethod elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller.from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output ) return LROPoller(self._client, raw_result, get_long_running_output, polling_method) begin_delete_update.metadata = {'url': "/deviceUpdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}"} # type: ignore @distributed_trace def list_providers( self, **kwargs: Any ) -> Iterable[str]: """Get a list of all update providers that have been imported to Device Update for IoT Hub. :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either str or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[str] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.StringsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_providers_request( instance_id=self._config.instance_id, api_version=api_version, template_url=self.list_providers.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_providers_request( instance_id=self._config.instance_id, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("StringsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_providers.metadata = {'url': "/deviceUpdate/{instanceId}/updates/providers"} # type: ignore @distributed_trace def list_names( self, provider: str, **kwargs: Any ) -> Iterable[str]: """Get a list of all update names that match the specified provider. :param provider: Update provider. Required. :type provider: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either str or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[str] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.StringsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_names_request( provider=provider, instance_id=self._config.instance_id, api_version=api_version, template_url=self.list_names.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_names_request( provider=provider, instance_id=self._config.instance_id, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("StringsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_names.metadata = {'url': "/deviceUpdate/{instanceId}/updates/providers/{provider}/names"} # type: ignore @distributed_trace def list_versions( self, provider: str, name: str, filter: Optional[str] = None, **kwargs: Any ) -> Iterable[str]: """Get a list of all update versions that match the specified provider and name. :param provider: Update provider. Required. :type provider: str :param name: Update name. Required. :type name: str :param filter: Optional to filter updates by isDeployable property. Default value is None. :type filter: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either str or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[str] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.StringsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_versions_request( provider=provider, name=name, instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=self.list_versions.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_versions_request( provider=provider, name=name, instance_id=self._config.instance_id, filter=filter, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("StringsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_versions.metadata = {'url': "/deviceUpdate/{instanceId}/updates/providers/{provider}/names/{name}/versions"} # type: ignore @distributed_trace def list_files( self, provider: str, name: str, version: str, **kwargs: Any ) -> Iterable[str]: """Get a list of all update file identifiers for the specified version. :param provider: Update provider. Required. :type provider: str :param name: Update name. Required. :type name: str :param version: Update version. Required. :type version: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either str or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[str] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.StringsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_files_request( provider=provider, name=name, version=version, instance_id=self._config.instance_id, api_version=api_version, template_url=self.list_files.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_files_request( provider=provider, name=name, version=version, instance_id=self._config.instance_id, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("StringsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_files.metadata = {'url': "/deviceUpdate/{instanceId}/updates/providers/{provider}/names/{name}/versions/{version}/files"} # type: ignore @distributed_trace def get_file( self, provider: str, name: str, version: str, file_id: str, access_condition: Optional[_models.AccessCondition] = None, **kwargs: Any ) -> _models.UpdateFile: """Get a specific update file from the version. :param provider: Update provider. Required. :type provider: str :param name: Update name. Required. :type name: str :param version: Update version. Required. :type version: str :param file_id: File identifier. Required. :type file_id: str :param access_condition: Parameter group. Default value is None. :type access_condition: ~deviceupdateclient.models.AccessCondition :keyword callable cls: A custom type or function that will be passed the direct response :return: UpdateFile or the result of cls(response) :rtype: ~deviceupdateclient.models.UpdateFile :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.UpdateFile] _if_none_match = None if access_condition is not None: _if_none_match = access_condition.if_none_match request = build_get_file_request( provider=provider, name=name, version=version, file_id=file_id, instance_id=self._config.instance_id, if_none_match=_if_none_match, api_version=api_version, template_url=self.get_file.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('UpdateFile', pipeline_response) 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 @distributed_trace def list_operation_statuses( self, filter: Optional[str] = None, top: Optional[int] = None, **kwargs: Any ) -> Iterable["_models.UpdateOperation"]: """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. :param filter: Optional to filter operations by status property. Only one specific filter is supported: "status eq 'NotStarted' or status eq 'Running'". Default value is None. :type filter: str :param top: Specifies a non-negative integer n that limits the number of items returned from a collection. The service returns the number of available items up to but not greater than the specified value n. Default value is None. :type top: int :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either UpdateOperation or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~deviceupdateclient.models.UpdateOperation] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.UpdateOperationsList] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) def prepare_request(next_link=None): if not next_link: request = build_list_operation_statuses_request( instance_id=self._config.instance_id, filter=filter, top=top, api_version=api_version, template_url=self.list_operation_statuses.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore else: request = build_list_operation_statuses_request( instance_id=self._config.instance_id, filter=filter, top=top, api_version=api_version, template_url=next_link, headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.method = "GET" return request def extract_data(pipeline_response): deserialized = self._deserialize("UpdateOperationsList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data ) list_operation_statuses.metadata = {'url': "/deviceUpdate/{instanceId}/updates/operations"} # type: ignore @distributed_trace def get_operation_status( self, operation_id: str, access_condition: Optional[_models.AccessCondition] = None, **kwargs: Any ) -> _models.UpdateOperation: """Retrieve operation status. :param operation_id: Operation identifier. Required. :type operation_id: str :param access_condition: Parameter group. Default value is None. :type access_condition: ~deviceupdateclient.models.AccessCondition :keyword callable cls: A custom type or function that will be passed the direct response :return: UpdateOperation or the result of cls(response) :rtype: ~deviceupdateclient.models.UpdateOperation :raises ~azure.core.exceptions.HttpResponseError: """ error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version = kwargs.pop('api_version', _params.pop('api-version', self._config.api_version)) # type: str cls = kwargs.pop('cls', None) # type: ClsType[_models.UpdateOperation] _if_none_match = None if access_condition is not None: _if_none_match = access_condition.if_none_match request = build_get_operation_status_request( operation_id=operation_id, instance_id=self._config.instance_id, if_none_match=_if_none_match, api_version=api_version, template_url=self.get_operation_status.metadata['url'], headers=_headers, params=_params, ) request = _convert_request(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), } request.url = self._client.format_url(request.url, **path_format_arguments) # type: ignore pipeline_response = self._client._pipeline.run( # type: ignore # pylint: disable=protected-access request, stream=False, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response) raise HttpResponseError(response=response, model=error) response_headers = {} response_headers['Retry-After']=self._deserialize('str', response.headers.get('Retry-After')) deserialized = self._deserialize('UpdateOperation', pipeline_response) if cls: return cls(pipeline_response, deserialized, response_headers) return deserialized get_operation_status.metadata = {'url': "/deviceUpdate/{instanceId}/updates/operations/{operationId}"} # type: ignore