Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use process_lambda_url_invocation method in localstack

Best Python code snippet using localstack_python
lambda_api.py
Source:lambda_api.py
...286            continue287        for key, value in dict(attr).items():288            attr[first_char_to_lower(key)] = attr.pop(key)289    return message_attrs290def process_lambda_url_invocation(lambda_url_config: dict, event: dict):291    inv_result = run_lambda(292        func_arn=lambda_url_config["FunctionArn"],293        event=event,294        asynchronous=False,295    )296    return inv_result.result297def get_event_sources(func_name=None, source_arn=None):298    result = []299    for region, details in LambdaRegion.regions().items():300        for m in details.event_source_mappings:301            if not func_name or (m["FunctionArn"] in [func_name, func_arn(func_name)]):302                if event_source_arn_matches(mapped=m.get("EventSourceArn"), searched=source_arn):303                    result.append(m)304    return result305def get_function_version(arn, version):306    region = LambdaRegion.get()307    func = region.lambdas.get(arn)308    return format_func_details(func, version=version, always_add_version=True)309def publish_new_function_version(arn):310    region = LambdaRegion.get()311    lambda_function = region.lambdas.get(arn)312    versions = lambda_function.versions313    max_version_number = lambda_function.max_version()314    next_version_number = max_version_number + 1315    latest_hash = versions.get(VERSION_LATEST).get("CodeSha256")316    max_version = versions.get(str(max_version_number))317    max_version_hash = max_version.get("CodeSha256") if max_version else ""318    if latest_hash != max_version_hash:319        versions[str(next_version_number)] = {320            "CodeSize": versions.get(VERSION_LATEST).get("CodeSize"),321            "CodeSha256": versions.get(VERSION_LATEST).get("CodeSha256"),322            "Function": versions.get(VERSION_LATEST).get("Function"),323            "RevisionId": str(uuid.uuid4()),324        }325        max_version_number = next_version_number326    return get_function_version(arn, str(max_version_number))327def do_list_versions(arn):328    region = LambdaRegion.get()329    versions = [330        get_function_version(arn, version) for version in region.lambdas.get(arn).versions.keys()331    ]332    return sorted(versions, key=lambda k: str(k.get("Version")))333def do_update_alias(arn, alias, version, description=None):334    region = LambdaRegion.get()335    new_alias = {336        "AliasArn": arn + ":" + alias,337        "FunctionVersion": version,338        "Name": alias,339        "Description": description or "",340        "RevisionId": str(uuid.uuid4()),341    }342    region.lambdas.get(arn).aliases[alias] = new_alias343    return new_alias344def run_lambda(345    func_arn,346    event,347    context=None,348    version=None,349    suppress_output=False,350    asynchronous=False,351    callback=None,352    lock_discriminator: str = None,353) -> InvocationResult:354    if context is None:355        context = {}356    # Ensure that the service provider has been initialized. This is required to ensure all lifecycle hooks357    # (e.g., persistence) have been executed when the run_lambda(..) function gets called (e.g., from API GW).358    if not hasattr(run_lambda, "_provider_initialized"):359        aws_stack.connect_to_service("lambda").list_functions()360        run_lambda._provider_initialized = True361    region_name = extract_region_from_arn(func_arn)362    region = LambdaRegion.get(region_name)363    if suppress_output:364        stdout_ = sys.stdout365        stderr_ = sys.stderr366        stream = StringIO()367        sys.stdout = stream368        sys.stderr = stream369    try:370        func_arn = aws_stack.fix_arn(func_arn)371        lambda_function = region.lambdas.get(func_arn)372        if not lambda_function:373            LOG.debug("Unable to find details for Lambda %s in region %s", func_arn, region_name)374            result = not_found_error(msg="The resource specified in the request does not exist.")375            return InvocationResult(result)376        if lambda_function.state != "Active":377            result = error_response(378                f"The operation cannot be performed at this time. The function is currently in the following state: {lambda_function.state}",379                409,380                "ResourceConflictException",381            )382            raise ClientError(result)383        context = LambdaContext(lambda_function, version, context)384        result = LAMBDA_EXECUTOR.execute(385            func_arn,386            lambda_function,387            event,388            context=context,389            version=version,390            asynchronous=asynchronous,391            callback=callback,392            lock_discriminator=lock_discriminator,393        )394        return result395    except ClientError:396        raise397    except Exception as e:398        exc_type, exc_value, exc_traceback = sys.exc_info()399        response = {400            "errorType": str(exc_type.__name__),401            "errorMessage": str(e),402            "stackTrace": traceback.format_tb(exc_traceback),403        }404        LOG.info("Error executing Lambda function %s: %s %s", func_arn, e, traceback.format_exc())405        if isinstance(e, lambda_executors.InvocationException):406            exc_result = e.result407            response = run_safe(lambda: json.loads(exc_result)) or response408        log_output = e.log_output if isinstance(e, lambda_executors.InvocationException) else ""409        return InvocationResult(Response(json.dumps(response), status=500), log_output)410    finally:411        if suppress_output:412            sys.stdout = stdout_413            sys.stderr = stderr_414def load_source(name, file):415    return importlib.machinery.SourceFileLoader(name, file).load_module()416def exec_lambda_code(script, handler_function="handler", lambda_cwd=None, lambda_env=None):417    # TODO: The code in this function is generally not thread-safe and potentially insecure418    #  (e.g., mutating environment variables, and globally loaded modules). Should be redesigned.419    def _do_exec_lambda_code():420        if lambda_cwd or lambda_env:421            if lambda_cwd:422                previous_cwd = os.getcwd()423                os.chdir(lambda_cwd)424                sys.path = [lambda_cwd] + sys.path425            if lambda_env:426                previous_env = dict(os.environ)427                os.environ.update(lambda_env)428        # generate lambda file name429        lambda_id = "l_%s" % short_uid()430        lambda_file = LAMBDA_SCRIPT_PATTERN.replace("*", lambda_id)431        save_file(lambda_file, script)432        # delete temporary .py and .pyc files on exit433        TMP_FILES.append(lambda_file)434        TMP_FILES.append("%sc" % lambda_file)435        try:436            pre_sys_modules_keys = set(sys.modules.keys())437            # set default env variables required for most Lambda handlers438            env_vars_before = lambda_executors.LambdaExecutorLocal.set_default_env_variables()439            try:440                handler_module = load_source(lambda_id, lambda_file)441                module_vars = handler_module.__dict__442            finally:443                lambda_executors.LambdaExecutorLocal.reset_default_env_variables(env_vars_before)444                # the above import can bring files for the function445                # (eg settings.py) into the global namespace. subsequent446                # calls can pick up file from another function, causing447                # general issues.448                post_sys_modules_keys = set(sys.modules.keys())449                for key in post_sys_modules_keys:450                    if key not in pre_sys_modules_keys:451                        sys.modules.pop(key)452        except Exception as e:453            LOG.error("Unable to exec: %s %s", script, traceback.format_exc())454            raise e455        finally:456            if lambda_cwd or lambda_env:457                if lambda_cwd:458                    os.chdir(previous_cwd)459                    sys.path.pop(0)460                if lambda_env:461                    os.environ = previous_env462        return module_vars[handler_function]463    lock = EXEC_MUTEX if lambda_cwd or lambda_env else empty_context_manager()464    with lock:465        return _do_exec_lambda_code()466def get_handler_function_from_name(handler_name, runtime=None):467    runtime = runtime or LAMBDA_DEFAULT_RUNTIME468    if runtime.startswith(tuple(DOTNET_LAMBDA_RUNTIMES)):469        return handler_name.split(":")[-1]470    return handler_name.split(".")[-1]471def get_java_handler(zip_file_content, main_file, lambda_function=None):472    """Creates a Java handler from an uploaded ZIP or JAR.473    :type zip_file_content: bytes474    :param zip_file_content: ZIP file bytes.475    :type handler: str476    :param handler: The lambda handler path.477    :type main_file: str478    :param main_file: Filepath to the uploaded ZIP or JAR file.479    :returns: function or flask.Response480    """481    if is_zip_file(zip_file_content):482        def execute(event, context):483            result = lambda_executors.EXECUTOR_LOCAL.execute_java_lambda(484                event, context, main_file=main_file, lambda_function=lambda_function485            )486            return result487        return execute488    raise ClientError(489        error_response(490            "Unable to extract Java Lambda handler - file is not a valid zip/jar file (%s, %s bytes)"491            % (main_file, len(zip_file_content or "")),492            400,493            error_type="ValidationError",494        )495    )496def set_archive_code(code: Dict, lambda_name: str, zip_file_content: bytes = None) -> Optional[str]:497    region = LambdaRegion.get()498    # get metadata499    lambda_arn = func_arn(lambda_name)500    lambda_details = region.lambdas[lambda_arn]501    is_local_mount = code.get("S3Bucket") == config.BUCKET_MARKER_LOCAL502    if is_local_mount and config.LAMBDA_REMOTE_DOCKER:503        msg = 'Please note that Lambda mounts (bucket name "%s") cannot be used with LAMBDA_REMOTE_DOCKER=1'504        raise Exception(msg % config.BUCKET_MARKER_LOCAL)505    # Stop/remove any containers that this arn uses.506    LAMBDA_EXECUTOR.cleanup(lambda_arn)507    if is_local_mount:508        # Mount or use a local folder lambda executors can reference509        # WARNING: this means we're pointing lambda_cwd to a local path in the user's510        # file system! We must ensure that there is no data loss (i.e., we must *not* add511        # this folder to TMP_FILES or similar).512        lambda_details.cwd = code.get("S3Key")513        return code["S3Key"]514    # get file content515    zip_file_content = zip_file_content or get_zip_bytes(code)516    if not zip_file_content:517        return518    # Save the zip file to a temporary file that the lambda executors can reference519    code_sha_256 = base64.standard_b64encode(hashlib.sha256(zip_file_content).digest())520    latest_version = lambda_details.get_version(VERSION_LATEST)521    latest_version["CodeSize"] = len(zip_file_content)522    latest_version["CodeSha256"] = code_sha_256.decode("utf-8")523    zip_dir_name = f"function.zipfile.{short_uid()}"524    zip_dir = f"{config.dirs.tmp}/{zip_dir_name}"525    mkdir(zip_dir)526    tmp_file = f"{zip_dir}/{LAMBDA_ZIP_FILE_NAME}"527    save_file(tmp_file, zip_file_content)528    TMP_FILES.append(zip_dir)529    lambda_details.zip_dir = zip_dir530    lambda_details.cwd = f"{get_lambda_extraction_dir()}/{zip_dir_name}"531    mkdir(lambda_details.cwd)532    return zip_dir533def set_function_code(lambda_function: LambdaFunction):534    def _set_and_configure(*args, **kwargs):535        try:536            before = time.perf_counter()537            do_set_function_code(lambda_function)538            # initialize function code via plugins539            for plugin in lambda_executors.LambdaExecutorPlugin.get_plugins():540                plugin.init_function_code(lambda_function)541            lambda_function.state = "Active"542            LOG.debug(543                "Function code initialization for function '%s' complete. State => Active (in %.3fs)",544                lambda_function.name(),545                time.perf_counter() - before,546            )547        except Exception:548            lambda_function.state = "Failed"549            raise550    # unzipping can take some time - limit the execution time to avoid client/network timeout issues551    run_for_max_seconds(config.LAMBDA_CODE_EXTRACT_TIME, _set_and_configure)552    return {"FunctionName": lambda_function.name()}553def store_and_get_lambda_code_archive(554    lambda_function: LambdaFunction, zip_file_content: bytes = None555) -> Optional[Tuple[str, str, bytes]]:556    """Store the Lambda code referenced in the LambdaFunction details to disk as a zip file,557    and return the Lambda CWD, file name, and zip bytes content. May optionally return None558    in case this is a Lambda with the special bucket marker __local__, used for code mounting."""559    code_passed = lambda_function.code560    is_local_mount = code_passed.get("S3Bucket") == config.BUCKET_MARKER_LOCAL561    lambda_zip_dir = lambda_function.zip_dir562    if code_passed:563        lambda_zip_dir = lambda_zip_dir or set_archive_code(code_passed, lambda_function.name())564        if not zip_file_content and not is_local_mount:565            # Save the zip file to a temporary file that the lambda executors can reference566            zip_file_content = get_zip_bytes(code_passed)567    else:568        lambda_details = LambdaRegion.get().lambdas[lambda_function.arn()]569        lambda_zip_dir = lambda_zip_dir or lambda_details.zip_dir570    if not lambda_zip_dir:571        return572    # construct archive name573    archive_file = os.path.join(lambda_zip_dir, LAMBDA_ZIP_FILE_NAME)574    if not zip_file_content:575        zip_file_content = load_file(archive_file, mode="rb")576    else:577        # override lambda archive with fresh code if we got an update578        save_file(archive_file, zip_file_content)579    # remove content from code attribute, if present580    lambda_function.code.pop("ZipFile", None)581    return lambda_zip_dir, archive_file, zip_file_content582def do_set_function_code(lambda_function: LambdaFunction):583    """Main function that creates the local zip archive for the given Lambda function, and584    optionally creates the handler function references (for LAMBDA_EXECUTOR=local)"""585    def generic_handler(*_):586        raise ClientError(587            (588                'Unable to find executor for Lambda function "%s". Note that '589                + "Node.js, Golang, and .Net Core Lambdas currently require LAMBDA_EXECUTOR=docker"590            )591            % lambda_name592        )593    lambda_name = lambda_function.name()594    arn = lambda_function.arn()595    runtime = get_lambda_runtime(lambda_function)596    lambda_environment = lambda_function.envvars597    handler_name = lambda_function.handler = lambda_function.handler or LAMBDA_DEFAULT_HANDLER598    code_passed = lambda_function.code599    is_local_mount = code_passed.get("S3Bucket") == config.BUCKET_MARKER_LOCAL600    # cleanup any left-over Lambda executor instances601    LAMBDA_EXECUTOR.cleanup(arn)602    # get local Lambda code archive path603    _result = store_and_get_lambda_code_archive(lambda_function)604    if not _result:605        return606    lambda_zip_dir, archive_file, zip_file_content = _result607    lambda_cwd = lambda_function.cwd608    # Set the appropriate Lambda handler.609    lambda_handler = generic_handler610    is_java = lambda_executors.is_java_lambda(runtime)611    if is_java:612        # The Lambda executors for Docker subclass LambdaExecutorContainers, which613        # runs Lambda in Docker by passing all *.jar files in the function working614        # directory as part of the classpath. Obtain a Java handler function below.615        try:616            lambda_handler = get_java_handler(617                zip_file_content, archive_file, lambda_function=lambda_function618            )619        except Exception as e:620            # this can happen, e.g., for Lambda code mounted via __local__ -> ignore621            LOG.debug("Unable to determine Lambda Java handler: %s", e)622    if not is_local_mount:623        # Lambda code must be uploaded in Zip format624        if not is_zip_file(zip_file_content):625            raise ClientError(f"Uploaded Lambda code for runtime ({runtime}) is not in Zip format")626        # Unzip the Lambda archive contents627        if get_unzipped_size(archive_file) >= FUNCTION_MAX_UNZIPPED_SIZE:628            raise ClientError(629                error_response(630                    f"Unzipped size must be smaller than {FUNCTION_MAX_UNZIPPED_SIZE} bytes",631                    code=400,632                    error_type="InvalidParameterValueException",633                )634            )635        unzip(archive_file, lambda_cwd)636    # Obtain handler details for any non-Java Lambda function637    if not is_java:638        handler_file = get_handler_file_from_name(handler_name, runtime=runtime)639        main_file = f"{lambda_cwd}/{handler_file}"640        if CHECK_HANDLER_ON_CREATION and not os.path.exists(main_file):641            # Raise an error if (1) this is not a local mount lambda, or (2) we're642            # running Lambdas locally (not in Docker), or (3) we're using remote Docker.643            # -> We do *not* want to raise an error if we're using local mount in non-remote Docker644            if not is_local_mount or not use_docker() or config.LAMBDA_REMOTE_DOCKER:645                file_list = run(f'cd "{lambda_cwd}"; du -d 3 .')646                config_debug = f'Config for local mount, docker, remote: "{is_local_mount}", "{use_docker()}", "{config.LAMBDA_REMOTE_DOCKER}"'647                LOG.debug("Lambda archive content:\n%s", file_list)648                raise ClientError(649                    error_response(650                        f"Unable to find handler script ({main_file}) in Lambda archive. {config_debug}",651                        400,652                        error_type="ValidationError",653                    )654                )655        # TODO: init code below should be moved into LambdaExecutorLocal!656        if runtime.startswith("python") and not use_docker():657            try:658                # make sure the file is actually readable, then read contents659                ensure_readable(main_file)660                zip_file_content = load_file(main_file, mode="rb")661                # extract handler662                handler_function = get_handler_function_from_name(handler_name, runtime=runtime)663                lambda_handler = exec_lambda_code(664                    zip_file_content,665                    handler_function=handler_function,666                    lambda_cwd=lambda_cwd,667                    lambda_env=lambda_environment,668                )669            except Exception as e:670                raise ClientError("Unable to get handler function from lambda code: %s" % e)671        if runtime.startswith("node") and not use_docker():672            ensure_readable(main_file)673            def execute(event, context):674                result = lambda_executors.EXECUTOR_LOCAL.execute_javascript_lambda(675                    event, context, main_file=main_file, lambda_function=lambda_function676                )677                return result678            lambda_handler = execute679        if runtime.startswith("go1") and not use_docker():680            install_go_lambda_runtime()681            ensure_readable(main_file)682            def execute_go(event, context):683                result = lambda_executors.EXECUTOR_LOCAL.execute_go_lambda(684                    event, context, main_file=main_file, lambda_function=lambda_function685                )686                return result687            lambda_handler = execute_go688    if lambda_handler:689        lambda_executors.LambdaExecutorLocal.add_function_callable(lambda_function, lambda_handler)690    return lambda_handler691def do_list_functions():692    funcs = []693    region = LambdaRegion.get()694    this_region = aws_stack.get_region()695    for f_arn, func in region.lambdas.items():696        if type(func) != LambdaFunction:697            continue698        # filter out functions of current region699        func_region = extract_region_from_arn(f_arn)700        if func_region != this_region:701            continue702        func_name = f_arn.split(":function:")[-1]703        arn = func_arn(func_name)704        lambda_function = region.lambdas.get(arn)705        if not lambda_function:706            # this can happen if we're accessing Lambdas from a different region (ARN mismatch)707            continue708        details = format_func_details(lambda_function)709        details["Tags"] = func.tags710        funcs.append(details)711    return funcs712def format_func_details(713    lambda_function: LambdaFunction, version: str = None, always_add_version=False714) -> Dict[str, Any]:715    version = version or VERSION_LATEST716    func_version = lambda_function.get_version(version)717    result = {718        "CodeSha256": func_version.get("CodeSha256"),719        "Role": lambda_function.role,720        "KMSKeyArn": lambda_function.kms_key_arn,721        "Version": version,722        "VpcConfig": lambda_function.vpc_config,723        "FunctionArn": lambda_function.arn(),724        "FunctionName": lambda_function.name(),725        "CodeSize": func_version.get("CodeSize"),726        "Handler": lambda_function.handler,727        "Runtime": lambda_function.runtime,728        "Timeout": lambda_function.timeout,729        "Description": lambda_function.description,730        "MemorySize": lambda_function.memory_size,731        "LastModified": format_timestamp(lambda_function.last_modified),732        "TracingConfig": lambda_function.tracing_config or {"Mode": "PassThrough"},733        "RevisionId": func_version.get("RevisionId"),734        "State": lambda_function.state,735        "LastUpdateStatus": "Successful",736        "PackageType": lambda_function.package_type,737        "ImageConfig": getattr(lambda_function, "image_config", None),738        "Architectures": lambda_function.architectures,739    }740    if lambda_function.dead_letter_config:741        result["DeadLetterConfig"] = lambda_function.dead_letter_config742    if lambda_function.envvars:743        result["Environment"] = {"Variables": lambda_function.envvars}744    arn_parts = result["FunctionArn"].split(":")745    if (always_add_version or version != VERSION_LATEST) and len(arn_parts) <= 7:746        result["FunctionArn"] += ":%s" % version747    return result748def forward_to_fallback_url(func_arn, data):749    """If LAMBDA_FALLBACK_URL is configured, forward the invocation of this non-existing750    Lambda to the configured URL."""751    if not config.LAMBDA_FALLBACK_URL:752        return753    lambda_name = aws_stack.lambda_function_name(func_arn)754    if config.LAMBDA_FALLBACK_URL.startswith("dynamodb://"):755        table_name = urlparse(config.LAMBDA_FALLBACK_URL.replace("dynamodb://", "http://")).netloc756        dynamodb = aws_stack.connect_to_service("dynamodb")757        item = {758            "id": {"S": short_uid()},759            "timestamp": {"N": str(now_utc())},760            "payload": {"S": data},761            "function_name": {"S": lambda_name},762        }763        aws_stack.create_dynamodb_table(table_name, partition_key="id")764        dynamodb.put_item(TableName=table_name, Item=item)765        return ""766    if re.match(r"^https?://.+", config.LAMBDA_FALLBACK_URL):767        headers = {768            "lambda-function-name": lambda_name,769            "Content-Type": APPLICATION_JSON,770        }771        response = safe_requests.post(config.LAMBDA_FALLBACK_URL, data, headers=headers)772        content = response.content773        try:774            # parse the response into a dictionary to get details775            # like function error etc.776            content = json.loads(content)777        except Exception:778            pass779        return content780    raise ClientError("Unexpected value for LAMBDA_FALLBACK_URL: %s" % config.LAMBDA_FALLBACK_URL)781def get_lambda_policy(function, qualifier=None):782    iam_client = aws_stack.connect_to_service("iam")783    policies = iam_client.list_policies(Scope="Local", MaxItems=500)["Policies"]784    docs = []785    for p in policies:786        # !TODO: Cache policy documents instead of running N+1 API calls here!787        versions = iam_client.list_policy_versions(PolicyArn=p["Arn"])["Versions"]788        default_version = [v for v in versions if v.get("IsDefaultVersion")]789        versions = default_version or versions790        doc = versions[0]["Document"]791        doc = doc if isinstance(doc, dict) else json.loads(doc)792        if not isinstance(doc["Statement"], list):793            doc["Statement"] = [doc["Statement"]]794        for stmt in doc["Statement"]:795            stmt["Principal"] = stmt.get("Principal") or {"AWS": get_aws_account_id()}796        doc["PolicyArn"] = p["Arn"]797        doc["PolicyName"] = p["PolicyName"]798        doc["Id"] = "default"799        docs.append(doc)800    # find policy by name801    policy_name = get_lambda_policy_name(802        aws_stack.lambda_function_name(function), qualifier=qualifier803    )804    policy = [d for d in docs if d["PolicyName"] == policy_name]805    if policy:806        return policy[0]807    # find policy by target Resource in statement (TODO: check if this heuristic holds in the general case)808    res_qualifier = func_qualifier(function, qualifier)809    policy = [d for d in docs if d["Statement"][0]["Resource"] == res_qualifier]810    return (policy or [None])[0]811def get_lambda_policy_name(resource_name: str, qualifier: str = None) -> str:812    qualifier = qualifier or "latest"813    if ":function:" in resource_name:814        resource_name = function_name_from_arn(resource_name)815    return LAMBDA_POLICY_NAME_PATTERN.format(name=resource_name, qualifier=qualifier)816def lookup_function(function, region, request_url):817    result = {818        "Configuration": function,819        "Code": {"Location": "%s/code" % request_url},820        "Tags": function["Tags"],821    }822    lambda_details = region.lambdas.get(function["FunctionArn"])823    # patch for image lambdas (still missing RepositoryType and ResolvedImageUri)824    # please note that usage is still only available with a PRO license825    if lambda_details.package_type == "Image":826        result["Code"] = lambda_details.code827        result["Configuration"]["CodeSize"] = 0828        result["Configuration"].pop("Handler", None)829        result["Configuration"].pop("Layers", None)830    if lambda_details.concurrency is not None:831        result["Concurrency"] = lambda_details.concurrency832    return jsonify(result)833def not_found_error(ref=None, msg=None):834    if not msg:835        msg = "The resource you requested does not exist."836        if ref:837            msg = "%s not found: %s" % (838                "Function" if ":function:" in ref else "Resource",839                ref,840            )841    return error_response(msg, 404, error_type="ResourceNotFoundException")842def delete_lambda_function(function_name: str) -> Dict[None, None]:843    arn = func_arn(function_name)844    region = LambdaRegion.get()845    # Stop/remove any containers that this arn uses.846    LAMBDA_EXECUTOR.cleanup(arn)847    try:848        region.lambdas.pop(arn)849    except KeyError:850        raise ResourceNotFoundException(851            f"Unable to delete non-existing Lambda function {func_arn(function_name)}"852        )853    i = 0854    while i < len(region.event_source_mappings):855        mapping = region.event_source_mappings[i]856        if mapping["FunctionArn"] == arn:857            del region.event_source_mappings[i]858            i -= 1859        i += 1860    return {}861def get_lambda_url_config(api_id, region=None):862    lambda_backend = LambdaRegion.get(region)863    url_configs = lambda_backend.url_configs.values()864    lambda_url_configs = [config for config in url_configs if config.get("CustomId") == api_id]865    return lambda_url_configs[0]866def event_for_lambda_url(api_id, path, data, headers, method) -> dict:867    raw_path = path.split("?")[0]868    raw_query_string = path.split("?")[1] if len(path.split("?")) > 1 else ""869    query_string_parameters = (870        {} if not raw_query_string else dict(urllib.parse.parse_qsl(raw_query_string))871    )872    now = datetime.utcnow()873    readable = timestamp(time=now, format=TIMESTAMP_READABLE_FORMAT)874    if not any(char in readable for char in ["+", "-"]):875        readable += "+0000"876    source_ip = headers.get("Remote-Addr", "")877    request_context = {878        "accountId": "anonymous",879        "apiId": api_id,880        "domainName": headers.get("Host", ""),881        "domainPrefix": api_id,882        "http": {883            "method": method,884            "path": raw_path,885            "protocol": "HTTP/1.1",886            "sourceIp": source_ip,887            "userAgent": headers.get("User-Agent", ""),888        },889        "requestId": long_uid(),890        "routeKey": "$default",891        "stage": "$default",892        "time": readable,893        "timeEpoch": mktime(ts=now, millis=True),894    }895    content_type = headers.get("Content-Type", "").lower()896    content_type_is_text = any(text_type in content_type for text_type in ["text", "json", "xml"])897    is_base64_encoded = not (data.isascii() and content_type_is_text) if data else False898    body = base64.b64encode(data).decode() if is_base64_encoded else data899    ignored_headers = ["connection", "x-localstack-tgt-api", "x-localstack-request-url"]900    event_headers = {k.lower(): v for k, v in headers.items() if k.lower() not in ignored_headers}901    event_headers.update(902        {903            "x-amzn-tls-cipher-suite": "ECDHE-RSA-AES128-GCM-SHA256",904            "x-amzn-tls-version": "TLSv1.2",905            "x-forwarded-proto": "http",906            "x-forwarded-for": source_ip,907            "x-forwarded-port": str(config.EDGE_PORT),908        }909    )910    event = {911        "version": "2.0",912        "routeKey": "$default",913        "rawPath": raw_path,914        "rawQueryString": raw_query_string,915        "headers": event_headers,916        "queryStringParameters": query_string_parameters,917        "requestContext": request_context,918        "body": body,919        "isBase64Encoded": is_base64_encoded,920    }921    if not data:922        event.pop("body")923    return event924def handle_lambda_url_invocation(925    request: Request, api_id: str, region: str, **url_params: Dict[str, str]926) -> HttpResponse:927    response = HttpResponse(headers={"Content-type": "application/json"})928    try:929        lambda_url_config = get_lambda_url_config(api_id, region)930    except IndexError as e:931        LOG.warning(f"Lambda URL ({api_id}) not found: {e}")932        response.set_json({"Message": None})933        response.status = "404"934        return response935    event = event_for_lambda_url(936        api_id, request.full_path, request.data, request.headers, request.method937    )938    try:939        result = process_lambda_url_invocation(lambda_url_config, event)940    except Exception as e:941        LOG.warning(f"Lambda URL ({api_id}) failed during execution: {e}")942        response.set_json({"Message": "lambda function failed during execution"})943        response.status = "403"944        return response945    response = lambda_result_to_response(result)946    return response947def json_or_eval(body: str):948    try:949        return json.loads(body)950    except JSONDecodeError:951        try:952            return ast.literal_eval(body)953        except Exception as e:...
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.