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

Best Python code snippet using localstack_python
lambda_api.py
Source:lambda_api.py
...213    return result214def get_function_version(arn, version):215    func = arn_to_lambda.get(arn)216    return format_func_details(func, version=version, always_add_version=True)217def publish_new_function_version(arn):218    versions = arn_to_lambda.get(arn).versions219    if len(versions) == 1:220        last_version = 0221    else:222        last_version = max([int(key) for key in versions.keys() if key != '$LATEST'])223    versions[str(last_version + 1)] = {'CodeSize': versions.get('$LATEST').get('CodeSize'),224                                    'Function': versions.get('$LATEST').get('Function')}225    return get_function_version(arn, str(last_version + 1))226def do_list_versions(arn):227    return sorted([get_function_version(arn, version) for version in228                   arn_to_lambda.get(arn).versions.keys()], key=lambda k: str(k.get('Version')))229def do_update_alias(arn, alias, version, description=None):230    new_alias = {231        'AliasArn': arn + ':' + alias,232        'FunctionVersion': version,233        'Name': alias,234        'Description': description or ''235    }236    arn_to_lambda.get(arn).aliases[alias] = new_alias237    return new_alias238@cloudwatched('lambda')239def run_lambda(event, context, func_arn, version=None, suppress_output=False, asynchronous=False):240    if suppress_output:241        stdout_ = sys.stdout242        stderr_ = sys.stderr243        stream = StringIO()244        sys.stdout = stream245        sys.stderr = stream246    try:247        func_details = arn_to_lambda.get(func_arn)248        if not context:249            context = LambdaContext(func_details, version)250        result, log_output = LAMBDA_EXECUTOR.execute(func_arn, func_details,251            event, context=context, version=version, asynchronous=asynchronous)252    except Exception as e:253        return error_response('Error executing Lambda function: %s %s' % (e, traceback.format_exc()))254    finally:255        if suppress_output:256            sys.stdout = stdout_257            sys.stderr = stderr_258    return result259def exec_lambda_code(script, handler_function='handler', lambda_cwd=None, lambda_env=None):260    if lambda_cwd or lambda_env:261        exec_mutex.acquire()262        if lambda_cwd:263            previous_cwd = os.getcwd()264            os.chdir(lambda_cwd)265            sys.path = [lambda_cwd] + sys.path266        if lambda_env:267            previous_env = dict(os.environ)268            os.environ.update(lambda_env)269    # generate lambda file name270    lambda_id = 'l_%s' % short_uid()271    lambda_file = LAMBDA_SCRIPT_PATTERN.replace('*', lambda_id)272    save_file(lambda_file, script)273    # delete temporary .py and .pyc files on exit274    TMP_FILES.append(lambda_file)275    TMP_FILES.append('%sc' % lambda_file)276    try:277        handler_module = imp.load_source(lambda_id, lambda_file)278        module_vars = handler_module.__dict__279    except Exception as e:280        LOG.error('Unable to exec: %s %s' % (script, traceback.format_exc()))281        raise e282    finally:283        if lambda_cwd or lambda_env:284            if lambda_cwd:285                os.chdir(previous_cwd)286                sys.path.pop(0)287            if lambda_env:288                os.environ = previous_env289            exec_mutex.release()290    return module_vars[handler_function]291def get_handler_file_from_name(handler_name, runtime=LAMBDA_DEFAULT_RUNTIME):292    # TODO: support Java Lambdas in the future293    delimiter = '.'294    if runtime.startswith(LAMBDA_RUNTIME_NODEJS):295        file_ext = '.js'296    elif runtime.startswith(LAMBDA_RUNTIME_GOLANG):297        file_ext = ''298    elif runtime.startswith(LAMBDA_RUNTIME_DOTNETCORE2):299        file_ext = '.dll'300        delimiter = ':'301    else:302        file_ext = '.py'303    return '%s%s' % (handler_name.split(delimiter)[0], file_ext)304def get_handler_function_from_name(handler_name, runtime=LAMBDA_DEFAULT_RUNTIME):305    # TODO: support Java Lambdas in the future306    if runtime.startswith(LAMBDA_RUNTIME_DOTNETCORE2):307        return handler_name.split(':')[-1]308    else:309        return handler_name.split('.')[-1]310def error_response(msg, code=500, error_type='InternalFailure'):311    LOG.warning(msg)312    return aws_responses.flask_error_response(msg, code=code, error_type=error_type)313def get_zip_bytes(function_code):314    """Returns the ZIP file contents from a FunctionCode dict.315    :type function_code: dict316    :param function_code: https://docs.aws.amazon.com/lambda/latest/dg/API_FunctionCode.html317    :returns: bytes of the Zip file.318    """319    if 'S3Bucket' in function_code:320        s3_client = aws_stack.connect_to_service('s3')321        bytes_io = BytesIO()322        try:323            s3_client.download_fileobj(function_code['S3Bucket'], function_code['S3Key'], bytes_io)324            zip_file_content = bytes_io.getvalue()325        except Exception as e:326            return error_response('Unable to fetch Lambda archive from S3: %s' % e, 404)327    elif 'ZipFile' in function_code:328        zip_file_content = function_code['ZipFile']329        zip_file_content = base64.b64decode(zip_file_content)330    else:331        return error_response('No valid Lambda archive specified.', 400)332    return zip_file_content333def get_java_handler(zip_file_content, handler, main_file):334    """Creates a Java handler from an uploaded ZIP or JAR.335    :type zip_file_content: bytes336    :param zip_file_content: ZIP file bytes.337    :type handler: str338    :param handler: The lambda handler path.339    :type main_file: str340    :param main_file: Filepath to the uploaded ZIP or JAR file.341    :returns: function or flask.Response342    """343    if not is_jar_archive(zip_file_content):344        with zipfile.ZipFile(BytesIO(zip_file_content)) as zip_ref:345            jar_entries = [e for e in zip_ref.infolist() if e.filename.endswith('.jar')]346            if len(jar_entries) != 1:347                raise Exception('Expected exactly one *.jar entry in zip file, found %s' % len(jar_entries))348            zip_file_content = zip_ref.read(jar_entries[0].filename)349            LOG.info('Found jar file %s with %s bytes in Lambda zip archive' %350                     (jar_entries[0].filename, len(zip_file_content)))351            main_file = new_tmp_file()352            save_file(main_file, zip_file_content)353    if is_jar_archive(zip_file_content):354        def execute(event, context):355            result, log_output = lambda_executors.EXECUTOR_LOCAL.execute_java_lambda(356                event, context, handler=handler, main_file=main_file)357            return result358        return execute359    return error_response(360        'Unable to extract Java Lambda handler - file is not a valid zip/jar files', 400, error_type='ValidationError')361def set_function_code(code, lambda_name):362    def generic_handler(event, context):363        raise Exception(('Unable to find executor for Lambda function "%s". ' +364            'Note that Node.js and .NET Core Lambdas currently require LAMBDA_EXECUTOR=docker') % lambda_name)365    lambda_cwd = None366    arn = func_arn(lambda_name)367    runtime = arn_to_lambda[arn].runtime368    handler_name = arn_to_lambda.get(arn).handler369    lambda_environment = arn_to_lambda.get(arn).envvars370    if not handler_name:371        handler_name = LAMBDA_DEFAULT_HANDLER372    # Stop/remove any containers that this arn uses.373    LAMBDA_EXECUTOR.cleanup(arn)374    # Save the zip file to a temporary file that the lambda executors can reference.375    zip_file_content = get_zip_bytes(code)376    if isinstance(zip_file_content, Response):377        return zip_file_content378    tmp_dir = '%s/zipfile.%s' % (config.TMP_FOLDER, short_uid())379    mkdir(tmp_dir)380    tmp_file = '%s/%s' % (tmp_dir, LAMBDA_ZIP_FILE_NAME)381    save_file(tmp_file, zip_file_content)382    TMP_FILES.append(tmp_dir)383    lambda_cwd = tmp_dir384    # Set the appropriate lambda handler.385    lambda_handler = generic_handler386    if runtime == LAMBDA_RUNTIME_JAVA8:387        # The Lambda executors for Docker subclass LambdaExecutorContainers,388        # which runs Lambda in Docker by passing all *.jar files in the function389        # working directory as part of the classpath. Because of this, we need to390        # save the zip_file_content as a .jar here.391        if is_jar_archive(zip_file_content):392            jar_tmp_file = '{working_dir}/{file_name}'.format(393                working_dir=tmp_dir, file_name=LAMBDA_JAR_FILE_NAME)394            save_file(jar_tmp_file, zip_file_content)395        lambda_handler = get_java_handler(zip_file_content, handler_name, tmp_file)396        if isinstance(lambda_handler, Response):397            return lambda_handler398    else:399        handler_file = get_handler_file_from_name(handler_name, runtime=runtime)400        handler_function = get_handler_function_from_name(handler_name, runtime=runtime)401        # Lambda code must be uploaded in the Zip format.402        if not is_zip_file(zip_file_content):403            raise Exception(404                'Uploaded Lambda code for runtime ({}) is not in Zip format'.format(runtime))405        unzip(tmp_file, tmp_dir)406        main_file = '%s/%s' % (tmp_dir, handler_file)407        if os.path.isfile(main_file):408            # make sure the file is actually readable, then read contents409            ensure_readable(main_file)410            with open(main_file, 'rb') as file_obj:411                zip_file_content = file_obj.read()412        else:413            file_list = run('ls -la %s' % tmp_dir)414            LOG.debug('Lambda archive content:\n%s' % file_list)415            return error_response(416                'Unable to find handler script in Lambda archive.', 400,417                error_type='ValidationError')418        if runtime.startswith('python') and not use_docker():419            try:420                lambda_handler = exec_lambda_code(421                    zip_file_content,422                    handler_function=handler_function,423                    lambda_cwd=lambda_cwd,424                    lambda_env=lambda_environment)425            except Exception as e:426                raise Exception('Unable to get handler function from lambda code.', e)427    add_function_mapping(lambda_name, lambda_handler, lambda_cwd)428    return {'FunctionName': lambda_name}429def do_list_functions():430    funcs = []431    for f_arn, func in iteritems(arn_to_lambda):432        func_name = f_arn.split(':function:')[-1]433        arn = func_arn(func_name)434        func_details = arn_to_lambda.get(arn)435        funcs.append(format_func_details(func_details))436    return funcs437def format_func_details(func_details, version=None, always_add_version=False):438    version = version or '$LATEST'439    result = {440        'Version': version,441        'FunctionArn': func_details.arn(),442        'FunctionName': func_details.name(),443        'CodeSize': func_details.get_version(version).get('CodeSize'),444        'Handler': func_details.handler,445        'Runtime': func_details.runtime,446        'Timeout': func_details.timeout,447        'Environment': func_details.envvars,448        # 'Description': ''449        # 'MemorySize': 192,450    }451    if (always_add_version or version != '$LATEST') and len(result['FunctionArn'].split(':')) <= 7:452        result['FunctionArn'] += ':%s' % (version)453    return result454# ------------455# API METHODS456# ------------457@app.route('%s/functions' % PATH_ROOT, methods=['POST'])458def create_function():459    """ Create new function460        ---461        operationId: 'createFunction'462        parameters:463            - name: 'request'464              in: body465    """466    arn = 'n/a'467    try:468        data = json.loads(to_str(request.data))469        lambda_name = data['FunctionName']470        event_publisher.fire_event(event_publisher.EVENT_LAMBDA_CREATE_FUNC,471            payload={'n': event_publisher.get_hash(lambda_name)})472        arn = func_arn(lambda_name)473        if arn in arn_to_lambda:474            return error_response('Function already exist: %s' %475                lambda_name, 409, error_type='ResourceConflictException')476        arn_to_lambda[arn] = func_details = LambdaFunction(arn)477        func_details.versions = {'$LATEST': {'CodeSize': 50}}478        func_details.handler = data['Handler']479        func_details.runtime = data['Runtime']480        func_details.envvars = data.get('Environment', {}).get('Variables', {})481        func_details.timeout = data.get('Timeout')482        result = set_function_code(data['Code'], lambda_name)483        if isinstance(result, Response):484            del arn_to_lambda[arn]485            return result486        result.update({487            'DeadLetterConfig': data.get('DeadLetterConfig'),488            'Description': data.get('Description'),489            'Environment': {'Error': {}, 'Variables': func_details.envvars},490            'FunctionArn': arn,491            'FunctionName': lambda_name,492            'Handler': func_details.handler,493            'MemorySize': data.get('MemorySize'),494            'Role': data.get('Role'),495            'Runtime': func_details.runtime,496            'Timeout': data.get('Timeout'),497            'TracingConfig': {},498            'VpcConfig': {'SecurityGroupIds': [None], 'SubnetIds': [None], 'VpcId': None}499        })500        if data.get('Publish', False):501            result['Version'] = publish_new_function_version(arn)['Version']502        return jsonify(result or {})503    except Exception as e:504        del arn_to_lambda[arn]505        return error_response('Unknown error: %s %s' % (e, traceback.format_exc()))506@app.route('%s/functions/<function>' % PATH_ROOT, methods=['GET'])507def get_function(function):508    """ Get details for a single function509        ---510        operationId: 'getFunction'511        parameters:512            - name: 'request'513              in: body514            - name: 'function'515              in: path516    """517    funcs = do_list_functions()518    for func in funcs:519        if func['FunctionName'] == function:520            result = {521                'Configuration': func,522                'Code': {523                    'Location': '%s/code' % request.url524                }525            }526            lambda_details = arn_to_lambda.get(func['FunctionArn'])527            if lambda_details.concurrency is not None:528                result['Concurrency'] = lambda_details.concurrency529            return jsonify(result)530    return error_response(531        'Function not found: %s' % func_arn(function), 404, error_type='ResourceNotFoundException')532@app.route('%s/functions/' % PATH_ROOT, methods=['GET'])533def list_functions():534    """ List functions535        ---536        operationId: 'listFunctions'537        parameters:538            - name: 'request'539              in: body540    """541    funcs = do_list_functions()542    result = {}543    result['Functions'] = funcs544    return jsonify(result)545@app.route('%s/functions/<function>' % PATH_ROOT, methods=['DELETE'])546def delete_function(function):547    """ Delete an existing function548        ---549        operationId: 'deleteFunction'550        parameters:551            - name: 'request'552              in: body553    """554    arn = func_arn(function)555    # Stop/remove any containers that this arn uses.556    LAMBDA_EXECUTOR.cleanup(arn)557    try:558        arn_to_lambda.pop(arn)559    except KeyError:560        return error_response('Function does not exist: %s' % function, 404, error_type='ResourceNotFoundException')561    event_publisher.fire_event(event_publisher.EVENT_LAMBDA_DELETE_FUNC,562        payload={'n': event_publisher.get_hash(function)})563    i = 0564    while i < len(event_source_mappings):565        mapping = event_source_mappings[i]566        if mapping['FunctionArn'] == arn:567            del event_source_mappings[i]568            i -= 1569        i += 1570    result = {}571    return jsonify(result)572@app.route('%s/functions/<function>/code' % PATH_ROOT, methods=['PUT'])573def update_function_code(function):574    """ Update the code of an existing function575        ---576        operationId: 'updateFunctionCode'577        parameters:578            - name: 'request'579              in: body580    """581    data = json.loads(to_str(request.data))582    result = set_function_code(data, function)583    return jsonify(result or {})584@app.route('%s/functions/<function>/code' % PATH_ROOT, methods=['GET'])585def get_function_code(function):586    """ Get the code of an existing function587        ---588        operationId: 'getFunctionCode'589        parameters:590    """591    arn = func_arn(function)592    lambda_cwd = arn_to_lambda[arn].cwd593    tmp_file = '%s/%s' % (lambda_cwd, LAMBDA_ZIP_FILE_NAME)594    return Response(load_file(tmp_file, mode='rb'),595            mimetype='application/zip',596            headers={'Content-Disposition': 'attachment; filename=lambda_archive.zip'})597@app.route('%s/functions/<function>/configuration' % PATH_ROOT, methods=['GET'])598def get_function_configuration(function):599    """ Get the configuration of an existing function600        ---601        operationId: 'getFunctionConfiguration'602        parameters:603    """604    arn = func_arn(function)605    lambda_details = arn_to_lambda.get(arn)606    if not lambda_details:607        return error_response('Function not found: %s' % arn, 404, error_type='ResourceNotFoundException')608    result = format_func_details(lambda_details)609    return jsonify(result)610@app.route('%s/functions/<function>/configuration' % PATH_ROOT, methods=['PUT'])611def update_function_configuration(function):612    """ Update the configuration of an existing function613        ---614        operationId: 'updateFunctionConfiguration'615        parameters:616            - name: 'request'617              in: body618    """619    data = json.loads(to_str(request.data))620    arn = func_arn(function)621    # Stop/remove any containers that this arn uses.622    LAMBDA_EXECUTOR.cleanup(arn)623    lambda_details = arn_to_lambda[arn]624    if data.get('Handler'):625        lambda_details.handler = data['Handler']626    if data.get('Runtime'):627        lambda_details.runtime = data['Runtime']628    if data.get('Environment'):629        lambda_details.envvars = data.get('Environment', {}).get('Variables', {})630    if data.get('Timeout'):631        lambda_details.timeout = data['Timeout']632    result = {}633    return jsonify(result)634@app.route('%s/functions/<function>/invocations' % PATH_ROOT, methods=['POST'])635def invoke_function(function):636    """ Invoke an existing function637        ---638        operationId: 'invokeFunction'639        parameters:640            - name: 'request'641              in: body642    """643    # function here can either be an arn or a function name644    arn = func_arn(function)645    # arn can also contain a qualifier, extract it from there if so646    m = re.match('(arn:aws:lambda:.*:.*:function:[a-zA-Z0-9-_]+)(:.*)?', arn)647    if m and m.group(2):648        qualifier = m.group(2)[1:]649        arn = m.group(1)650    else:651        qualifier = request.args.get('Qualifier')652    if arn not in arn_to_lambda:653        return error_response('Function does not exist: %s' % arn, 404, error_type='ResourceNotFoundException')654    if qualifier and not arn_to_lambda.get(arn).qualifier_exists(qualifier):655        return error_response('Function does not exist: {0}:{1}'.format(arn, qualifier), 404,656                              error_type='ResourceNotFoundException')657    data = None658    if request.data:659        try:660            data = json.loads(to_str(request.data))661        except Exception:662            return error_response('The payload is not JSON', 415, error_type='UnsupportedMediaTypeException')663    # Default invocation type is RequestResponse664    invocation_type = request.environ.get('HTTP_X_AMZ_INVOCATION_TYPE', 'RequestResponse')665    if invocation_type == 'RequestResponse':666        result = run_lambda(asynchronous=False, func_arn=arn, event=data, context={}, version=qualifier)667        if isinstance(result, dict):668            return jsonify(result)669        if result:670            return result671        return make_response('', 200)672    elif invocation_type == 'Event':673        run_lambda(asynchronous=True, func_arn=arn, event=data, context={}, version=qualifier)674        return make_response('', 202)675    elif invocation_type == 'DryRun':676        # Assume the dry run always passes.677        return make_response('', 204)678    else:679        return error_response('Invocation type not one of: RequestResponse, Event or DryRun',680                              code=400,681                              error_type='InvalidParameterValueException')682@app.route('%s/event-source-mappings/' % PATH_ROOT, methods=['GET'])683def list_event_source_mappings():684    """ List event source mappings685        ---686        operationId: 'listEventSourceMappings'687    """688    event_source_arn = request.args.get('EventSourceArn')689    function_name = request.args.get('FunctionName')690    mappings = event_source_mappings691    if event_source_arn:692        mappings = [m for m in mappings if event_source_arn == m.get('EventSourceArn')]693    if function_name:694        function_arn = func_arn(function_name)695        mappings = [m for m in mappings if function_arn == m.get('FunctionArn')]696    response = {697        'EventSourceMappings': mappings698    }699    return jsonify(response)700@app.route('%s/event-source-mappings/<mapping_uuid>' % PATH_ROOT, methods=['GET'])701def get_event_source_mapping(mapping_uuid):702    """ Get an existing event source mapping703        ---704        operationId: 'getEventSourceMapping'705        parameters:706            - name: 'request'707              in: body708    """709    mappings = event_source_mappings710    mappings = [m for m in mappings if mapping_uuid == m.get('UUID')]711    if len(mappings) == 0:712        return error_response('The resource you requested does not exist.', 404, error_type='ResourceNotFoundException')713    return jsonify(mappings[0])714@app.route('%s/event-source-mappings/' % PATH_ROOT, methods=['POST'])715def create_event_source_mapping():716    """ Create new event source mapping717        ---718        operationId: 'createEventSourceMapping'719        parameters:720            - name: 'request'721              in: body722    """723    data = json.loads(to_str(request.data))724    mapping = add_event_source(data['FunctionName'], data['EventSourceArn'])725    return jsonify(mapping)726@app.route('%s/event-source-mappings/<mapping_uuid>' % PATH_ROOT, methods=['PUT'])727def update_event_source_mapping(mapping_uuid):728    """ Update an existing event source mapping729        ---730        operationId: 'updateEventSourceMapping'731        parameters:732            - name: 'request'733              in: body734    """735    data = json.loads(request.data)736    if not mapping_uuid:737        return jsonify({})738    function_name = data.get('FunctionName') or ''739    enabled = data.get('Enabled') or True740    batch_size = data.get('BatchSize') or 100741    mapping = update_event_source(mapping_uuid, function_name, enabled, batch_size)742    return jsonify(mapping)743@app.route('%s/event-source-mappings/<mapping_uuid>' % PATH_ROOT, methods=['DELETE'])744def delete_event_source_mapping(mapping_uuid):745    """ Delete an event source mapping746        ---747        operationId: 'deleteEventSourceMapping'748    """749    if not mapping_uuid:750        return jsonify({})751    mapping = delete_event_source(mapping_uuid)752    return jsonify(mapping)753@app.route('%s/functions/<function>/versions' % PATH_ROOT, methods=['POST'])754def publish_version(function):755    arn = func_arn(function)756    if arn not in arn_to_lambda:757        return error_response('Function not found: %s' % arn, 404, error_type='ResourceNotFoundException')758    return jsonify(publish_new_function_version(arn))759@app.route('%s/functions/<function>/versions' % PATH_ROOT, methods=['GET'])760def list_versions(function):761    arn = func_arn(function)762    if arn not in arn_to_lambda:763        return error_response('Function not found: %s' % arn, 404, error_type='ResourceNotFoundException')764    return jsonify({'Versions': do_list_versions(arn)})765@app.route('%s/functions/<function>/aliases' % PATH_ROOT, methods=['POST'])766def create_alias(function):767    arn = func_arn(function)768    if arn not in arn_to_lambda:769        return error_response('Function not found: %s' % arn, 404, error_type='ResourceNotFoundException')770    data = json.loads(request.data)771    alias = data.get('Name')772    if alias in arn_to_lambda.get(arn).aliases:...
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.