iotidentity.py

# Copyright Amazon.com, Inc. or its affiliates. All rights reserved.
# SPDX-License-Identifier: Apache-2.0.

# This file is generated

importawsiot
importconcurrent.futures
importtyping

classIotIdentityClient(awsiot.MqttServiceClient):
"""

 An AWS IoT service that assists with provisioning a device and installing unique client certificates on it

 AWS Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html

 """

defpublish_create_certificate_from_csr(self, request, qos):
# type: (CreateCertificateFromCsrRequest, int) -> concurrent.futures.Future
"""

 Creates a certificate from a certificate signing request (CSR). AWS IoT provides client certificates that are signed by the Amazon Root certificate authority (CA). The new certificate has a PENDING_ACTIVATION status. When you call RegisterThing to provision a thing with this certificate, the certificate status changes to ACTIVE or INACTIVE as described in the template.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `CreateCertificateFromCsrRequest` instance.
 qos: The Quality of Service guarantee of this message

 Returns:
 A Future whose result will be None if the
 request is successfully published. The Future's result will be an
 exception if the request cannot be published.
 """

returnself._publish_operation(
topic='$aws/certificates/create-from-csr/json',
qos=qos,
payload=request.to_payload())

defpublish_create_keys_and_certificate(self, request, qos):
# type: (CreateKeysAndCertificateRequest, int) -> concurrent.futures.Future
"""

 Creates new keys and a certificate. AWS IoT provides client certificates that are signed by the Amazon Root certificate authority (CA). The new certificate has a PENDING_ACTIVATION status. When you call RegisterThing to provision a thing with this certificate, the certificate status changes to ACTIVE or INACTIVE as described in the template.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `CreateKeysAndCertificateRequest` instance.
 qos: The Quality of Service guarantee of this message

 Returns:
 A Future whose result will be None if the
 request is successfully published. The Future's result will be an
 exception if the request cannot be published.
 """

returnself._publish_operation(
topic='$aws/certificates/create/json',
qos=qos,
payload=None)

defpublish_register_thing(self, request, qos):
# type: (RegisterThingRequest, int) -> concurrent.futures.Future
"""

 Provisions an AWS IoT thing using a pre-defined template.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `RegisterThingRequest` instance.
 qos: The Quality of Service guarantee of this message

 Returns:
 A Future whose result will be None if the
 request is successfully published. The Future's result will be an
 exception if the request cannot be published.
 """
ifnotrequest.template_name:
raiseValueError("request.template_name is required")

returnself._publish_operation(
topic='$aws/provisioning-templates/{0.template_name}/provision/json'.format(request),
qos=qos,
payload=request.to_payload())

defsubscribe_to_create_certificate_from_csr_accepted(self, request, qos, callback):
# type: (CreateCertificateFromCsrSubscriptionRequest, int, typing.Callable[[CreateCertificateFromCsrResponse], None]) -> typing.Tuple[concurrent.futures.Future, str]
"""

 Subscribes to the accepted topic of the CreateCertificateFromCsr operation.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `CreateCertificateFromCsrSubscriptionRequest` instance.
 qos: The Quality of Service guarantee of this message
 callback: Callback to invoke each time the event is received.
 The callback should take 1 argument of type `CreateCertificateFromCsrResponse`.
 The callback is not expected to return anything.

 Returns:
 Tuple with two values. The first is a `Future` whose result will be the
 `awscrt.mqtt.QoS` granted by the server, or an exception if the
 subscription fails. The second value is a topic which may be passed
 to `unsubscribe()` to stop receiving messages. Note that messages
 may arrive before the subscription is acknowledged.
 """

ifnotcallable(callback):
raiseValueError("callback is required")

returnself._subscribe_operation(
topic='$aws/certificates/create-from-csr/json/accepted',
qos=qos,
callback=callback,
payload_to_class_fn=CreateCertificateFromCsrResponse.from_payload)

