Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_lambda_callable method in localstack

Best Python code snippet using localstack_python

lambda_executors.py

Source:lambda_executors.py

...1146            environment["AWS_LAMBDA_FUNCTION_INVOKED_ARN"] = context.invoked_function_arn1147            environment["AWS_LAMBDA_FUNCTION_MEMORY_SIZE"] = str(context.memory_limit_in_mb)1148        # execute the Lambda function in a forked sub-process, sync result via queue1149        queue = Queue()1150        lambda_function_callable = self.get_lambda_callable(1151            lambda_function, qualifier=inv_context.function_version1152        )1153        def do_execute():1154            # now we're executing in the child process, safe to change CWD and ENV1155            result = None1156            try:1157                if lambda_cwd:1158                    os.chdir(lambda_cwd)1159                    sys.path.insert(0, "")1160                if environment:1161                    os.environ.update(environment)1162                # set default env variables required for most Lambda handlers1163                self.set_default_env_variables()1164                # run the actual handler function1165                result = lambda_function_callable(inv_context.event, context)1166            except Exception as e:1167                result = str(e)1168                sys.stderr.write("%s %s" % (e, traceback.format_exc()))1169                raise1170            finally:1171                queue.put(result)1172        process = Process(target=do_execute)1173        start_time = now(millis=True)1174        error = None1175        with CaptureOutput() as c:1176            try:1177                process.run()1178            except Exception as e:1179                error = e1180        result = queue.get()1181        end_time = now(millis=True)1182        # Make sure to keep the log line below, to ensure the log stream gets created1183        request_id = long_uid()1184        log_output = 'START %s: Lambda %s started via "local" executor ...' % (1185            request_id,1186            lambda_function.arn(),1187        )1188        # TODO: Interweaving stdout/stderr currently not supported1189        for stream in (c.stdout(), c.stderr()):1190            if stream:1191                log_output += ("\n" if log_output else "") + stream1192        if isinstance(result, InvocationResult) and result.log_output:1193            log_output += "\n" + result.log_output1194        log_output += "\nEND RequestId: %s" % request_id1195        log_output += "\nREPORT RequestId: %s Duration: %s ms" % (1196            request_id,1197            int((end_time - start_time) * 1000),1198        )1199        # store logs to CloudWatch1200        store_lambda_logs(lambda_function, log_output)1201        result = result.result if isinstance(result, InvocationResult) else result1202        if error:1203            LOG.info(1204                'Error executing Lambda "%s": %s %s',1205                lambda_function.arn(),1206                error,1207                "".join(traceback.format_tb(error.__traceback__)),1208            )1209            result = json.dumps(1210                {1211                    "errorType": error.__class__.__name__,1212                    "errorMessage": error.args[0],1213                    "stackTrace": traceback.format_tb(error.__traceback__),1214                }1215            )1216            raise InvocationException(result, log_output=log_output, result=result)1217        # construct final invocation result1218        invocation_result = InvocationResult(result, log_output=log_output)1219        # run plugins post-processing logic1220        invocation_result = self.process_result_via_plugins(inv_context, invocation_result)1221        return invocation_result1222    def provide_file_to_lambda(self, local_file: str, inv_context: InvocationContext) -> str:1223        # This is a no-op for local executors - simply return the given local file path1224        return local_file1225    def execute_java_lambda(1226        self, event, context, main_file, lambda_function: LambdaFunction = None1227    ) -> InvocationResult:1228        lambda_function.envvars = lambda_function.envvars or {}1229        java_opts = config.LAMBDA_JAVA_OPTS or ""1230        handler = lambda_function.handler1231        lambda_function.envvars[LAMBDA_HANDLER_ENV_VAR_NAME] = handler1232        event_file = EVENT_FILE_PATTERN.replace("*", short_uid())1233        save_file(event_file, json.dumps(json_safe(event)))1234        TMP_FILES.append(event_file)1235        classpath = "%s:%s:%s" % (1236            main_file,1237            Util.get_java_classpath(main_file),1238            LAMBDA_EXECUTOR_JAR,1239        )1240        cmd = "java %s -cp %s %s %s" % (1241            java_opts,1242            classpath,1243            LAMBDA_EXECUTOR_CLASS,1244            event_file,1245        )1246        # apply plugin patches1247        inv_context = InvocationContext(1248            lambda_function, event, environment=lambda_function.envvars, lambda_command=cmd1249        )1250        result = self.apply_plugin_patches(inv_context)1251        if isinstance(result, InvocationResult):1252            return result1253        cmd = inv_context.lambda_command1254        LOG.info(cmd)1255        # execute Lambda and get invocation result1256        invocation_result = self._execute_in_custom_runtime(cmd, lambda_function=lambda_function)1257        return invocation_result1258    def execute_javascript_lambda(1259        self, event, context, main_file, lambda_function: LambdaFunction = None1260    ):1261        handler = lambda_function.handler1262        function = handler.split(".")[-1]1263        event_json_string = "%s" % (json.dumps(json_safe(event)) if event else "{}")1264        context_json_string = "%s" % (json.dumps(context.__dict__) if context else "{}")1265        cmd = [1266            "node",1267            "-e",1268            f'const res = require("{main_file}").{function}({event_json_string},{context_json_string}); '1269            f"const log = (rs) => console.log(JSON.stringify(rs)); "1270            "res && res.then ? res.then(r => log(r)) : log(res)",1271        ]1272        LOG.info(cmd)1273        result = self._execute_in_custom_runtime(cmd, lambda_function=lambda_function)1274        return result1275    def execute_go_lambda(self, event, context, main_file, lambda_function: LambdaFunction = None):1276        if lambda_function:1277            lambda_function.envvars["AWS_LAMBDA_FUNCTION_HANDLER"] = main_file1278            lambda_function.envvars["AWS_LAMBDA_EVENT_BODY"] = json.dumps(json_safe(event))1279        else:1280            LOG.warning("Unable to get function details for local execution of Golang Lambda")1281        cmd = GO_LAMBDA_RUNTIME1282        LOG.debug("Running Golang Lambda with runtime: %s", cmd)1283        result = self._execute_in_custom_runtime(cmd, lambda_function=lambda_function)1284        return result1285    @staticmethod1286    def set_default_env_variables():1287        # set default env variables required for most Lambda handlers1288        default_env_vars = {"AWS_DEFAULT_REGION": aws_stack.get_region()}1289        env_vars_before = {var: os.environ.get(var) for var in default_env_vars}1290        os.environ.update({k: v for k, v in default_env_vars.items() if not env_vars_before.get(k)})1291        return env_vars_before1292    @staticmethod1293    def reset_default_env_variables(env_vars_before):1294        for env_name, env_value in env_vars_before.items():1295            env_value_before = env_vars_before.get(env_name)1296            os.environ[env_name] = env_value_before or ""1297            if env_value_before is None:1298                os.environ.pop(env_name, None)1299    @classmethod1300    def get_lambda_callable(cls, function: LambdaFunction, qualifier: str = None) -> Callable:1301        """Returns the function Callable for invoking the given function locally"""1302        qualifier = function.get_qualifier_version(qualifier)1303        func_dict = cls.FUNCTION_CALLABLES.get(function.arn()) or {}1304        # TODO: function versioning and qualifiers should be refactored and designed properly!1305        callable = func_dict.get(qualifier) or func_dict.get(LambdaFunction.QUALIFIER_LATEST)1306        if not callable:1307            raise Exception(1308                f"Unable to find callable for Lambda function {function.arn()} - {qualifier}"1309            )1310        return callable1311    @classmethod1312    def add_function_callable(cls, function: LambdaFunction, lambda_handler: Callable):1313        """Sets the function Callable for invoking the $LATEST version of the Lambda function."""1314        func_dict = cls.FUNCTION_CALLABLES.setdefault(function.arn(), {})...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.