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

How to use handle_lambda_url_invocation method in localstack

Best Python code snippet using localstack_python

lambda_api.py

Source:lambda_api.py

...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:...

lambda_starter.py

Source:lambda_starter.py

1import logging2from moto.awslambda import models as moto_awslambda_models3from localstack import config4from localstack.services.awslambda.lambda_api import handle_lambda_url_invocation5from localstack.services.edge import ROUTER6from localstack.utils.aws import aws_stack7from localstack.utils.aws.request_context import AWS_REGION_REGEX8from localstack.utils.patch import patch9from localstack.utils.platform import is_linux10from localstack.utils.strings import to_bytes11LOG = logging.getLogger(__name__)12# Key for tracking patch applience13PATCHES_APPLIED = "LAMBDA_PATCHED"14def start_lambda(port=None, asynchronous=False):15    from localstack.services.awslambda import lambda_api, lambda_utils16    from localstack.services.infra import start_local_api17    ROUTER.add(18        "/",19        host=f"<api_id>.lambda-url.<regex('{AWS_REGION_REGEX}'):region>.<regex('.*'):server>",20        endpoint=handle_lambda_url_invocation,21        defaults={"path": ""},22    )23    ROUTER.add(24        "/<path:path>",25        host=f"<api_id>.lambda-url.<regex('{AWS_REGION_REGEX}'):region>.<regex('.*'):server>",26        endpoint=handle_lambda_url_invocation,27    )28    # print a warning if we're not running in Docker but using Docker based LAMBDA_EXECUTOR29    if "docker" in lambda_utils.get_executor_mode() and not config.is_in_docker and not is_linux():30        LOG.warning(31            (32                "!WARNING! - Running outside of Docker with $LAMBDA_EXECUTOR=%s can lead to "33                "problems on your OS. The environment variable $LOCALSTACK_HOSTNAME may not "34                "be properly set in your Lambdas."35            ),36            lambda_utils.get_executor_mode(),37        )38    if (39        config.is_in_docker40        and not config.LAMBDA_REMOTE_DOCKER41        and not config.dirs.functions42        and config.LEGACY_DIRECTORIES43    ):44        LOG.warning(45            "!WARNING! - Looks like you have configured $LAMBDA_REMOTE_DOCKER=0 - "46            "please make sure to configure $HOST_TMP_FOLDER to point to your host's $TMPDIR"47        )48    port = port or config.service_port("lambda")49    return start_local_api(50        "Lambda", port, api="lambda", method=lambda_api.serve, asynchronous=asynchronous51    )52def stop_lambda() -> None:53    from localstack.services.awslambda.lambda_api import cleanup54    """55    Stops / cleans up the Lambda Executor56    """57    # TODO actually stop flask server58    cleanup()59def check_lambda(expect_shutdown=False, print_error=False):60    out = None61    try:62        from localstack.services.infra import PROXY_LISTENERS63        from localstack.utils.aws import aws_stack64        from localstack.utils.common import wait_for_port_open65        # wait for port to be opened66        # TODO get lambda port in a cleaner way67        port = PROXY_LISTENERS.get("lambda")[1]68        wait_for_port_open(port, sleep_time=0.5, retries=20)69        endpoint_url = f"http://127.0.0.1:{port}"70        out = aws_stack.connect_to_service(71            service_name="lambda", endpoint_url=endpoint_url72        ).list_functions()73    except Exception:74        if print_error:75            LOG.exception("Lambda health check failed")76    if expect_shutdown:77        assert out is None78    else:79        assert out and isinstance(out.get("Functions"), list)80@patch(moto_awslambda_models.LambdaBackend.get_function)81def get_function(fn, self, *args, **kwargs):82    result = fn(self, *args, **kwargs)83    if result:84        return result85    client = aws_stack.connect_to_service("lambda")86    lambda_name = aws_stack.lambda_function_name(args[0])87    response = client.get_function(FunctionName=lambda_name)88    spec = response["Configuration"]89    spec["Code"] = {"ZipFile": "ZW1wdHkgc3RyaW5n"}90    region = aws_stack.extract_region_from_arn(spec["FunctionArn"])91    new_function = moto_awslambda_models.LambdaFunction(spec, region)92    return new_function93@patch(moto_awslambda_models.LambdaFunction.invoke)94def invoke(fn, self, *args, **kwargs):95    payload = to_bytes(args[0])96    client = aws_stack.connect_to_service("lambda")...

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.