defsubscribe_to_create_certificate_from_csr_rejected(self, request, qos, callback):
# type: (CreateCertificateFromCsrSubscriptionRequest, int, typing.Callable[[ErrorResponse], None]) -> typing.Tuple[concurrent.futures.Future, str]
"""

 Subscribes to the rejected topic of the CreateCertificateFromCsr operation.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `CreateCertificateFromCsrSubscriptionRequest` instance.
 qos: The Quality of Service guarantee of this message
 callback: Callback to invoke each time the event is received.
 The callback should take 1 argument of type `ErrorResponse`.
 The callback is not expected to return anything.

 Returns:
 Tuple with two values. The first is a `Future` whose result will be the
 `awscrt.mqtt.QoS` granted by the server, or an exception if the
 subscription fails. The second value is a topic which may be passed
 to `unsubscribe()` to stop receiving messages. Note that messages
 may arrive before the subscription is acknowledged.
 """

ifnotcallable(callback):
raiseValueError("callback is required")

returnself._subscribe_operation(
topic='$aws/certificates/create-from-csr/json/rejected',
qos=qos,
callback=callback,
payload_to_class_fn=ErrorResponse.from_payload)

defsubscribe_to_create_keys_and_certificate_accepted(self, request, qos, callback):
# type: (CreateKeysAndCertificateSubscriptionRequest, int, typing.Callable[[CreateKeysAndCertificateResponse], None]) -> typing.Tuple[concurrent.futures.Future, str]
"""

 Subscribes to the accepted topic of the CreateKeysAndCertificate operation.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `CreateKeysAndCertificateSubscriptionRequest` instance.
 qos: The Quality of Service guarantee of this message
 callback: Callback to invoke each time the event is received.
 The callback should take 1 argument of type `CreateKeysAndCertificateResponse`.
 The callback is not expected to return anything.

 Returns:
 Tuple with two values. The first is a `Future` whose result will be the
 `awscrt.mqtt.QoS` granted by the server, or an exception if the
 subscription fails. The second value is a topic which may be passed
 to `unsubscribe()` to stop receiving messages. Note that messages
 may arrive before the subscription is acknowledged.
 """

ifnotcallable(callback):
raiseValueError("callback is required")

returnself._subscribe_operation(
topic='$aws/certificates/create/json/accepted',
qos=qos,
callback=callback,
payload_to_class_fn=CreateKeysAndCertificateResponse.from_payload)

defsubscribe_to_create_keys_and_certificate_rejected(self, request, qos, callback):
# type: (CreateKeysAndCertificateSubscriptionRequest, int, typing.Callable[[ErrorResponse], None]) -> typing.Tuple[concurrent.futures.Future, str]
"""

 Subscribes to the rejected topic of the CreateKeysAndCertificate operation.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `CreateKeysAndCertificateSubscriptionRequest` instance.
 qos: The Quality of Service guarantee of this message
 callback: Callback to invoke each time the event is received.
 The callback should take 1 argument of type `ErrorResponse`.
 The callback is not expected to return anything.

 Returns:
 Tuple with two values. The first is a `Future` whose result will be the
 `awscrt.mqtt.QoS` granted by the server, or an exception if the
 subscription fails. The second value is a topic which may be passed
 to `unsubscribe()` to stop receiving messages. Note that messages
 may arrive before the subscription is acknowledged.
 """

ifnotcallable(callback):
raiseValueError("callback is required")

returnself._subscribe_operation(
topic='$aws/certificates/create/json/rejected',
qos=qos,
callback=callback,
payload_to_class_fn=ErrorResponse.from_payload)

