KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use forward_to_fallback_url method in localstack

Best Python code snippet using localstack_python
lambda_api.py
Source:lambda_api.py
...527        }528    if (always_add_version or version != '$LATEST') and len(result['FunctionArn'].split(':')) <= 7:529        result['FunctionArn'] += ':%s' % (version)530    return result531def forward_to_fallback_url(func_arn, data):532    """ If LAMBDA_FALLBACK_URL is configured, forward the invocation of this non-existing533        Lambda to the configured URL. """534    if not config.LAMBDA_FALLBACK_URL:535        return None536    if config.LAMBDA_FALLBACK_URL.startswith('dynamodb://'):537        table_name = urlparse(config.LAMBDA_FALLBACK_URL.replace('dynamodb://', 'http://')).netloc538        dynamodb = aws_stack.connect_to_service('dynamodb')539        item = {540            'id': {'S': short_uid()},541            'timestamp': {'N': str(now_utc())},542            'payload': {'S': str(data)}543        }544        aws_stack.create_dynamodb_table(table_name, partition_key='id')545        dynamodb.put_item(TableName=table_name, Item=item)546        return ''547    if re.match(r'^https?://.+', config.LAMBDA_FALLBACK_URL):548        response = safe_requests.post(config.LAMBDA_FALLBACK_URL, data)549        return response.content550    raise ClientError('Unexpected value for LAMBDA_FALLBACK_URL: %s' % config.LAMBDA_FALLBACK_URL)551# ------------552# API METHODS553# ------------554@app.before_request555def before_request():556    # fix to enable chunked encoding, as this is used by some Lambda clients557    transfer_encoding = request.headers.get('Transfer-Encoding', '').lower()558    if transfer_encoding == 'chunked':559        request.environ['wsgi.input_terminated'] = True560@app.route('%s/functions' % PATH_ROOT, methods=['POST'])561def create_function():562    """ Create new function563        ---564        operationId: 'createFunction'565        parameters:566            - name: 'request'567              in: body568    """569    arn = 'n/a'570    try:571        data = json.loads(to_str(request.data))572        lambda_name = data['FunctionName']573        event_publisher.fire_event(event_publisher.EVENT_LAMBDA_CREATE_FUNC,574            payload={'n': event_publisher.get_hash(lambda_name)})575        arn = func_arn(lambda_name)576        if arn in arn_to_lambda:577            return error_response('Function already exist: %s' %578                lambda_name, 409, error_type='ResourceConflictException')579        arn_to_lambda[arn] = func_details = LambdaFunction(arn)580        func_details.versions = {'$LATEST': {'RevisionId': str(uuid.uuid4())}}581        func_details.last_modified = isoformat_milliseconds(datetime.utcnow()) + '+0000'582        func_details.description = data.get('Description', '')583        func_details.handler = data['Handler']584        func_details.runtime = data['Runtime']585        func_details.envvars = data.get('Environment', {}).get('Variables', {})586        func_details.tags = data.get('Tags', {})587        func_details.timeout = data.get('Timeout', LAMBDA_DEFAULT_TIMEOUT)588        func_details.role = data['Role']589        func_details.memory_size = data.get('MemorySize')590        func_details.code = data['Code']591        result = set_function_code(func_details.code, lambda_name)592        if isinstance(result, Response):593            del arn_to_lambda[arn]594            return result595        # remove content from code attribute, if present596        func_details.code.pop('ZipFile', None)597        # prepare result598        result.update(format_func_details(func_details))599        if data.get('Publish', False):600            result['Version'] = publish_new_function_version(arn)['Version']601        return jsonify(result or {})602    except Exception as e:603        arn_to_lambda.pop(arn, None)604        if isinstance(e, ClientError):605            return e.get_response()606        return error_response('Unknown error: %s %s' % (e, traceback.format_exc()))607@app.route('%s/functions/<function>' % PATH_ROOT, methods=['GET'])608def get_function(function):609    """ Get details for a single function610        ---611        operationId: 'getFunction'612        parameters:613            - name: 'request'614              in: body615            - name: 'function'616              in: path617    """618    funcs = do_list_functions()619    for func in funcs:620        if func['FunctionName'] == function:621            result = {622                'Configuration': func,623                'Code': {624                    'Location': '%s/code' % request.url625                }626            }627            lambda_details = arn_to_lambda.get(func['FunctionArn'])628            if lambda_details.concurrency is not None:629                result['Concurrency'] = lambda_details.concurrency630            return jsonify(result)631    return error_response(632        'Function not found: %s' % func_arn(function), 404, error_type='ResourceNotFoundException')633@app.route('%s/functions/' % PATH_ROOT, methods=['GET'])634def list_functions():635    """ List functions636        ---637        operationId: 'listFunctions'638        parameters:639            - name: 'request'640              in: body641    """642    funcs = do_list_functions()643    result = {}644    result['Functions'] = funcs645    return jsonify(result)646@app.route('%s/functions/<function>' % PATH_ROOT, methods=['DELETE'])647def delete_function(function):648    """ Delete an existing function649        ---650        operationId: 'deleteFunction'651        parameters:652            - name: 'request'653              in: body654    """655    arn = func_arn(function)656    # Stop/remove any containers that this arn uses.657    LAMBDA_EXECUTOR.cleanup(arn)658    try:659        arn_to_lambda.pop(arn)660    except KeyError:661        return error_response('Function does not exist: %s' % function, 404, error_type='ResourceNotFoundException')662    event_publisher.fire_event(event_publisher.EVENT_LAMBDA_DELETE_FUNC,663        payload={'n': event_publisher.get_hash(function)})664    i = 0665    while i < len(event_source_mappings):666        mapping = event_source_mappings[i]667        if mapping['FunctionArn'] == arn:668            del event_source_mappings[i]669            i -= 1670        i += 1671    result = {}672    return jsonify(result)673@app.route('%s/functions/<function>/code' % PATH_ROOT, methods=['PUT'])674def update_function_code(function):675    """ Update the code of an existing function676        ---677        operationId: 'updateFunctionCode'678        parameters:679            - name: 'request'680              in: body681    """682    data = json.loads(to_str(request.data))683    result = set_function_code(data, function)684    arn = func_arn(function)685    func_details = arn_to_lambda.get(arn)686    result.update(format_func_details(func_details))687    if isinstance(result, Response):688        return result689    return jsonify(result or {})690@app.route('%s/functions/<function>/code' % PATH_ROOT, methods=['GET'])691def get_function_code(function):692    """ Get the code of an existing function693        ---694        operationId: 'getFunctionCode'695        parameters:696    """697    arn = func_arn(function)698    lambda_cwd = arn_to_lambda[arn].cwd699    tmp_file = '%s/%s' % (lambda_cwd, LAMBDA_ZIP_FILE_NAME)700    return Response(load_file(tmp_file, mode='rb'),701            mimetype='application/zip',702            headers={'Content-Disposition': 'attachment; filename=lambda_archive.zip'})703@app.route('%s/functions/<function>/configuration' % PATH_ROOT, methods=['GET'])704def get_function_configuration(function):705    """ Get the configuration of an existing function706        ---707        operationId: 'getFunctionConfiguration'708        parameters:709    """710    arn = func_arn(function)711    lambda_details = arn_to_lambda.get(arn)712    if not lambda_details:713        return error_response('Function not found: %s' % arn, 404, error_type='ResourceNotFoundException')714    result = format_func_details(lambda_details)715    return jsonify(result)716@app.route('%s/functions/<function>/configuration' % PATH_ROOT, methods=['PUT'])717def update_function_configuration(function):718    """ Update the configuration of an existing function719        ---720        operationId: 'updateFunctionConfiguration'721        parameters:722            - name: 'request'723              in: body724    """725    data = json.loads(to_str(request.data))726    arn = func_arn(function)727    # Stop/remove any containers that this arn uses.728    LAMBDA_EXECUTOR.cleanup(arn)729    lambda_details = arn_to_lambda[arn]730    if data.get('Handler'):731        lambda_details.handler = data['Handler']732    if data.get('Runtime'):733        lambda_details.runtime = data['Runtime']734    env_vars = data.get('Environment', {}).get('Variables')735    if env_vars is not None:736        lambda_details.envvars = env_vars737    if data.get('Timeout'):738        lambda_details.timeout = data['Timeout']739    result = {}740    return jsonify(result)741@app.route('%s/functions/<function>/invocations' % PATH_ROOT, methods=['POST'])742def invoke_function(function):743    """ Invoke an existing function744        ---745        operationId: 'invokeFunction'746        parameters:747            - name: 'request'748              in: body749    """750    # function here can either be an arn or a function name751    arn = func_arn(function)752    # arn can also contain a qualifier, extract it from there if so753    m = re.match('(arn:aws:lambda:.*:.*:function:[a-zA-Z0-9-_]+)(:.*)?', arn)754    if m and m.group(2):755        qualifier = m.group(2)[1:]756        arn = m.group(1)757    else:758        qualifier = request.args.get('Qualifier')759    data = request.get_data()760    if data:761        data = to_str(data)762        try:763            data = json.loads(data)764        except Exception:765            try:766                # try to read chunked content767                data = json.loads(parse_chunked_data(data))768            except Exception:769                return error_response('The payload is not JSON: %s' % data, 415,770                                      error_type='UnsupportedMediaTypeException')771    # Default invocation type is RequestResponse772    invocation_type = request.environ.get('HTTP_X_AMZ_INVOCATION_TYPE', 'RequestResponse')773    def _create_response(result, status_code=200):774        """ Create the final response for the given invocation result """775        if isinstance(result, Response):776            return result777        details = {778            'StatusCode': status_code,779            'Payload': result,780            'Headers': {}781        }782        if isinstance(result, dict):783            for key in ('StatusCode', 'Payload', 'FunctionError'):784                if result.get(key):785                    details[key] = result[key]786        # Try to parse parse payload as JSON787        payload = details['Payload']788        if payload and isinstance(payload, (str, bytes)) and payload[0] in ('[', '{', '"'):789            try:790                details['Payload'] = json.loads(details['Payload'])791            except Exception:792                pass793        # Set error headers794        if details.get('FunctionError'):795            details['Headers']['X-Amz-Function-Error'] = str(details['FunctionError'])796        # Construct response object797        response_obj = details['Payload']798        if isinstance(response_obj, (dict, list, bool)) or is_number(response_obj):799            # Assume this is a JSON response800            response_obj = jsonify(response_obj)801        else:802            response_obj = str(response_obj)803            details['Headers']['Content-Type'] = 'text/plain'804        return response_obj, details['StatusCode'], details['Headers']805    # check if this lambda function exists806    not_found = None807    if arn not in arn_to_lambda:808        not_found = error_response('Function does not exist: %s' % arn, 404, error_type='ResourceNotFoundException')809    if qualifier and not arn_to_lambda.get(arn).qualifier_exists(qualifier):810        not_found = error_response('Function does not exist: {0}:{1}'.format(arn, qualifier), 404,811                              error_type='ResourceNotFoundException')812    if not_found:813        forward_result = forward_to_fallback_url(func_arn, data)814        if forward_result is not None:815            return _create_response(forward_result)816        return not_found817    if invocation_type == 'RequestResponse':818        result = run_lambda(asynchronous=False, func_arn=arn, event=data, context={}, version=qualifier)819        return _create_response(result)820    elif invocation_type == 'Event':821        run_lambda(asynchronous=True, func_arn=arn, event=data, context={}, version=qualifier)822        return _create_response('', status_code=202)823    elif invocation_type == 'DryRun':824        # Assume the dry run always passes.825        return _create_response('', status_code=204)826    return error_response('Invocation type not one of: RequestResponse, Event or DryRun',827                          code=400, error_type='InvalidParameterValueException')...
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.