Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_code_path_for_function method in localstack

Best Python code snippet using localstack_python

runtime_executor.py

Source:runtime_executor.py

...45def get_path_for_function(function_version: FunctionVersion) -> Path:46    return Path(47        f"{config.dirs.tmp}/lambda/{function_version.id.qualified_arn().replace(':', '_').replace('$', '_')}/"48    )49def get_code_path_for_function(function_version: FunctionVersion) -> Path:50    return get_path_for_function(function_version) / "code"51def get_image_name_for_function(function_version: FunctionVersion) -> str:52    return f"localstack/lambda-{function_version.id.qualified_arn().replace(':', '_').replace('$', '_').lower()}"53def get_image_for_runtime(runtime: str) -> str:54    runtime, version = get_runtime_split(runtime)55    return f"{IMAGE_PREFIX}{runtime}:{version}"56def get_runtime_client_path() -> Path:57    return Path(LAMBDA_RUNTIME_INIT_PATH)58def prepare_image(target_path: Path, function_version: FunctionVersion) -> None:59    if not function_version.config.runtime:60        raise NotImplementedError("Custom images are currently not supported")61    src_init = get_runtime_client_path()62    # copy init file63    target_init = target_path / "aws-lambda-rie"64    shutil.copy(src_init, target_init)65    target_init.chmod(0o755)66    # copy code67    # create dockerfile68    docker_file_path = target_path / "Dockerfile"69    docker_file = LAMBDA_DOCKERFILE.format(70        base_img=get_image_for_runtime(function_version.config.runtime),71        rapid_entrypoint=RAPID_ENTRYPOINT,72    )73    with docker_file_path.open(mode="w") as f:74        f.write(docker_file)75    try:76        CONTAINER_CLIENT.build_image(77            dockerfile_path=str(docker_file_path),78            image_name=get_image_name_for_function(function_version),79        )80    except Exception as e:81        if LOG.isEnabledFor(logging.DEBUG):82            LOG.exception(83                "Error while building prebuilt lambda image for '%s'",84                function_version.qualified_arn,85            )86        else:87            LOG.error(88                "Error while building prebuilt lambda image for '%s', Error: %s",89                function_version.qualified_arn,90                e,91            )92def prepare_version(function_version: FunctionVersion) -> None:93    if not function_version.code.zip_file:94        raise NotImplementedError("Images without zipfile are currently not supported")95    time_before = time.perf_counter()96    target_path = get_path_for_function(function_version)97    target_path.mkdir(parents=True, exist_ok=True)98    # write code to disk99    target_code = get_code_path_for_function(function_version)100    with NamedTemporaryFile() as file:101        file.write(function_version.code.zip_file)102        file.flush()103        unzip(file.name, str(target_code))104    if config.LAMBDA_PREBUILD_IMAGES:105        prepare_image(target_path, function_version)106    LOG.debug("Version preparation took %0.2fms", (time.perf_counter() - time_before) * 1000)107def cleanup_version(function_version: FunctionVersion) -> None:108    function_path = get_path_for_function(function_version)109    try:110        shutil.rmtree(function_path)111    except OSError as e:112        LOG.debug(113            "Could not cleanup function %s due to error %s while deleting file %s",114            function_version.qualified_arn,115            e.strerror,116            e.filename,117        )118    if config.LAMBDA_PREBUILD_IMAGES:119        CONTAINER_CLIENT.remove_image(get_image_name_for_function(function_version))120class LambdaRuntimeException(Exception):121    def __init__(self, message: str):122        super().__init__(message)123class RuntimeExecutor:124    id: str125    function_version: FunctionVersion126    ip: Optional[str]127    executor_endpoint: Optional[ExecutorEndpoint]128    def __init__(129        self, id: str, function_version: FunctionVersion, service_endpoint: ServiceEndpoint130    ) -> None:131        self.id = id132        self.function_version = function_version133        self.ip = None134        self.executor_endpoint = self._build_executor_endpoint(service_endpoint)135    def get_image(self) -> str:136        if not self.function_version.config.runtime:137            raise NotImplementedError("Custom images are currently not supported")138        return (139            get_image_name_for_function(self.function_version)140            if config.LAMBDA_PREBUILD_IMAGES141            else get_image_for_runtime(self.function_version.config.runtime)142        )143    def _build_executor_endpoint(self, service_endpoint: ServiceEndpoint) -> ExecutorEndpoint:144        port = get_free_tcp_port()145        LOG.debug(146            "Creating service endpoint for function %s executor %s",147            self.function_version.qualified_arn,148            self.id,149        )150        executor_endpoint = ExecutorEndpoint(port, service_endpoint=service_endpoint)151        LOG.debug(152            "Finished creating service endpoint for function %s executor %s",153            self.function_version.qualified_arn,154            self.id,155        )156        return executor_endpoint157    def start(self, env_vars: Dict[str, str]) -> None:158        self.executor_endpoint.start()159        network = self.get_network_for_executor()160        container_config = ContainerConfiguration(161            image_name=self.get_image(),162            name=self.id,163            env_vars=env_vars,164            network=network,165            entrypoint=RAPID_ENTRYPOINT,166        )167        CONTAINER_CLIENT.create_container_from_config(container_config)168        if not config.LAMBDA_PREBUILD_IMAGES:169            CONTAINER_CLIENT.copy_into_container(170                self.id, str(get_runtime_client_path()), RAPID_ENTRYPOINT171            )172            CONTAINER_CLIENT.copy_into_container(173                self.id, f"{str(get_code_path_for_function(self.function_version))}/", "/var/task/"174            )175        CONTAINER_CLIENT.start_container(self.id)176        self.ip = CONTAINER_CLIENT.get_container_ipv4_for_network(177            container_name_or_id=self.id, container_network=network178        )179        self.executor_endpoint.container_address = self.ip180    def stop(self) -> None:181        CONTAINER_CLIENT.stop_container(container_name=self.id, timeout=5)182        CONTAINER_CLIENT.remove_container(container_name=self.id)183        try:184            self.executor_endpoint.shutdown()185        except Exception as e:186            LOG.debug(187                "Error while stopping executor endpoint for lambda %s, error: %s",...

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.