KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use func_arn method in localstack

Best Python code snippet using localstack_python

awslambda.py

Source:awslambda.py

...27    dotnet31 = "dotnetcore3.1"28    dotnet21 = "dotnetcore2.1"29    custom_amzlinux2 = "provided.al2"30    custom_amzlinux1 = "provided"31def _to_func_arn(func: Union[str, awslambda.Function]):32    if isinstance(func, str):33        function_name = Sub(34            string="arn:aws:lambda:${aws_region}:${aws_account_id}:function:${func_name}",35            data=dict(36                aws_account_id=AWS_ACCOUNT_ID,37                aws_region=AWS_REGION,38                func_name=func,39            ),40        )41    else:42        function_name = func.rv_Arn43    return function_name44def create_permission_for_s3_event(45    logic_id: str,46    func: Union[str, awslambda.Function],47    bucket: Union[str, s3.Bucket],48) -> awslambda.Permission:49    """50    .. note::51        The s3 bucket has to be in the same region of lambda function.52    .. note::53        The source arn has to be a bucket without any prefix, otherwise it54        won't pass the validation.55    Ref:56    - https://docs.aws.amazon.com/lambda/latest/dg/with-s3-example.html57    """58    func_arn = _to_func_arn(func)59    if isinstance(bucket, str):60        if bucket.startswith("arn:aws:s3:::"):61            source_arn = bucket62        else:63            source_arn = f"arn:aws:s3:::{bucket}"64    else:65        source_arn = Sub(66            string="arn:aws:s3:::${bucket}",67            data=dict(bucket=bucket.ref())68        )69    return awslambda.Permission(70        logic_id,71        rp_FunctionName=func_arn,72        rp_Action="lambda:InvokeFunction",73        rp_Principal="s3.amazonaws.com",74        p_SourceArn=source_arn,75        p_SourceAccount=AWS_ACCOUNT_ID,76    )77def create_permission_for_kinesis_data_stream(78    logic_id: str,79    func: Union[str, awslambda.Function],80    data_stream: Union[str, kinesis.Stream]81):82    """83    Ref:84    - https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html85    """86    func_arn = _to_func_arn(func)87    if isinstance(data_stream, str):88        if data_stream.startswith("arn:aws:kinesis:"):89            source_arn = data_stream90        else:91            source_arn = Sub(92                string="arn:aws:kinesis:${aws_region}:${aws_account_id}:stream/${data_stream}",93                data=dict(94                    aws_account_id=AWS_ACCOUNT_ID,95                    aws_region=AWS_REGION,96                    data_stream=data_stream,97                )98            )99    else:100        source_arn = data_stream.rv_Arn101    return awslambda.Permission(102        logic_id,103        rp_FunctionName=func_arn,104        rp_Action="lambda:InvokeFunction",105        rp_Principal="kinesis.amazonaws.com",106        p_SourceArn=source_arn,107        p_SourceAccount=AWS_ACCOUNT_ID,108    )109def create_permission_for_firehose_delivery_stream(110    logic_id: str,111    func: Union[str, awslambda.Function],112    delivery_stream: Union[str, kinesisfirehose.DeliveryStream]113):114    """115    Ref:116    - https://docs.aws.amazon.com/lambda/latest/dg/services-kinesisfirehose.html117    """118    func_arn = _to_func_arn(func)119    if isinstance(delivery_stream, str):120        if delivery_stream.startswith("arn:aws:firehose:"):121            source_arn = delivery_stream122        else:123            source_arn = Sub(124                string="arn:aws:firehose:${aws_region}:${aws_account_id}:deliverystream/${delivery_stream}",125                data=dict(126                    aws_account_id=AWS_ACCOUNT_ID,127                    aws_region=AWS_REGION,128                    delivery_stream=delivery_stream,129                )130            )131    else:132        source_arn = delivery_stream.rv_Arn133    return awslambda.Permission(134        logic_id,135        rp_FunctionName=func_arn,136        rp_Action="lambda:InvokeFunction",137        rp_Principal="firehose.amazonaws.com",138        p_SourceArn=source_arn,139        p_SourceAccount=AWS_ACCOUNT_ID,140    )141def create_permission_for_sns(142    logic_id: str,143    func: Union[str, awslambda.Function],144    topic: Union[str, sns.Topic]145):146    """147    Ref:148    - https://docs.aws.amazon.com/lambda/latest/dg/with-sns.html149    """150    func_arn = _to_func_arn(func)151    if isinstance(topic, str):152        if topic.startswith("arn:aws:sns:"):153            source_arn = topic154        else:155            source_arn = Sub(156                string="arn:aws:sns:${aws_region}:${aws_account_id}:${sns_topic}",157                data=dict(158                    aws_account_id=AWS_ACCOUNT_ID,159                    aws_region=AWS_REGION,160                    sns_topic=topic,161                )162            )163    else:164        source_arn = topic.ref()165    return awslambda.Permission(166        logic_id,167        rp_FunctionName=func_arn,168        rp_Action="lambda:InvokeFunction",169        rp_Principal="sns.amazonaws.com",170        p_SourceArn=source_arn,171        p_SourceAccount=AWS_ACCOUNT_ID,172    )173def create_permission_for_sqs(174    logic_id: str,175    func: Union[str, awslambda.Function],176    queue: Union[str, sqs.Queue]177):178    """179    Ref:180    - https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html181    """182    func_arn = _to_func_arn(func)183    if isinstance(queue, str):184        if queue.startswith("arn:aws:sqs:"):185            source_arn = queue186        else:187            source_arn = Sub(188                string="arn:aws:sqs:${aws_region}:${aws_account_id}:${queue}",189                data=dict(190                    aws_account_id=AWS_ACCOUNT_ID,191                    aws_region=AWS_REGION,192                    queue=queue,193                )194            )195    else:196        source_arn = queue.ref()197    return awslambda.Permission(198        logic_id,199        rp_FunctionName=func_arn,200        rp_Action="lambda:InvokeFunction",201        rp_Principal="sqs.amazonaws.com",202        p_SourceArn=source_arn,203        p_SourceAccount=AWS_ACCOUNT_ID,204    )205def create_permission_for_dynamodb(206    logic_id: str,207    func: Union[str, awslambda.Function],208    table: Union[str, dynamodb.Table]209):210    """211    Ref:212    - https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html213    """214    func_arn = _to_func_arn(func)215    if isinstance(table, str):216        if table.startswith("arn:aws:dynamodb:"):217            source_arn = table218        else:219            source_arn = Sub(220                string="arn:aws:dynamodb:${aws_region}:${aws_account_id}:table/${table}",221                data=dict(222                    aws_account_id=AWS_ACCOUNT_ID,223                    aws_region=AWS_REGION,224                    table=table,225                )226            )227    else:228        source_arn = table.rv_Arn229    return awslambda.Permission(230        logic_id,231        rp_FunctionName=func_arn,232        rp_Action="lambda:InvokeFunction",233        rp_Principal="dynamodb.amazonaws.com",234        p_SourceArn=source_arn,235        p_SourceAccount=AWS_ACCOUNT_ID,236    )237def create_permission_for_code_commit(238    logic_id: str,239    func: Union[str, awslambda.Function],240    repo: Union[str, codecommit.Repository]241):242    """243    Ref:244    - https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html245    """246    func_arn = _to_func_arn(func)247    if isinstance(repo, str):248        if repo.startswith("arn:aws:codecommit:"):249            source_arn = repo250        else:251            source_arn = Sub(252                string="arn:aws:codecommit:${aws_region}:${aws_account_id}:${repo}",253                data=dict(254                    aws_account_id=AWS_ACCOUNT_ID,255                    aws_region=AWS_REGION,256                    repo=repo,257                )258            )259    else:260        source_arn = repo.rv_Arn261    return awslambda.Permission(262        logic_id,263        rp_FunctionName=func_arn,264        rp_Action="lambda:InvokeFunction",265        rp_Principal="codecommit.amazonaws.com",266        p_SourceArn=source_arn,267        p_SourceAccount=AWS_ACCOUNT_ID,268    )269def create_permission_for_cloudwatch_event(270    logic_id: str,271    func: Union[str, awslambda.Function],272    rule: Union[str, events.Rule]273):274    """275    Ref:276    - https://docs.aws.amazon.com/lambda/latest/dg/services-cloudwatchevents.html277    """278    func_arn = _to_func_arn(func)279    if isinstance(rule, str):280        if rule.startswith("arn:aws:events:"):281            source_arn = rule282        else:283            source_arn = Sub(284                string="arn:aws:events:${aws_region}:${aws_account_id}:rule/${rule}",285                data=dict(286                    aws_account_id=AWS_ACCOUNT_ID,287                    aws_region=AWS_REGION,288                    rule=rule,289                )290            )291    else:292        source_arn = rule.rv_Arn293    return awslambda.Permission(294        logic_id,295        rp_FunctionName=func_arn,296        rp_Action="lambda:InvokeFunction",297        rp_Principal="events.amazonaws.com",298        p_SourceArn=source_arn,299        p_SourceAccount=AWS_ACCOUNT_ID,300    )301def create_permission_for_msk(302    logic_id: str,303    func: Union[str, awslambda.Function],304    cluster: Union[str, msk.Cluster],305    topic: str,306    cluster_uuid: str = None,307):308    """309    Ref:310    - https://docs.aws.amazon.com/lambda/latest/dg/with-msk.html311    """312    func_arn = _to_func_arn(func)313    if isinstance(cluster, str):314        if cluster.startswith("arn:aws:kafka:"):315            source_arn = cluster316        else:317            source_arn = Sub(318                string="arn:aws:kafka:${aws_region}:${aws_account_id}:topic/${cluster_name}/${cluster_uuid}/${topic}",319                data=dict(320                    aws_account_id=AWS_ACCOUNT_ID,321                    aws_region=AWS_REGION,322                    cluster_name=cluster,323                    cluster_uuid=cluster_uuid,324                    topic=topic,325                )326            )...

