How to use get_lambda_version_manager method in localstack

Best Python code snippet using localstack_python

lambda_service.py

Source:lambda_service.py

...54        version_manager = self.lambda_version_managers.pop(qualified_arn)55        if not version_manager:56            raise ValueError(f"Unable to find version manager for {qualified_arn}")57        self.task_executor.submit(version_manager.stop)58    def get_lambda_version_manager(self, function_arn: str) -> LambdaVersionManager:59        """60        Get the lambda version for the given arn61        :param function_arn: qualified arn for the lambda version62        :return: LambdaVersionManager for the arn63        """64        version_manager = self.lambda_version_managers.get(function_arn)65        if not version_manager:66            raise ValueError(f"Could not find version '{function_arn}'. Is it created?")67        return version_manager68    def create_function(69        self,70        account_id: str,71        region_name: str,72        function_name: str,73        function_config: VersionFunctionConfiguration,74        code: Code,75    ) -> FunctionVersion:76        state = LambdaServiceBackend.get(region_name)77        fn = Function(function_name=function_name)78        with self.create_fn_lock:79            arn = VersionIdentifier(function_name, "$LATEST", region_name, account_id)80            zip_file_content = code.zip_file81            code_sha_256 = base64.standard_b64encode(82                hashlib.sha256(zip_file_content).digest()83            ).decode("utf-8")84            version = FunctionVersion(85                id=arn,86                qualifier="$LATEST",87                code=code,88                config_meta=FunctionConfigurationMeta(89                    function_arn=arn.qualified_arn(),90                    revision_id="?",91                    code_size=len(code.zip_file),92                    coda_sha256=code_sha_256,93                    last_modified="asdf",94                    last_update=UpdateStatus(status="Successful"),95                ),96                config=function_config,97            )98            fn.versions["$LATEST"] = version99            state.functions[function_name] = fn100        self.create_function_version(version)101        return version102    # TODO: is this sync?103    def delete_function(self, region_name: str, function_name: str):104        state = LambdaServiceBackend.get(region_name)105        function = state.functions.pop(function_name)106        for version in function.versions.values():107            self.stop_version(qualified_arn=version.id.qualified_arn())108    def delete_version(self, region_name: str, function_name: str, version_qualifier: str):109        state = LambdaServiceBackend.get(region_name)110        version = state.functions[function_name].versions[version_qualifier]111        self.stop_version(qualified_arn=version.id.qualified_arn())112    def get_function_version(113        self, region_name: str, function_name: str, qualifier: Optional[str] = "$LATEST"114    ) -> FunctionVersion:115        state = LambdaServiceBackend.get(region_name)116        return state.functions[function_name].versions[qualifier]117    def list_function_versions(self, region_name: str) -> List[FunctionVersion]:118        state = LambdaServiceBackend.get(region_name)119        return [f.latest() for f in state.functions.values()]  # TODO: qualifier120    def create_function_version(self, function_version: FunctionVersion) -> None:121        with self.lambda_version_manager_lock:122            qualified_arn = function_version.id.qualified_arn()123            version_manager = self.lambda_version_managers.get(qualified_arn)124            if version_manager:125                raise Exception("Version '%s' already created", qualified_arn)126            version_manager = LambdaVersionManager(127                function_arn=qualified_arn,128                function_version=function_version,129            )130            self.lambda_version_managers[qualified_arn] = version_manager131            self.task_executor.submit(version_manager.start)132    # Commands133    def invoke(134        self,135        function_arn_qualified: str,136        invocation_type: str,137        client_context: Optional[str],138        payload: bytes,139    ) -> "Future[InvocationResult]":140        version_manager = self.get_lambda_version_manager(function_arn_qualified)141        return version_manager.invoke(142            invocation=Invocation(143                payload=payload, client_context=client_context, invocation_type=invocation_type144            )...

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.