azext_iot/sdk/digitaltwins/dataplane/operations/digital_twins_operations.py (659 lines of code) (raw):
# 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 .. import models
class DigitalTwinsOperations(object):
"""DigitalTwinsOperations 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: The requested API version. Constant value: "2023-10-31".
"""
models = models
def __init__(self, client, config, serializer, deserializer):
self._client = client
self._serialize = serializer
self._deserialize = deserializer
self.api_version = "2023-10-31"
self.config = config
def get_by_id(
self, id, digital_twins_get_by_id_options=None, custom_headers=None, raw=False, **operation_config):
"""Retrieves a digital twin.
Status codes:
* 200 OK
* 400 Bad Request
* InvalidArgument - The digital twin id is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param digital_twins_get_by_id_options: Additional parameters for the
operation
:type digital_twins_get_by_id_options:
~dataplane.models.DigitalTwinsGetByIdOptions
: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: dict or ClientRawResponse if raw=true
:rtype: dict[str, object] or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
traceparent = None
if digital_twins_get_by_id_options is not None:
traceparent = digital_twins_get_by_id_options.traceparent
tracestate = None
if digital_twins_get_by_id_options is not None:
tracestate = digital_twins_get_by_id_options.tracestate
# Construct URL
url = self.get_by_id.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', min_length=1)
# 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')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, '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]:
raise models.ErrorResponseException(self._deserialize, response)
deserialized = None
header_dict = {}
if response.status_code == 200:
deserialized = self._deserialize('{object}', response)
header_dict = {
'ETag': 'str',
'x-ms-error-code': 'str',
}
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
client_raw_response.add_headers(header_dict)
return client_raw_response
return deserialized
get_by_id.metadata = {'url': '/digitaltwins/{id}'}
def add(
self, id, twin, digital_twins_add_options=None, custom_headers=None, raw=False, **operation_config):
"""Adds or replaces a digital twin.
Status codes:
* 200 OK
* 400 Bad Request
* InvalidArgument - The digital twin id or payload is invalid.
* ModelDecommissioned - The model for the digital twin is
decommissioned.
* TwinLimitReached - The maximum number of digital twins allowed has
been reached.
* ValidationFailed - The digital twin payload is not valid.
* 412 Precondition Failed
* PreconditionFailed - The precondition check (If-Match or
If-None-Match) failed.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param twin: The digital twin instance being added. If provided, the
$dtId property is ignored.
:type twin: object
:param digital_twins_add_options: Additional parameters for the
operation
:type digital_twins_add_options:
~dataplane.models.DigitalTwinsAddOptions
: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: dict or ClientRawResponse if raw=true
:rtype: dict[str, object] or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
if_none_match = None
if digital_twins_add_options is not None:
if_none_match = digital_twins_add_options.if_none_match
traceparent = None
if digital_twins_add_options is not None:
traceparent = digital_twins_add_options.traceparent
tracestate = None
if digital_twins_add_options is not None:
tracestate = digital_twins_add_options.tracestate
# Construct URL
url = self.add.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', min_length=1)
# 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')
if if_none_match is not None:
header_parameters['If-None-Match'] = self._serialize.header("if_none_match", if_none_match, 'str')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct body
body_content = self._serialize.body(twin, 'object')
# 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]:
raise models.ErrorResponseException(self._deserialize, response)
deserialized = None
header_dict = {}
if response.status_code == 200:
deserialized = self._deserialize('{object}', response)
header_dict = {
'ETag': 'str',
'x-ms-error-code': 'str',
}
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
client_raw_response.add_headers(header_dict)
return client_raw_response
return deserialized
add.metadata = {'url': '/digitaltwins/{id}'}
def delete(
self, id, digital_twins_delete_options=None, custom_headers=None, raw=False, **operation_config):
"""Deletes a digital twin. All relationships referencing the digital twin
must already be deleted.
Status codes:
* 204 No Content
* 400 Bad Request
* InvalidArgument - The digital twin id is invalid.
* RelationshipsNotDeleted - The digital twin contains relationships.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* 412 Precondition Failed
* PreconditionFailed - The precondition check (If-Match or
If-None-Match) failed.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param digital_twins_delete_options: Additional parameters for the
operation
:type digital_twins_delete_options:
~dataplane.models.DigitalTwinsDeleteOptions
: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:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
if_match = None
if digital_twins_delete_options is not None:
if_match = digital_twins_delete_options.if_match
traceparent = None
if digital_twins_delete_options is not None:
traceparent = digital_twins_delete_options.traceparent
tracestate = None
if digital_twins_delete_options is not None:
tracestate = digital_twins_delete_options.tracestate
# 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', min_length=1)
# 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')
if if_match is not None:
header_parameters['If-Match'] = self._serialize.header("if_match", if_match, 'str')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct and send request
request = self._client.delete(url, query_parameters)
response = self._client.send(request, header_parameters, stream=False, **operation_config)
# @vilit - added 202
if response.status_code not in [202, 204]:
raise models.ErrorResponseException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'x-ms-error-code': 'str',
})
return client_raw_response
delete.metadata = {'url': '/digitaltwins/{id}'}
def update(
self, id, patch_document, digital_twins_update_options=None, custom_headers=None, raw=False, **operation_config):
"""Updates a digital twin.
Status codes:
* 204 No Content
* 400 Bad Request
* InvalidArgument - The digital twin id or payload is invalid.
* JsonPatchInvalid - The JSON Patch provided is invalid.
* ValidationFailed - Applying the patch results in an invalid digital
twin.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* 412 Precondition Failed
* PreconditionFailed - The precondition check (If-Match or
If-None-Match) failed.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param patch_document: An update specification described by JSON
Patch. Updates to property values and $model elements may happen in
the same request. Operations are limited to add, replace and remove.
:type patch_document: list[object]
:param digital_twins_update_options: Additional parameters for the
operation
:type digital_twins_update_options:
~dataplane.models.DigitalTwinsUpdateOptions
: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:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
if_match = None
if digital_twins_update_options is not None:
if_match = digital_twins_update_options.if_match
traceparent = None
if digital_twins_update_options is not None:
traceparent = digital_twins_update_options.traceparent
tracestate = None
if digital_twins_update_options is not None:
tracestate = digital_twins_update_options.tracestate
# Construct URL
url = self.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', min_length=1)
# Construct headers
header_parameters = {}
header_parameters['Content-Type'] = 'application/json-patch+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')
if if_match is not None:
header_parameters['If-Match'] = self._serialize.header("if_match", if_match, 'str')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct body
body_content = self._serialize.body(patch_document, '[object]')
# Construct and send request
request = self._client.patch(url, query_parameters)
response = self._client.send(
request, header_parameters, body_content, stream=False, **operation_config)
# @vilit - added 202
if response.status_code not in [202, 204]:
raise models.ErrorResponseException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'ETag': 'str',
'x-ms-error-code': 'str',
})
return client_raw_response
update.metadata = {'url': '/digitaltwins/{id}'}
def get_relationship_by_id(
self, id, relationship_id, digital_twins_get_relationship_by_id_options=None, custom_headers=None, raw=False, **operation_config):
"""Retrieves a relationship between two digital twins.
Status codes:
* 200 OK
* 400 Bad Request
* InvalidArgument - The digital twin id or relationship id is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* RelationshipNotFound - The relationship was not found.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param relationship_id: The id of the relationship. The id is unique
within the digital twin and case sensitive.
:type relationship_id: str
:param digital_twins_get_relationship_by_id_options: Additional
parameters for the operation
:type digital_twins_get_relationship_by_id_options:
~dataplane.models.DigitalTwinsGetRelationshipByIdOptions
: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: dict or ClientRawResponse if raw=true
:rtype: dict[str, object] or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
traceparent = None
if digital_twins_get_relationship_by_id_options is not None:
traceparent = digital_twins_get_relationship_by_id_options.traceparent
tracestate = None
if digital_twins_get_relationship_by_id_options is not None:
tracestate = digital_twins_get_relationship_by_id_options.tracestate
# Construct URL
url = self.get_relationship_by_id.metadata['url']
path_format_arguments = {
'id': self._serialize.url("id", id, 'str'),
'relationshipId': self._serialize.url("relationship_id", relationship_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', min_length=1)
# 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')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, '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]:
raise models.ErrorResponseException(self._deserialize, response)
deserialized = None
header_dict = {}
if response.status_code == 200:
deserialized = self._deserialize('{object}', response)
header_dict = {
'ETag': 'str',
'x-ms-error-code': 'str',
}
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
client_raw_response.add_headers(header_dict)
return client_raw_response
return deserialized
get_relationship_by_id.metadata = {'url': '/digitaltwins/{id}/relationships/{relationshipId}'}
def add_relationship(
self, id, relationship_id, relationship, digital_twins_add_relationship_options=None, custom_headers=None, raw=False, **operation_config):
"""Adds a relationship between two digital twins.
Status codes:
* 200 OK
* 400 Bad Request
* InvalidArgument - The digital twin id, relationship id, or payload is
invalid.
* InvalidRelationship - The relationship is invalid.
* OperationNotAllowed - The relationship cannot connect to the same
digital twin.
* ValidationFailed - The relationship content is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* TargetTwinNotFound - The digital twin target of the relationship was
not found.
* 412 Precondition Failed
* PreconditionFailed - The precondition check (If-Match or
If-None-Match) failed.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param relationship_id: The id of the relationship. The id is unique
within the digital twin and case sensitive.
:type relationship_id: str
:param relationship: The data for the relationship.
:type relationship: object
:param digital_twins_add_relationship_options: Additional parameters
for the operation
:type digital_twins_add_relationship_options:
~dataplane.models.DigitalTwinsAddRelationshipOptions
: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: dict or ClientRawResponse if raw=true
:rtype: dict[str, object] or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
if_none_match = None
if digital_twins_add_relationship_options is not None:
if_none_match = digital_twins_add_relationship_options.if_none_match
traceparent = None
if digital_twins_add_relationship_options is not None:
traceparent = digital_twins_add_relationship_options.traceparent
tracestate = None
if digital_twins_add_relationship_options is not None:
tracestate = digital_twins_add_relationship_options.tracestate
# Construct URL
url = self.add_relationship.metadata['url']
path_format_arguments = {
'id': self._serialize.url("id", id, 'str'),
'relationshipId': self._serialize.url("relationship_id", relationship_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', min_length=1)
# 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')
if if_none_match is not None:
header_parameters['If-None-Match'] = self._serialize.header("if_none_match", if_none_match, 'str')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct body
body_content = self._serialize.body(relationship, 'object')
# 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]:
raise models.ErrorResponseException(self._deserialize, response)
deserialized = None
header_dict = {}
if response.status_code == 200:
deserialized = self._deserialize('{object}', response)
header_dict = {
'ETag': 'str',
'x-ms-error-code': 'str',
}
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
client_raw_response.add_headers(header_dict)
return client_raw_response
return deserialized
add_relationship.metadata = {'url': '/digitaltwins/{id}/relationships/{relationshipId}'}
def delete_relationship(
self, id, relationship_id, digital_twins_delete_relationship_options=None, custom_headers=None, raw=False, **operation_config):
"""Deletes a relationship between two digital twins.
Status codes:
* 204 No Content
* 400 Bad Request
* InvalidArgument - The digital twin id or relationship id is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* RelationshipNotFound - The relationship was not found.
* 412 Precondition Failed
* PreconditionFailed - The precondition check (If-Match or
If-None-Match) failed.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param relationship_id: The id of the relationship. The id is unique
within the digital twin and case sensitive.
:type relationship_id: str
:param digital_twins_delete_relationship_options: Additional
parameters for the operation
:type digital_twins_delete_relationship_options:
~dataplane.models.DigitalTwinsDeleteRelationshipOptions
: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:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
if_match = None
if digital_twins_delete_relationship_options is not None:
if_match = digital_twins_delete_relationship_options.if_match
traceparent = None
if digital_twins_delete_relationship_options is not None:
traceparent = digital_twins_delete_relationship_options.traceparent
tracestate = None
if digital_twins_delete_relationship_options is not None:
tracestate = digital_twins_delete_relationship_options.tracestate
# Construct URL
url = self.delete_relationship.metadata['url']
path_format_arguments = {
'id': self._serialize.url("id", id, 'str'),
'relationshipId': self._serialize.url("relationship_id", relationship_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', min_length=1)
# 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')
if if_match is not None:
header_parameters['If-Match'] = self._serialize.header("if_match", if_match, 'str')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct and send request
request = self._client.delete(url, query_parameters)
response = self._client.send(request, header_parameters, stream=False, **operation_config)
# @vilit - added 202
if response.status_code not in [202, 204]:
raise models.ErrorResponseException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'x-ms-error-code': 'str',
})
return client_raw_response
delete_relationship.metadata = {'url': '/digitaltwins/{id}/relationships/{relationshipId}'}
def update_relationship(
self, id, relationship_id, patch_document, digital_twins_update_relationship_options=None, custom_headers=None, raw=False, **operation_config):
"""Updates the properties on a relationship between two digital twins.
Status codes:
* 204 No Content
* 400 Bad Request
* InvalidArgument - The digital twin id or relationship id is invalid.
* InvalidRelationship - The relationship is invalid.
* JsonPatchInvalid - The JSON Patch provided is invalid.
* ValidationFailed - The relationship content is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* RelationshipNotFound - The relationship was not found.
* 409 Conflict
* RelationshipAlreadyExists - The relationship already exists.
* 412 Precondition Failed
* PreconditionFailed - The precondition check (If-Match or
If-None-Match) failed.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param relationship_id: The id of the relationship. The id is unique
within the digital twin and case sensitive.
:type relationship_id: str
:param patch_document: JSON Patch description of the update to the
relationship properties.
:type patch_document: list[object]
:param digital_twins_update_relationship_options: Additional
parameters for the operation
:type digital_twins_update_relationship_options:
~dataplane.models.DigitalTwinsUpdateRelationshipOptions
: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:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
if_match = None
if digital_twins_update_relationship_options is not None:
if_match = digital_twins_update_relationship_options.if_match
traceparent = None
if digital_twins_update_relationship_options is not None:
traceparent = digital_twins_update_relationship_options.traceparent
tracestate = None
if digital_twins_update_relationship_options is not None:
tracestate = digital_twins_update_relationship_options.tracestate
# Construct URL
url = self.update_relationship.metadata['url']
path_format_arguments = {
'id': self._serialize.url("id", id, 'str'),
'relationshipId': self._serialize.url("relationship_id", relationship_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', min_length=1)
# Construct headers
header_parameters = {}
header_parameters['Content-Type'] = 'application/json-patch+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')
if if_match is not None:
header_parameters['If-Match'] = self._serialize.header("if_match", if_match, 'str')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct body
body_content = self._serialize.body(patch_document, '[object]')
# Construct and send request
request = self._client.patch(url, query_parameters)
response = self._client.send(
request, header_parameters, body_content, stream=False, **operation_config)
# @vilit - added 202
if response.status_code not in [202, 204]:
raise models.ErrorResponseException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'ETag': 'str',
'x-ms-error-code': 'str',
})
return client_raw_response
update_relationship.metadata = {'url': '/digitaltwins/{id}/relationships/{relationshipId}'}
def list_relationships(
self, id, relationship_name=None, digital_twins_list_relationships_options=None, custom_headers=None, raw=False, **operation_config):
"""Retrieves the relationships from a digital twin.
Status codes:
* 200 OK
* 400 Bad Request
* InvalidArgument - The digital twin id is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param relationship_name: The name of the relationship.
:type relationship_name: str
:param digital_twins_list_relationships_options: Additional parameters
for the operation
:type digital_twins_list_relationships_options:
~dataplane.models.DigitalTwinsListRelationshipsOptions
: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: An iterator like instance of object
:rtype: ~dataplane.models.ObjectPaged[object]
:raises:
:class:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
traceparent = None
if digital_twins_list_relationships_options is not None:
traceparent = digital_twins_list_relationships_options.traceparent
tracestate = None
if digital_twins_list_relationships_options is not None:
tracestate = digital_twins_list_relationships_options.tracestate
def internal_paging(next_link=None, raw=False):
if not next_link:
# Construct URL
url = self.list_relationships.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 = {}
if relationship_name is not None:
query_parameters['relationshipName'] = self._serialize.query("relationship_name", relationship_name, 'str')
query_parameters['api-version'] = self._serialize.query("self.api_version", self.api_version, 'str', min_length=1)
else:
url = next_link
query_parameters = {}
# 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')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, '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]:
raise models.ErrorResponseException(self._deserialize, response)
return response
# Deserialize response
deserialized = models.ObjectPaged(internal_paging, self._deserialize.dependencies)
if raw:
header_dict = {}
client_raw_response = models.ObjectPaged(internal_paging, self._deserialize.dependencies, header_dict)
return client_raw_response
return deserialized
list_relationships.metadata = {'url': '/digitaltwins/{id}/relationships'}
def list_incoming_relationships(
self, id, digital_twins_list_incoming_relationships_options=None, custom_headers=None, raw=False, **operation_config):
"""Retrieves all incoming relationship for a digital twin.
Status codes:
* 200 OK
* 400 Bad Request
* InvalidArgument - The digital twin id is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param digital_twins_list_incoming_relationships_options: Additional
parameters for the operation
:type digital_twins_list_incoming_relationships_options:
~dataplane.models.DigitalTwinsListIncomingRelationshipsOptions
: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: An iterator like instance of IncomingRelationship
:rtype:
~dataplane.models.IncomingRelationshipPaged[~dataplane.models.IncomingRelationship]
:raises:
:class:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
traceparent = None
if digital_twins_list_incoming_relationships_options is not None:
traceparent = digital_twins_list_incoming_relationships_options.traceparent
tracestate = None
if digital_twins_list_incoming_relationships_options is not None:
tracestate = digital_twins_list_incoming_relationships_options.tracestate
def internal_paging(next_link=None, raw=False):
if not next_link:
# Construct URL
url = self.list_incoming_relationships.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', min_length=1)
else:
url = next_link
query_parameters = {}
# 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')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, '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]:
raise models.ErrorResponseException(self._deserialize, response)
return response
# Deserialize response
deserialized = models.IncomingRelationshipPaged(internal_paging, self._deserialize.dependencies)
if raw:
header_dict = {}
client_raw_response = models.IncomingRelationshipPaged(internal_paging, self._deserialize.dependencies, header_dict)
return client_raw_response
return deserialized
list_incoming_relationships.metadata = {'url': '/digitaltwins/{id}/incomingrelationships'}
def send_telemetry(
self, id, telemetry, message_id, telemetry_source_time=None, digital_twins_send_telemetry_options=None, custom_headers=None, raw=False, **operation_config):
"""Sends telemetry on behalf of a digital twin.
Status codes:
* 204 No Content
* 400 Bad Request
* InvalidArgument - The digital twin id or message id is invalid.
* ValidationFailed - The telemetry content is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param telemetry: The telemetry measurements to send from the digital
twin.
:type telemetry: object
:param message_id: A unique message identifier (in the scope of the
digital twin id) that is commonly used for de-duplicating messages.
:type message_id: str
:param telemetry_source_time: An RFC 3339 timestamp that identifies
the time the telemetry was measured.
:type telemetry_source_time: str
:param digital_twins_send_telemetry_options: Additional parameters for
the operation
:type digital_twins_send_telemetry_options:
~dataplane.models.DigitalTwinsSendTelemetryOptions
: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:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
traceparent = None
if digital_twins_send_telemetry_options is not None:
traceparent = digital_twins_send_telemetry_options.traceparent
tracestate = None
if digital_twins_send_telemetry_options is not None:
tracestate = digital_twins_send_telemetry_options.tracestate
# Construct URL
url = self.send_telemetry.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', min_length=1)
# 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)
header_parameters['Message-Id'] = self._serialize.header("message_id", message_id, 'str')
if telemetry_source_time is not None:
header_parameters['Telemetry-Source-Time'] = self._serialize.header("telemetry_source_time", telemetry_source_time, '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')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct body
body_content = self._serialize.body(telemetry, 'object')
# 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 - added 202
if response.status_code not in [202, 204]:
raise models.ErrorResponseException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'x-ms-error-code': 'str',
})
return client_raw_response
send_telemetry.metadata = {'url': '/digitaltwins/{id}/telemetry'}
def send_component_telemetry(
self, id, component_path, telemetry, message_id, telemetry_source_time=None, digital_twins_send_component_telemetry_options=None, custom_headers=None, raw=False, **operation_config):
"""Sends telemetry on behalf of a component in a digital twin.
Status codes:
* 204 No Content
* 400 Bad Request
* InvalidArgument - The digital twin id, message id, or component path
is invalid.
* ValidationFailed - The telemetry content is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* ComponentNotFound - The component path was not found.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param component_path: The name of the DTDL component.
:type component_path: str
:param telemetry: The telemetry measurements to send from the digital
twin's component.
:type telemetry: object
:param message_id: A unique message identifier (in the scope of the
digital twin id) that is commonly used for de-duplicating messages.
:type message_id: str
:param telemetry_source_time: An RFC 3339 timestamp that identifies
the time the telemetry was measured.
:type telemetry_source_time: str
:param digital_twins_send_component_telemetry_options: Additional
parameters for the operation
:type digital_twins_send_component_telemetry_options:
~dataplane.models.DigitalTwinsSendComponentTelemetryOptions
: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:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
traceparent = None
if digital_twins_send_component_telemetry_options is not None:
traceparent = digital_twins_send_component_telemetry_options.traceparent
tracestate = None
if digital_twins_send_component_telemetry_options is not None:
tracestate = digital_twins_send_component_telemetry_options.tracestate
# Construct URL
url = self.send_component_telemetry.metadata['url']
path_format_arguments = {
'id': self._serialize.url("id", id, 'str'),
'componentPath': self._serialize.url("component_path", component_path, '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', min_length=1)
# 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)
header_parameters['Message-Id'] = self._serialize.header("message_id", message_id, 'str')
if telemetry_source_time is not None:
header_parameters['Telemetry-Source-Time'] = self._serialize.header("telemetry_source_time", telemetry_source_time, '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')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct body
body_content = self._serialize.body(telemetry, 'object')
# 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 - added 202
if response.status_code not in [202, 204]:
raise models.ErrorResponseException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'x-ms-error-code': 'str',
})
return client_raw_response
send_component_telemetry.metadata = {'url': '/digitaltwins/{id}/components/{componentPath}/telemetry'}
def get_component(
self, id, component_path, digital_twins_get_component_options=None, custom_headers=None, raw=False, **operation_config):
"""Retrieves a component from a digital twin.
Status codes:
* 200 OK
* 400 Bad Request
* InvalidArgument - The digital twin id or component path is invalid.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* ComponentNotFound - The component path was not found.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param component_path: The name of the DTDL component.
:type component_path: str
:param digital_twins_get_component_options: Additional parameters for
the operation
:type digital_twins_get_component_options:
~dataplane.models.DigitalTwinsGetComponentOptions
: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: dict or ClientRawResponse if raw=true
:rtype: dict[str, object] or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
traceparent = None
if digital_twins_get_component_options is not None:
traceparent = digital_twins_get_component_options.traceparent
tracestate = None
if digital_twins_get_component_options is not None:
tracestate = digital_twins_get_component_options.tracestate
# Construct URL
url = self.get_component.metadata['url']
path_format_arguments = {
'id': self._serialize.url("id", id, 'str'),
'componentPath': self._serialize.url("component_path", component_path, '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', min_length=1)
# 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')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, '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]:
raise models.ErrorResponseException(self._deserialize, response)
deserialized = None
header_dict = {}
if response.status_code == 200:
deserialized = self._deserialize('{object}', response)
header_dict = {
'ETag': 'str',
'x-ms-error-code': 'str',
}
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
client_raw_response.add_headers(header_dict)
return client_raw_response
return deserialized
get_component.metadata = {'url': '/digitaltwins/{id}/components/{componentPath}'}
def update_component(
self, id, component_path, patch_document, digital_twins_update_component_options=None, custom_headers=None, raw=False, **operation_config):
"""Updates a component on a digital twin.
Status codes:
* 204 No Content
* 400 Bad Request
* InvalidArgument - The digital twin id, component path, or payload is
invalid.
* JsonPatchInvalid - The JSON Patch provided is invalid.
* ValidationFailed - Applying the patch results in an invalid digital
twin.
* 404 Not Found
* DigitalTwinNotFound - The digital twin was not found.
* 412 Precondition Failed
* PreconditionFailed - The precondition check (If-Match or
If-None-Match) failed.
:param id: The id of the digital twin. The id is unique within the
service and case sensitive.
:type id: str
:param component_path: The name of the DTDL component.
:type component_path: str
:param patch_document: An update specification described by JSON
Patch. Updates to property values and $model elements may happen in
the same request. Operations are limited to add, replace and remove.
:type patch_document: list[object]
:param digital_twins_update_component_options: Additional parameters
for the operation
:type digital_twins_update_component_options:
~dataplane.models.DigitalTwinsUpdateComponentOptions
: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:`ErrorResponseException<dataplane.models.ErrorResponseException>`
"""
if_match = None
if digital_twins_update_component_options is not None:
if_match = digital_twins_update_component_options.if_match
traceparent = None
if digital_twins_update_component_options is not None:
traceparent = digital_twins_update_component_options.traceparent
tracestate = None
if digital_twins_update_component_options is not None:
tracestate = digital_twins_update_component_options.tracestate
# Construct URL
url = self.update_component.metadata['url']
path_format_arguments = {
'id': self._serialize.url("id", id, 'str'),
'componentPath': self._serialize.url("component_path", component_path, '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', min_length=1)
# Construct headers
header_parameters = {}
header_parameters['Content-Type'] = 'application/json-patch+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')
if if_match is not None:
header_parameters['If-Match'] = self._serialize.header("if_match", if_match, 'str')
if traceparent is not None:
header_parameters['traceparent'] = self._serialize.header("traceparent", traceparent, 'str')
if tracestate is not None:
header_parameters['tracestate'] = self._serialize.header("tracestate", tracestate, 'str')
# Construct body
body_content = self._serialize.body(patch_document, '[object]')
# Construct and send request
request = self._client.patch(url, query_parameters)
response = self._client.send(
request, header_parameters, body_content, stream=False, **operation_config)
# @vilit - added 202
if response.status_code not in [202, 204]:
raise models.ErrorResponseException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'ETag': 'str',
'x-ms-error-code': 'str',
})
return client_raw_response
update_component.metadata = {'url': '/digitaltwins/{id}/components/{componentPath}'}