defsubscribe_to_register_thing_accepted(self, request, qos, callback):
# type: (RegisterThingSubscriptionRequest, int, typing.Callable[[RegisterThingResponse], None]) -> typing.Tuple[concurrent.futures.Future, str]
"""

 Subscribes to the accepted topic of the RegisterThing operation.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `RegisterThingSubscriptionRequest` instance.
 qos: The Quality of Service guarantee of this message
 callback: Callback to invoke each time the event is received.
 The callback should take 1 argument of type `RegisterThingResponse`.
 The callback is not expected to return anything.

 Returns:
 Tuple with two values. The first is a `Future` whose result will be the
 `awscrt.mqtt.QoS` granted by the server, or an exception if the
 subscription fails. The second value is a topic which may be passed
 to `unsubscribe()` to stop receiving messages. Note that messages
 may arrive before the subscription is acknowledged.
 """
ifnotrequest.template_name:
raiseValueError("request.template_name is required")

ifnotcallable(callback):
raiseValueError("callback is required")

returnself._subscribe_operation(
topic='$aws/provisioning-templates/{0.template_name}/provision/json/accepted'.format(request),
qos=qos,
callback=callback,
payload_to_class_fn=RegisterThingResponse.from_payload)

defsubscribe_to_register_thing_rejected(self, request, qos, callback):
# type: (RegisterThingSubscriptionRequest, int, typing.Callable[[ErrorResponse], None]) -> typing.Tuple[concurrent.futures.Future, str]
"""

 Subscribes to the rejected topic of the RegisterThing operation.

 API Docs: https://docs.aws.amazon.com/iot/latest/developerguide/provision-wo-cert.html#fleet-provision-api

 Args:
 request: `RegisterThingSubscriptionRequest` instance.
 qos: The Quality of Service guarantee of this message
 callback: Callback to invoke each time the event is received.
 The callback should take 1 argument of type `ErrorResponse`.
 The callback is not expected to return anything.

 Returns:
 Tuple with two values. The first is a `Future` whose result will be the
 `awscrt.mqtt.QoS` granted by the server, or an exception if the
 subscription fails. The second value is a topic which may be passed
 to `unsubscribe()` to stop receiving messages. Note that messages
 may arrive before the subscription is acknowledged.
 """
ifnotrequest.template_name:
raiseValueError("request.template_name is required")

ifnotcallable(callback):
raiseValueError("callback is required")

returnself._subscribe_operation(
topic='$aws/provisioning-templates/{0.template_name}/provision/json/rejected'.format(request),
qos=qos,
callback=callback,
payload_to_class_fn=ErrorResponse.from_payload)

classCreateCertificateFromCsrRequest(awsiot.ModeledClass):
"""

 Data needed to perform a CreateCertificateFromCsr operation.

 All attributes are None by default, and may be set by keyword in the constructor.

 Keyword Args:
 certificate_signing_request (str): The CSR, in PEM format.

 Attributes:
 certificate_signing_request (str): The CSR, in PEM format.
 """

__slots__= ['certificate_signing_request']

def__init__(self, *args, **kwargs):
self.certificate_signing_request=kwargs.get('certificate_signing_request')

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip(['certificate_signing_request'], args):
setattr(self, key, val)

defto_payload(self):
# type: () -> typing.Dict[str, typing.Any]
payload= {} # type: typing.Dict[str, typing.Any]
ifself.certificate_signing_requestisnotNone:
payload['certificateSigningRequest'] =self.certificate_signing_request
returnpayload

classCreateCertificateFromCsrResponse(awsiot.ModeledClass):
"""

 Response payload to a CreateCertificateFromCsr request.

 All attributes are None by default, and may be set by keyword in the constructor.

 Keyword Args:
 certificate_id (str): The ID of the certificate.
 certificate_ownership_token (str): The token to prove ownership of the certificate during provisioning.
 certificate_pem (str): The certificate data, in PEM format.

 Attributes:
 certificate_id (str): The ID of the certificate.
 certificate_ownership_token (str): The token to prove ownership of the certificate during provisioning.
 certificate_pem (str): The certificate data, in PEM format.
 """

__slots__= ['certificate_id', 'certificate_ownership_token', 'certificate_pem']

