# 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 uuid from msrest.pipeline import ClientRawResponse from msrestazure.azure_exceptions import CloudError from .. import models class ConfigurationOperations(object): """ConfigurationOperations operations. :param client: Client for service requests. :param config: Configuration of service client. :param serializer: An object model serializer. :param deserializer: An object model deserializer. :ivar api_version: Version of the Api. Constant value: "2024-03-31". """ models = models def __init__(self, client, config, serializer, deserializer): self._client = client self._serialize = serializer self._deserialize = deserializer self.api_version = "2024-03-31" self.config = config def get( self, id, custom_headers=None, raw=False, **operation_config): """Gets a configuration on the IoT Hub for automatic device/module management. :param id: The unique identifier of the configuration. :type id: str :param dict custom_headers: headers that will be added to the request :param bool raw: returns the direct response alongside the deserialized response :param operation_config: :ref:`Operation configuration overrides`. :return: Configuration or ClientRawResponse if raw=true :rtype: ~service.models.Configuration or ~msrest.pipeline.ClientRawResponse :raises: :class:`CloudError` """ # Construct URL url = self.get.metadata['url'] path_format_arguments = { 'id': self._serialize.url("id", id, 'str') } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} query_parameters['api-version'] = self._serialize.query("self.api_version", self.api_version, 'str') # Construct headers header_parameters = {} header_parameters['Content-Type'] = 'application/json; charset=utf-8' if self.config.generate_client_request_id: header_parameters['x-ms-client-request-id'] = str(uuid.uuid1()) if custom_headers: header_parameters.update(custom_headers) if self.config.accept_language is not None: header_parameters['accept-language'] = self._serialize.header("self.config.accept_language", self.config.accept_language, 'str') # Construct and send request request = self._client.get(url, query_parameters) response = self._client.send(request, header_parameters, stream=False, **operation_config) if response.status_code not in [200]: exp = CloudError(response) exp.request_id = response.headers.get('x-ms-request-id') raise exp deserialized = None if response.status_code == 200: deserialized = self._deserialize('Configuration', response) if raw: client_raw_response = ClientRawResponse(deserialized, response) return client_raw_response return deserialized get.metadata = {'url': '/configurations/{id}'} def create_or_update( self, id, configuration, if_match=None, custom_headers=None, raw=False, **operation_config): """Creates or updates a configuration on the IoT Hub for automatic device/module management. Configuration identifier and Content cannot be updated. :param id: The unique identifier of the configuration. :type id: str :param configuration: The configuration to be created or updated. :type configuration: ~service.models.Configuration :param if_match: The string representing a weak ETag for the configuration, as per RFC7232. This should not be set when creating a configuration, but may be set when updating a configuration. :type if_match: str :param dict custom_headers: headers that will be added to the request :param bool raw: returns the direct response alongside the deserialized response :param operation_config: :ref:`Operation configuration overrides`. :return: Configuration or ClientRawResponse if raw=true :rtype: ~service.models.Configuration or ~msrest.pipeline.ClientRawResponse :raises: :class:`CloudError` """ # Construct URL url = self.create_or_update.metadata['url'] path_format_arguments = { 'id': self._serialize.url("id", id, 'str') } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} query_parameters['api-version'] = self._serialize.query("self.api_version", self.api_version, 'str') # Construct headers header_parameters = {} header_parameters['Content-Type'] = 'application/json; charset=utf-8' if self.config.generate_client_request_id: header_parameters['x-ms-client-request-id'] = str(uuid.uuid1()) if custom_headers: header_parameters.update(custom_headers) if if_match is not None: header_parameters['If-Match'] = self._serialize.header("if_match", if_match, 'str') if self.config.accept_language is not None: header_parameters['accept-language'] = self._serialize.header("self.config.accept_language", self.config.accept_language, 'str') # Construct body body_content = self._serialize.body(configuration, 'Configuration') # Construct and send request request = self._client.put(url, query_parameters) response = self._client.send( request, header_parameters, body_content, stream=False, **operation_config) if response.status_code not in [200, 201]: exp = CloudError(response) exp.request_id = response.headers.get('x-ms-request-id') raise exp deserialized = None if response.status_code == 200: deserialized = self._deserialize('Configuration', response) if response.status_code == 201: deserialized = self._deserialize('Configuration', response) if raw: client_raw_response = ClientRawResponse(deserialized, response) return client_raw_response return deserialized create_or_update.metadata = {'url': '/configurations/{id}'} def delete( self, id, if_match=None, custom_headers=None, raw=False, **operation_config): """Deletes a configuration on the IoT Hub for automatic device/module management. :param id: The unique identifier of the configuration. :type id: str :param if_match: The string representing a weak ETag for the configuration, as per RFC7232. The delete operation is performed only if this ETag matches the value maintained by the server, indicating that the configuration has not been modified since it was last retrieved. To force an unconditional delete, set If-Match to the wildcard character (*). :type if_match: str :param dict custom_headers: headers that will be added to the request :param bool raw: returns the direct response alongside the deserialized response :param operation_config: :ref:`Operation configuration overrides`. :return: None or ClientRawResponse if raw=true :rtype: None or ~msrest.pipeline.ClientRawResponse :raises: :class:`CloudError` """ # Construct URL url = self.delete.metadata['url'] path_format_arguments = { 'id': self._serialize.url("id", id, 'str') } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} query_parameters['api-version'] = self._serialize.query("self.api_version", self.api_version, 'str') # Construct headers header_parameters = {} header_parameters['Content-Type'] = 'application/json; charset=utf-8' if self.config.generate_client_request_id: header_parameters['x-ms-client-request-id'] = str(uuid.uuid1()) if custom_headers: header_parameters.update(custom_headers) if if_match is not None: header_parameters['If-Match'] = self._serialize.header("if_match", if_match, 'str') if self.config.accept_language is not None: header_parameters['accept-language'] = self._serialize.header("self.config.accept_language", self.config.accept_language, 'str') # Construct and send request request = self._client.delete(url, query_parameters) response = self._client.send(request, header_parameters, stream=False, **operation_config) if response.status_code not in [204]: exp = CloudError(response) exp.request_id = response.headers.get('x-ms-request-id') raise exp if raw: client_raw_response = ClientRawResponse(None, response) return client_raw_response delete.metadata = {'url': '/configurations/{id}'} def get_configurations( self, top=None, custom_headers=None, raw=False, **operation_config): """Gets configurations on the IoT Hub for automatic device/module management. Pagination is not supported. :param top: The number of configurations to retrieve. Value will be overridden if greater than the maximum deployment count for the IoT Hub. :type top: int :param dict custom_headers: headers that will be added to the request :param bool raw: returns the direct response alongside the deserialized response :param operation_config: :ref:`Operation configuration overrides`. :return: list or ClientRawResponse if raw=true :rtype: list[~service.models.Configuration] or ~msrest.pipeline.ClientRawResponse :raises: :class:`CloudError` """ # Construct URL url = self.get_configurations.metadata['url'] # Construct parameters query_parameters = {} query_parameters['api-version'] = self._serialize.query("self.api_version", self.api_version, 'str') if top is not None: query_parameters['top'] = self._serialize.query("top", top, 'int') # Construct headers header_parameters = {} header_parameters['Content-Type'] = 'application/json; charset=utf-8' if self.config.generate_client_request_id: header_parameters['x-ms-client-request-id'] = str(uuid.uuid1()) if custom_headers: header_parameters.update(custom_headers) if self.config.accept_language is not None: header_parameters['accept-language'] = self._serialize.header("self.config.accept_language", self.config.accept_language, 'str') # Construct and send request request = self._client.get(url, query_parameters) response = self._client.send(request, header_parameters, stream=False, **operation_config) if response.status_code not in [200]: exp = CloudError(response) exp.request_id = response.headers.get('x-ms-request-id') raise exp deserialized = None if response.status_code == 200: deserialized = self._deserialize('[Configuration]', response) if raw: client_raw_response = ClientRawResponse(deserialized, response) return client_raw_response return deserialized get_configurations.metadata = {'url': '/configurations'} def test_queries( self, target_condition=None, custom_metric_queries=None, custom_headers=None, raw=False, **operation_config): """Validates target condition and custom metric queries for a configuration on the IoT Hub. :param target_condition: The query used to define targeted devices or modules. The query is based on twin tags and/or reported properties. :type target_condition: str :param custom_metric_queries: The key-value pairs with queries and their identifier. :type custom_metric_queries: dict[str, str] :param dict custom_headers: headers that will be added to the request :param bool raw: returns the direct response alongside the deserialized response :param operation_config: :ref:`Operation configuration overrides`. :return: ConfigurationQueriesTestResponse or ClientRawResponse if raw=true :rtype: ~service.models.ConfigurationQueriesTestResponse or ~msrest.pipeline.ClientRawResponse :raises: :class:`CloudError` """ input = models.ConfigurationQueriesTestInput(target_condition=target_condition, custom_metric_queries=custom_metric_queries) # Construct URL url = self.test_queries.metadata['url'] # Construct parameters query_parameters = {} query_parameters['api-version'] = self._serialize.query("self.api_version", self.api_version, 'str') # Construct headers header_parameters = {} header_parameters['Content-Type'] = 'application/json; charset=utf-8' if self.config.generate_client_request_id: header_parameters['x-ms-client-request-id'] = str(uuid.uuid1()) if custom_headers: header_parameters.update(custom_headers) if self.config.accept_language is not None: header_parameters['accept-language'] = self._serialize.header("self.config.accept_language", self.config.accept_language, 'str') # Construct body body_content = self._serialize.body(input, 'ConfigurationQueriesTestInput') # Construct and send request request = self._client.post(url, query_parameters) response = self._client.send( request, header_parameters, body_content, stream=False, **operation_config) if response.status_code not in [200]: exp = CloudError(response) exp.request_id = response.headers.get('x-ms-request-id') raise exp deserialized = None if response.status_code == 200: deserialized = self._deserialize('ConfigurationQueriesTestResponse', response) if raw: client_raw_response = ClientRawResponse(deserialized, response) return client_raw_response return deserialized test_queries.metadata = {'url': '/configurations/testQueries'} def apply_on_edge_device( self, id, content, custom_headers=None, raw=False, **operation_config): """Applies the configuration content to an edge device. :param id: The unique identifier of the edge device. :type id: str :param content: The configuration content. :type content: ~service.models.ConfigurationContent :param dict custom_headers: headers that will be added to the request :param bool raw: returns the direct response alongside the deserialized response :param operation_config: :ref:`Operation configuration overrides`. :return: None or ClientRawResponse if raw=true :rtype: None or ~msrest.pipeline.ClientRawResponse :raises: :class:`CloudError` """ # Construct URL url = self.apply_on_edge_device.metadata['url'] path_format_arguments = { 'id': self._serialize.url("id", id, 'str') } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} query_parameters['api-version'] = self._serialize.query("self.api_version", self.api_version, 'str') # Construct headers header_parameters = {} header_parameters['Content-Type'] = 'application/json; charset=utf-8' if self.config.generate_client_request_id: header_parameters['x-ms-client-request-id'] = str(uuid.uuid1()) if custom_headers: header_parameters.update(custom_headers) if self.config.accept_language is not None: header_parameters['accept-language'] = self._serialize.header("self.config.accept_language", self.config.accept_language, 'str') # Construct body body_content = self._serialize.body(content, 'ConfigurationContent') # Construct and send request request = self._client.post(url, query_parameters) response = self._client.send( request, header_parameters, body_content, stream=False, **operation_config) # @vilit handcrafted change if response.status_code not in [200, 204]: exp = CloudError(response) exp.request_id = response.headers.get('x-ms-request-id') raise exp deserialized = None if response.status_code == 200: deserialized = self._deserialize('object', response) if raw: client_raw_response = ClientRawResponse(deserialized, response) return client_raw_response return deserialized apply_on_edge_device.metadata = {'url': '/devices/{id}/applyConfigurationContent'}