def _invoke_internal()

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

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

    def _invoke_internal(self, function_arn, payload, client_context, invocation_type="RequestResponse"):
        """
        This private method is seperate from the main, public invoke method so that other code within this SDK can
        give this Lambda client a raw payload/client context to invoke with, rather than having it built for them.
        This lets you include custom ExtensionMap_ values like subject which are needed for our internal pinned Lambdas.
        """
        customer_logger.info('Invoking Lambda function "{}" with Greengrass Message "{}"'.format(function_arn, payload))

        try:
            invocation_id = self.ipc.post_work(function_arn, payload, client_context, invocation_type)

            if invocation_type == "Event":
                # TODO: Properly return errors based on BOTO response
                # https://boto3.readthedocs.io/en/latest/reference/services/lambda.html#Lambda.Client.invoke
                return {'Payload': b'', 'FunctionError': ''}

            work_result_output = self.ipc.get_work_result(function_arn, invocation_id)
            if not work_result_output.func_err:
                output_payload = StreamingBody(work_result_output.payload)
            else:
                output_payload = work_result_output.payload
            invoke_output = {
                'Payload': output_payload,
                'FunctionError': work_result_output.func_err,
            }
            return invoke_output
        except IPCException as e:
            customer_logger.exception(e)
            raise InvocationException('Failed to invoke function due to ' + str(e))