def__init__(self, *args, **kwargs):
self.certificate_id=kwargs.get('certificate_id')
self.certificate_ownership_token=kwargs.get('certificate_ownership_token')
self.certificate_pem=kwargs.get('certificate_pem')

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip(['certificate_id', 'certificate_ownership_token', 'certificate_pem'], args):
setattr(self, key, val)

@classmethod
deffrom_payload(cls, payload):
# type: (typing.Dict[str, typing.Any]) -> CreateCertificateFromCsrResponse
new=cls()
val=payload.get('certificateId')
ifvalisnotNone:
new.certificate_id=val
val=payload.get('certificateOwnershipToken')
ifvalisnotNone:
new.certificate_ownership_token=val
val=payload.get('certificatePem')
ifvalisnotNone:
new.certificate_pem=val
returnnew

classCreateCertificateFromCsrSubscriptionRequest(awsiot.ModeledClass):
"""

 Data needed to subscribe to the responses of the CreateCertificateFromCsr operation.

 This class has no attributes.
 """

__slots__= []

def__init__(self, *args, **kwargs):

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip([], args):
setattr(self, key, val)

classCreateKeysAndCertificateRequest(awsiot.ModeledClass):
"""

 Data needed to perform a CreateKeysAndCertificate operation.

 This class has no attributes.
 """

__slots__= []

def__init__(self, *args, **kwargs):

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip([], args):
setattr(self, key, val)

classCreateKeysAndCertificateResponse(awsiot.ModeledClass):
"""

 Response payload to a CreateKeysAndCertificate request.

 All attributes are None by default, and may be set by keyword in the constructor.

 Keyword Args:
 certificate_id (str): The certificate id.
 certificate_ownership_token (str): The token to prove ownership of the certificate during provisioning.
 certificate_pem (str): The certificate data, in PEM format.
 private_key (str): The private key.

 Attributes:
 certificate_id (str): The certificate id.
 certificate_ownership_token (str): The token to prove ownership of the certificate during provisioning.
 certificate_pem (str): The certificate data, in PEM format.
 private_key (str): The private key.
 """

__slots__= ['certificate_id', 'certificate_ownership_token', 'certificate_pem', 'private_key']

def__init__(self, *args, **kwargs):
self.certificate_id=kwargs.get('certificate_id')
self.certificate_ownership_token=kwargs.get('certificate_ownership_token')
self.certificate_pem=kwargs.get('certificate_pem')
self.private_key=kwargs.get('private_key')

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip(['certificate_id', 'certificate_ownership_token', 'certificate_pem', 'private_key'], args):
setattr(self, key, val)

@classmethod
deffrom_payload(cls, payload):
# type: (typing.Dict[str, typing.Any]) -> CreateKeysAndCertificateResponse
new=cls()
val=payload.get('certificateId')
ifvalisnotNone:
new.certificate_id=val
val=payload.get('certificateOwnershipToken')
ifvalisnotNone:
new.certificate_ownership_token=val
val=payload.get('certificatePem')
ifvalisnotNone:
new.certificate_pem=val
val=payload.get('privateKey')
ifvalisnotNone:
new.private_key=val
returnnew

classCreateKeysAndCertificateSubscriptionRequest(awsiot.ModeledClass):
"""

 Data needed to subscribe to the responses of the CreateKeysAndCertificate operation.

 This class has no attributes.
 """

__slots__= []

def__init__(self, *args, **kwargs):

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip([], args):
setattr(self, key, val)

classErrorResponse(awsiot.ModeledClass):
"""

 Response document containing details about a failed request.

 All attributes are None by default, and may be set by keyword in the constructor.

 Keyword Args:
 error_code (str): Response error code
 error_message (str): Response error message
 status_code (int): Response status code

 Attributes:
 error_code (str): Response error code
 error_message (str): Response error message
 status_code (int): Response status code
 """

__slots__= ['error_code', 'error_message', 'status_code']