custom_resource.py

Source:custom_resource.py

1import logging2import os3import boto34region = os.environ['AWS_REGION']5Lambda_ARN = os.environ['LE_ARN']6CF_DIST_ID = os.environ['CF_DIST_ID']7CF_BEHAVIOR = os.environ['CF_BEHAVIOR']8CF_STAGE = os.environ['CF_STAGE']9cloudfront_client = boto3.client('cloudfront')10lambda_client = boto3.client('lambda', region_name=region)11log = logging.getLogger()12log.setLevel('INFO')13def get_lambda_max_version(func_arn):14    list_result = lambda_client.list_versions_by_function(15        FunctionName=func_arn)16    all_versions = list_result['Versions']17    while 'NextMarker' in list_result:18        list_result = lambda_client.list_versions_by_function(19            FunctionName=func_arn, Marker=list_result['NextMarker'])20        all_versions += list_result['Versions']21    versions_without_latest = list(filter(lambda version: version != '$LATEST', map(22        lambda version: version['Version'], all_versions)))23    versions_without_latest = list(24        map(lambda version: int(version), versions_without_latest))25    if len(versions_without_latest) > 0:26        return max(versions_without_latest)27    else:28        raise ValueError(29            'No version is created in Lambda function: ' + func_arn)30def update_lambda_config(lambda_association, para, stage, func_arn):31    if lambda_association['Quantity'] == 0:32        lambda_association = para['Lambda']33    else:34        stage_occupied = False35        for i in range(len(lambda_association['Items'])):36            if lambda_association['Items'][i]['EventType'] == stage:37                # Replace the existed Lambda with the same stage38                lambda_association['Items'][i]['LambdaFunctionARN'] = func_arn39                stage_occupied = True40                break41        if not stage_occupied:42            lambda_association['Quantity'] = lambda_association['Quantity'] + 143            lambda_association['Items'].append(para['Lambda']['Items'][0])44    return lambda_association45def update_cff_config(cff_association, stage):46    # Remove the cff if the cff is deployed on the same viewer stage with the function47    if stage == 'viewer-request' or stage == 'viewer-response':48        if cff_association['Quantity'] != 0:49            for item in cff_association['Items']:50                if item['EventType'] == stage:51                    cff_association['Quantity'] = cff_association['Quantity'] - 152                    cff_association['Items'].remove(item)53                    break54    return cff_association55def update_cf_config(dist_id, stage, behavior, func_arn):56    '''Associate extensions with the distribution'''57    func_arn = func_arn + ':' + str(get_lambda_max_version(func_arn))58    log.info(func_arn)59    para = {60        'CacheBehavior': behavior,61        'Lambda': {62            'Quantity': 1,63            'Items': [64                {65                    'LambdaFunctionARN': func_arn,66                    'EventType': stage,67                    'IncludeBody': False68                }69            ]70        }71    }72    73    config_resp = cloudfront_client.get_distribution_config(Id=dist_id)74    cf_config = config_resp['DistributionConfig']75    e_tag = config_resp['ETag']76    log.info(cf_config)77    # Add function config into distribution config78    if para['CacheBehavior'] == 'Default (*)':79        cf_config['DefaultCacheBehavior']['LambdaFunctionAssociations'] = update_lambda_config(80            cf_config['DefaultCacheBehavior']['LambdaFunctionAssociations'], para, stage, func_arn)81        cf_config['DefaultCacheBehavior']['FunctionAssociations'] = update_cff_config(82            cf_config['DefaultCacheBehavior']['FunctionAssociations'], stage)83    else:84        for i in range(len(cf_config['CacheBehaviors']['Items'])):85            if cf_config['CacheBehaviors']['Items'][i]['PathPattern'] == para[86                    'CacheBehavior']:87                cf_config['CacheBehaviors']['Items'][i]['LambdaFunctionAssociations'] = update_lambda_config(88                    cf_config['CacheBehaviors']['Items'][i]['LambdaFunctionAssociations'], para, stage, func_arn)89                cf_config['CacheBehaviors']['Items'][i]['FunctionAssociations'] = update_cff_config(90                    cf_config['CacheBehaviors']['Items'][i]['FunctionAssociations'], stage)91                break92    log.info(cf_config)93    update_resp = cloudfront_client.update_distribution(94        DistributionConfig=cf_config, Id=dist_id, IfMatch=e_tag)95    log.info(update_resp)96def lambda_handler(event, context):97    log.info(event)98    request_type = event['RequestType'].upper() if (99        'RequestType' in event) else ""100    log.info(request_type)101    if event['ResourceType'] == "Custom::ResizeImage":102        if 'CREATE' in request_type or 'UPDATE' in request_type:103            behavior_list = CF_BEHAVIOR.replace(104                '[', '').replace(']', '').split(',')105            for behavior in behavior_list:106                behavior = behavior.strip()...

