def _shadow_op()

in static/code/worker-safety/deeplens-lambda/greengrasssdk/IoTDataPlane.py [0:0]

• 22 lines of code
• 4 McCabe index (conditional complexity)

    def _shadow_op(self, op, thing_name, payload):
        topic = '$aws/things/{thing_name}/shadow/{op}'.format(thing_name=thing_name, op=op)
        function_arn = SHADOW_FUNCTION_ARN
        client_context = {
            'custom': {
                'subject': topic
            }
        }

        customer_logger.info('Calling shadow service on topic "{}" with payload "{}"'.format(topic, payload))
        response = self.lambda_client._invoke_internal(
            function_arn,
            payload,
            base64.b64encode(json.dumps(client_context).encode())
        )

        payload = response['Payload'].read()
        if response:
            response_payload_map = json.loads(payload.decode('utf-8'))
            if 'code' in response_payload_map and 'message' in response_payload_map:
                raise ShadowError('Request for shadow state returned error code {} with message "{}"'.format(
                    response_payload_map['code'], response_payload_map['message']
                ))

        return {'payload': payload}