KaneAI - World’s First E2E Software Testing Agent.Signup for Private Beta Join TestMu Conference

How to use publish_layer_version method in localstack

Best Python code snippet using localstack_python

lambda_function.py

Source:lambda_function.py

...64            #     logs = retval.pop("logs")65            #     ops = retval.get("ops")66            props = None67            # if ops.get("publish_layer_version"):68            retval = publish_layer_version(desired_config, logs, ops)69            if retval.get("statusCode"):70                return retval            71            logs = retval.pop("logs")72            ops = retval.get("ops")73            state = retval.get("state")74            props = {75                "layer_name": layer_name,76                **remove_none_attributes(state)77            } if state else {}78            print(f"props: {props}")79            return creturn(200, 100, success=True, logs=logs, 80                state=state, 81                props=props,82                links={83                    "Layer": gen_layer_link(layer_name, region)84                }85            )86            87        elif event.get("op") == "delete":88            layer_name = prev_state['props'].get("layer_name")89            ops = {"remove_layer": layer_name}90            retval = remove_layer(layer_name, logs, ops)91            logs = retval.pop("logs")92            return creturn(200, 100, success=True, logs=logs)93    except Exception as e:94        msg = traceback.format_exc()95        print(msg)96        return creturn(400, 0, logs=logs, error=msg)97def publish_layer_version(desired_config, logs, ops):98    lambda_client = boto3.client("lambda")99    print(f"Inside publish_layer_version, desired_config = {desired_config}")100    try:101        lambda_response = lambda_client.publish_layer_version(102            **desired_config103        )104        logs.append(gen_log("Published Layer Version", lambda_response))105    except botocore.exceptions.ClientError as e:106        logs.append(gen_log(e.response["Error"]["Code"], {"error": str(e)}, is_error=True))107        msg = traceback.format_exc()108        print(msg)109        return creturn(400, 60, logs=logs, error = msg)110        # if e.response['Error']['Code'] in ['PreconditionFailed', 'CodeVerificationFailed', 'InvalidCodeSignature', 'CodeSigningConfigNotFound']:111        #     return creturn(200, 60, logs=logs, error = str(e))112        # else:113        #     print(f'Reached other exceptions, exception is {str(e)}')114        #     return creturn(200, 60, pass_back_data={115        #         "ops": ops,...

create_serverless_layer_version_use_case.py

Source:create_serverless_layer_version_use_case.py

1import os2from time import strftime3from typing import Dict4from entities.artifact import Artifact5from entities.bucket import Bucket6from entities.layer import Layer7from entities.serverless_service import ServerlessService8from use_cases.services.publish_layer_version.publish_layer_version import PublishLayerVersion9from use_cases.services.update_serverless_layer.update_serverless_layer import UpdateServerlessLayer10from use_cases.services.upload_code_service.upload_code_service import UploadCodeService11class CreateServerlessLayerVersionUseCase:12    def __init__(13        self,14        upload_code_service: UploadCodeService,15        publish_layer_version: PublishLayerVersion,16        update_serverless_layer: UpdateServerlessLayer17    ):18        self.__upload_code_service = upload_code_service19        self.__publish_layer_version = publish_layer_version20        self.__update_serverless_layer = update_serverless_layer21    def create(22        self,23        environment: str,24        serverless_info: Dict,25        layer_zip_file_name: str26    ) -> None:27        version = strftime("%Y%m%d%H%M%S")28        file_name = "{}_layers/{}_{}_build.zip".format(29            environment,30            version,31            serverless_info['id']32        )33        artifact = Artifact(34            file_name=file_name,35            temp_path=layer_zip_file_name36        )37        bucket_name = os.getenv('BUCKET_{}'.format(environment))38        bucket = Bucket(39            name=bucket_name,40            environment=environment41        )42        self.__upload_code_service.upload(43            bucket,44            artifact45        )46        layer_arn = self.__publish_layer_version.publish(47            bucket,48            artifact49        )50        serverless_service = ServerlessService(51            name=serverless_info['id'],52            environment=environment53        )54        layer = Layer(id=layer_arn)55        self.__update_serverless_layer.update(56            serverless_service,57            layer...

aws_publish_layer_version.py

Source:aws_publish_layer_version.py

...18            aws_secret_access_key=os.getenv('SECRET_ACCESS_KEY'),19            region_name=os.getenv('REGION_NAME')20        )21        version = strftime("%Y%m%d%H%M%S")22        response = client.publish_layer_version(23            LayerName=os.getenv('LAYER_NAME'),24            Description=version,25            Content={26                'S3Bucket': bucket.name,27                'S3Key': artifact.file_name,28            },29            CompatibleRuntimes=[30                'python3.8'31            ]32        )33        if response['ResponseMetadata']['HTTPStatusCode'] != self.STATUS_CODE_OK:34            raise Exception('Error publishing layer:\n{}'.format(str(response)))...

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.