def__init__(self, *args, **kwargs):
self.error_code=kwargs.get('error_code')
self.error_message=kwargs.get('error_message')
self.status_code=kwargs.get('status_code')

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip(['error_code', 'error_message', 'status_code'], args):
setattr(self, key, val)

@classmethod
deffrom_payload(cls, payload):
# type: (typing.Dict[str, typing.Any]) -> ErrorResponse
new=cls()
val=payload.get('errorCode')
ifvalisnotNone:
new.error_code=val
val=payload.get('errorMessage')
ifvalisnotNone:
new.error_message=val
val=payload.get('statusCode')
ifvalisnotNone:
new.status_code=val
returnnew

classRegisterThingRequest(awsiot.ModeledClass):
"""

 Data needed to perform a RegisterThing operation.

 All attributes are None by default, and may be set by keyword in the constructor.

 Keyword Args:
 certificate_ownership_token (str): The token to prove ownership of the certificate. The token is generated by AWS IoT when you create a certificate over MQTT.
 parameters (typing.Dict[str, str]): Optional. Key-value pairs from the device that are used by the pre-provisioning hooks to evaluate the registration request.
 template_name (str): The provisioning template name.

 Attributes:
 certificate_ownership_token (str): The token to prove ownership of the certificate. The token is generated by AWS IoT when you create a certificate over MQTT.
 parameters (typing.Dict[str, str]): Optional. Key-value pairs from the device that are used by the pre-provisioning hooks to evaluate the registration request.
 template_name (str): The provisioning template name.
 """

__slots__= ['certificate_ownership_token', 'parameters', 'template_name']

def__init__(self, *args, **kwargs):
self.certificate_ownership_token=kwargs.get('certificate_ownership_token')
self.parameters=kwargs.get('parameters')
self.template_name=kwargs.get('template_name')

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip(['certificate_ownership_token', 'parameters', 'template_name'], args):
setattr(self, key, val)

defto_payload(self):
# type: () -> typing.Dict[str, typing.Any]
payload= {} # type: typing.Dict[str, typing.Any]
ifself.certificate_ownership_tokenisnotNone:
payload['certificateOwnershipToken'] =self.certificate_ownership_token
ifself.parametersisnotNone:
payload['parameters'] =self.parameters
returnpayload

classRegisterThingResponse(awsiot.ModeledClass):
"""

 Response payload to a RegisterThing request.

 All attributes are None by default, and may be set by keyword in the constructor.

 Keyword Args:
 device_configuration (typing.Dict[str, str]): The device configuration defined in the template.
 thing_name (str): The name of the IoT thing created during provisioning.

 Attributes:
 device_configuration (typing.Dict[str, str]): The device configuration defined in the template.
 thing_name (str): The name of the IoT thing created during provisioning.
 """

__slots__= ['device_configuration', 'thing_name']

def__init__(self, *args, **kwargs):
self.device_configuration=kwargs.get('device_configuration')
self.thing_name=kwargs.get('thing_name')

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip(['device_configuration', 'thing_name'], args):
setattr(self, key, val)

@classmethod
deffrom_payload(cls, payload):
# type: (typing.Dict[str, typing.Any]) -> RegisterThingResponse
new=cls()
val=payload.get('deviceConfiguration')
ifvalisnotNone:
new.device_configuration=val
val=payload.get('thingName')
ifvalisnotNone:
new.thing_name=val
returnnew

classRegisterThingSubscriptionRequest(awsiot.ModeledClass):
"""

 Data needed to subscribe to the responses of the RegisterThing operation.

 All attributes are None by default, and may be set by keyword in the constructor.

 Keyword Args:
 template_name (str): Name of the provisioning template to listen for RegisterThing responses for.

 Attributes:
 template_name (str): Name of the provisioning template to listen for RegisterThing responses for.
 """

__slots__= ['template_name']

def__init__(self, *args, **kwargs):
self.template_name=kwargs.get('template_name')

# for backwards compatibility, read any arguments that used to be accepted by position
forkey, valinzip(['template_name'], args):
setattr(self, key, val)