handler.py

Source:handler.py

1"""2Removes old versions of Lambda functions.3"""4import logging5import sys6from pathlib import Path7file = Path(__file__).resolve()8sys.path.append(str(file.parent))9import logger10import boto311# Initialize log12logger.logger_init()13logger = logging.getLogger(__name__)14logger.setLevel(logging.DEBUG)15logger.addHandler(logging.StreamHandler())16logger.propagate = False17try:18    CLIENT = boto3.client('lambda', region_name='eu-west-1')19except Exception as exception:20    logger.error(str(exception), exc_info=True)21# Number of versions to keep22KEEP_LAST = 1023def clean_lambda_versions(event, context):24    """25    List all Lambda functions and call the delete_version function.26    Check if the paginator token exist that's included if more results are available.27    """28    next_marker = None29    try:30        response = CLIENT.list_functions()31    except UnboundLocalError as error:32        logger.error(str(error), exc_info=True)33    except Exception as exception:34        logger.error(str(exception), exc_info=True)35    while next_marker != '':36        next_marker = ''37        for functions in response['Functions']:38            func_arn = functions['FunctionArn']39            # Only production functions (production is part of the name of the lambda)40            if "production" in func_arn:41                delete_version(func_arn)42        # Verify if there is next marker43        if 'NextMarker' in response:44            next_marker = response['NextMarker']45            response = CLIENT.list_functions(46                Marker=next_marker47            )48def delete_version(func_arn):49    """50    Delete a specific function version using Qualifier parameter and keeping the last KEEP_LAST versions.51    Check if the paginator token exist that's included if more results are available.52    """53    next_marker = None54    try:55        response = CLIENT.list_versions_by_function(56            FunctionName=func_arn57        )58    except UnboundLocalError as error:59        logger.error(str(error), exc_info=True)60    except Exception as exception:61        logger.error(str(exception), exc_info=True)62    while next_marker != '':63        next_marker = ''64        function_versions = response['Versions']65        for function_version in function_versions[:-KEEP_LAST]:66            if function_version['Version'] != '$LATEST':67                try:68                    logger.info("Deleting function version: %s:%s ",69                                function_version['FunctionName'], function_version['Version'])70                    CLIENT.delete_function(71                        FunctionName=function_version['FunctionName'],72                        Qualifier=function_version['Version'],73                    )74                    logger.info("Successfully deleted function version: %s:%s",75                                function_version['FunctionName'], function_version['Version'])76                except Exception as exception:77                    logger.error(str(exception), exc_info=True)78        # Verify if there is next marker79        if 'NextMarker' in response:80            next_marker = response['NextMarker']81            response = CLIENT.list_versions_by_function(82                FunctionName=func_arn,83                Marker=next_marker84            )85if __name__ == '__main__':...

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.