azext_iot/sdk/iothub/device/operations/device_operations.py

# 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 DeviceOperations(object):
    """DeviceOperations 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: "2019-10-01".
    """

    models = models

    def __init__(self, client, config, serializer, deserializer):

        self._client = client
        self._serialize = serializer
        self._deserialize = deserializer
        self.api_version = "2019-10-01"

        self.config = config

    def get_devices_and_modules_in_scope(
            self, device_id, module_id, custom_headers=None, raw=False, **operation_config):
        """Query IotHub to retrieve information regarding devices which belong to
        the same deviceScope.

        Query IotHub to retrieve information regarding devices which belong to
        the same deviceScope. See
        https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-query-language
        for more information. Pagination of results is supported. This returns
        information about device twins only.

        :param device_id:
        :type device_id: str
        :param module_id:
        :type module_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<msrest:optionsforoperations>`.
        :return: ScopeResult or ClientRawResponse if raw=true
        :rtype: ~device.models.ScopeResult or
         ~msrest.pipeline.ClientRawResponse
        :raises: :class:`CloudError<msrestazure.azure_exceptions.CloudError>`
        """
        # Construct URL
        url = self.get_devices_and_modules_in_scope.metadata['url']
        path_format_arguments = {
            'deviceId': self._serialize.url("device_id", device_id, 'str'),
            'moduleId': self._serialize.url("module_id", module_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['Accept'] = 'application/json'
        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, header_parameters)
        response = self._client.send(request, 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('ScopeResult', response)

        if raw:
            client_raw_response = ClientRawResponse(deserialized, response)
            return client_raw_response

        return deserialized
    get_devices_and_modules_in_scope.metadata = {'url': '/devices/{deviceId}/modules/{moduleId}/devicesAndModulesInDeviceScope'}

    def get_device_and_module_in_scope(
            self, device_id, module_id, custom_headers=None, raw=False, **operation_config):
        """Get device registry information using module connect for a specific
        device in deviceScope.

        Get device registry information using module connect for a specific
        device in deviceScope. See
        https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-query-language
        for more information. Pagination of results is supported. This returns
        information about device twins only.

        :param device_id:
        :type device_id: str
        :param module_id:
        :type module_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<msrest:optionsforoperations>`.
        :return: ScopeResult or ClientRawResponse if raw=true
        :rtype: ~device.models.ScopeResult or
         ~msrest.pipeline.ClientRawResponse
        :raises: :class:`CloudError<msrestazure.azure_exceptions.CloudError>`
        """
        # Construct URL
        url = self.get_device_and_module_in_scope.metadata['url']
        path_format_arguments = {
            'deviceId': self._serialize.url("device_id", device_id, 'str'),
            'moduleId': self._serialize.url("module_id", module_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['Accept'] = 'application/json'
        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, header_parameters)
        response = self._client.send(request, 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('ScopeResult', response)

        if raw:
            client_raw_response = ClientRawResponse(deserialized, response)
            return client_raw_response

        return deserialized
    get_device_and_module_in_scope.metadata = {'url': '/devices/{deviceId}/modules/{moduleId}/deviceAndModuleInDeviceScope'}

    # @digimaun - Altered to support request body parameter of object
    def send_device_event(
            self, id, message, iothub_app_xxx=None, custom_headers=None, raw=False, **operation_config):
        """Send a device-to-cloud message.

        Send a device-to-cloud message. See
        https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-messaging for
        more information.

        :param id: Device ID.
        :type id: str
        :param iothub_app_xxx: Optional application property prefixed with
         'iothub-app-' to be placed in 'applicationProperties' section of the
         message ('iothub-app-' prefix will be stripped).
        :type iothub_app_xxx: 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<msrest:optionsforoperations>`.
        :return: None or ClientRawResponse if raw=true
        :rtype: None or ~msrest.pipeline.ClientRawResponse
        :raises: :class:`CloudError<msrestazure.azure_exceptions.CloudError>`
        """
        # Construct URL
        url = self.send_device_event.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 = {}

        # @digimaun - No assumption on http based d2c content.
        if self.config.generate_client_request_id:
            pass
        if custom_headers:
            header_parameters.update(custom_headers)
        if self.config.accept_language is not None:
            pass

        # Construct and send request
        # @digimaun - Construct Body
        body_content = self._serialize.body(message, 'object')

        # @digimaun - Construct and send request
        request = self._client.post(url, query_parameters, header_parameters)
        response = self._client.send(request, header_parameters, body_content, 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
    send_device_event.metadata = {'url': '/devices/{id}/messages/events'}

    def receive_device_bound_notification(
            self, id, custom_headers=None, raw=False, **operation_config):
        """This method is used to retrieve a cloud-to-device message.

        This method is used to retrieve a cloud-to-device message See
        https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-messaging for
        more information. This capability is only available in the standard
        tier IoT Hub. For more information, see [Choose the right IoT Hub
        tier](https://aka.ms/scaleyouriotsolution).

        :param id: Device ID.
        :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<msrest:optionsforoperations>`.
        :return: None or ClientRawResponse if raw=true
        :rtype: None or ~msrest.pipeline.ClientRawResponse
        :raises: :class:`CloudError<msrestazure.azure_exceptions.CloudError>`
        """
        # Construct URL
        url = self.receive_device_bound_notification.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 = {}
        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, header_parameters)
        response = self._client.send(request, stream=False, **operation_config)

        if response.status_code not in [200, 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
    receive_device_bound_notification.metadata = {'url': '/devices/{id}/messages/deviceBound'}

    def abandon_device_bound_notification(
            self, id, etag, custom_headers=None, raw=False, **operation_config):
        """This method abandons a cloud-to-device message.

        This method abandons a cloud-to-device message. The Etag obtained when
        the message was received must be provided to resolve race conditions
        when completing, rejecting, or abandoning a message. A abandoned
        message is put back in the device message queue for re-delivery to the
        device. See
        https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-messaging for
        more information. Currently, the use of the Etag in the header does not
        comply with RFC 7232. A fix for this issue is currently on our backlog.
        This capability is only available in the standard tier IoT Hub. For
        more information, see [Choose the right IoT Hub
        tier](https://aka.ms/scaleyouriotsolution).

        :param id: Device ID.
        :type id: str
        :param etag:
        :type etag: 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<msrest:optionsforoperations>`.
        :return: None or ClientRawResponse if raw=true
        :rtype: None or ~msrest.pipeline.ClientRawResponse
        :raises: :class:`CloudError<msrestazure.azure_exceptions.CloudError>`
        """
        # Construct URL
        url = self.abandon_device_bound_notification.metadata['url']
        path_format_arguments = {
            'id': self._serialize.url("id", id, 'str'),
            'etag': self._serialize.url("etag", etag, '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 = {}
        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.post(url, query_parameters, header_parameters)
        response = self._client.send(request, 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
    abandon_device_bound_notification.metadata = {'url': '/devices/{id}/messages/deviceBound/{etag}/abandon'}

    def create_file_upload_sas_uri(
            self, device_id, blob_name=None, custom_headers=None, raw=False, **operation_config):
        """This method is used to retrieve a storage SAS URI to upload a file.

        This method is used to retrieve a storage SAS URI to upload a file. See
        https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-file-upload
        for more information.

        :param device_id: Device ID.
        :type device_id: str
        :param blob_name: Name of the blob.
        :type blob_name: 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<msrest:optionsforoperations>`.
        :return: FileUploadResponse or ClientRawResponse if raw=true
        :rtype: ~device.models.FileUploadResponse or
         ~msrest.pipeline.ClientRawResponse
        :raises: :class:`CloudError<msrestazure.azure_exceptions.CloudError>`
        """
        file_upload_request = models.FileUploadRequest(blob_name=blob_name)

        # Construct URL
        url = self.create_file_upload_sas_uri.metadata['url']
        path_format_arguments = {
            'deviceId': self._serialize.url("device_id", device_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['Accept'] = 'application/json'
        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(file_upload_request, 'FileUploadRequest')

        # Construct and send request
        request = self._client.post(url, query_parameters, header_parameters, body_content)
        response = self._client.send(request, 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('FileUploadResponse', response)

        if raw:
            client_raw_response = ClientRawResponse(deserialized, response)
            return client_raw_response

        return deserialized
    create_file_upload_sas_uri.metadata = {'url': '/devices/{deviceId}/files'}

    def update_file_upload_status(
            self, device_id, file_upload_completion_status, custom_headers=None, raw=False, **operation_config):
        """This method is used to notify an IoT hub of a completed file upload.

        This method is used to notify an IoT hub of a completed file upload.
        See
        https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-file-upload
        for more information.

        :param device_id: Device ID.
        :type device_id: str
        :param file_upload_completion_status: File upload completion status
         object.
        :type file_upload_completion_status:
         ~device.models.FileUploadCompletionStatus
        :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<msrest:optionsforoperations>`.
        :return: FileUploadCompletionStatus or ClientRawResponse if raw=true
        :rtype: ~device.models.FileUploadCompletionStatus or
         ~msrest.pipeline.ClientRawResponse
        :raises: :class:`CloudError<msrestazure.azure_exceptions.CloudError>`
        """
        # Construct URL
        url = self.update_file_upload_status.metadata['url']
        path_format_arguments = {
            'deviceId': self._serialize.url("device_id", device_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['Accept'] = 'application/json'
        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(file_upload_completion_status, 'FileUploadCompletionStatus')

        # Construct and send request
        request = self._client.post(url, query_parameters, header_parameters, body_content)
        response = self._client.send(request, 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

        deserialized = None

        if response.status_code == 204:
            deserialized = self._deserialize('FileUploadCompletionStatus', response)

        if raw:
            client_raw_response = ClientRawResponse(deserialized, response)
            return client_raw_response

        return deserialized
    update_file_upload_status.metadata = {'url': '/devices/{deviceId}/files/notifications'}

    # @digimaun - added entire upload_file_to_container
    def upload_file_to_container(
            self, storage_endpoint, content, content_type):
        import requests

        """
        Uploads a file to the specified Azure storage endpoint.

        Args:
            storage_endpoint (str): target url of container to post file to.
            content (object): the content to post to the storage endpoint
            content_type (dict): the IANA Media Type of the content.

        Raises:
            CloudError: when http response is not 200 or 201.

        """
        # Construct headers
        header_parameters = {}
        header_parameters['Content-Type'] = content_type
        header_parameters['Content-Length'] = str(len(content))
        header_parameters['x-ms-blob-type'] = 'BlockBlob'

        blob_payload = content
        protocol = 'https://'
        response = requests.put('{}{}'.format(protocol, storage_endpoint),
                                headers=header_parameters,
                                data=blob_payload)

        if response.status_code in [200, 201]:
            return response

        exp = CloudError(response)
        exp.request_id = response.headers.get('x-ms-request-id')
        raise exp

    # @digimaun - Altered to support reject via query param
    def complete_device_bound_notification(
            self, id, etag, reject=None, custom_headers=None, raw=False, **operation_config):
        """This method completes or rejects a cloud-to-device message.

        This method completes or rejects a cloud-to-device message. The Etag
        obtained when the message was received must be provided to resolve race
        conditions when completing, rejecting, or abandoning a message. A
        completed message is deleted from the device message queue, and a
        positive acknowledgment is sent to the application back-end if
        requested. A rejected message causes it to be deadlettered. To reject a
        message, include a query parameter called \"reject\". See
        https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-messaging for
        more information. Currently, the use of the Etag in the header does not
        comply with RFC 7232. A fix for this issue is currently on our backlog.
        This capability is only available in the standard tier IoT Hub. For
        more information, see [Choose the right IoT Hub
        tier](https://aka.ms/scaleyouriotsolution).

        :param id: Device ID.
        :type id: str
        :param etag:
        :type etag: 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<msrest:optionsforoperations>`.
        :return: None or ClientRawResponse if raw=true
        :rtype: None or ~msrest.pipeline.ClientRawResponse
        :raises: :class:`CloudError<msrestazure.azure_exceptions.CloudError>`
        """
        # Construct URL
        url = self.complete_device_bound_notification.metadata['url']
        path_format_arguments = {
            'id': self._serialize.url("id", id, 'str'),
            'etag': self._serialize.url("etag", etag, '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')

        # @digimaun
        if reject is not None:
            query_parameters['reject'] = self._serialize.query("reject", reject, 'str')

        # Construct headers
        header_parameters = {}
        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.delete(url, query_parameters, header_parameters)
        response = self._client.send(request, 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
    complete_device_bound_notification.metadata = {'url': '/devices/{id}/messages/deviceBound/{etag}'}