Method load_settings

zappa/cli.py:2581–2843

Load the local zappa_settings file. An existing boto session can be supplied, though this is likely for testing purposes. Returns the loaded Zappa object.

(self, settings_file=None, session=None)

            return

    def load_settings(self, settings_file=None, session=None):
        """
        Load the local zappa_settings file.
        An existing boto session can be supplied, though this is likely for testing purposes.
        Returns the loaded Zappa object.
        """

        # Load up settings (from file or environment)
        self.load_settings_file(settings_file)

        # Make sure that this stage is our settings
        if self.api_stage not in self.zappa_settings.keys():
            raise ClickException("Please define stage '{0!s}' in your Zappa settings.".format(self.api_stage))

        # We need a working title for this project. Use one if supplied, else cwd dirname.
        if "project_name" in self.stage_config:  # pragma: no cover
            # If the name is invalid, this will throw an exception with message up stack
            self.project_name = validate_name(self.stage_config["project_name"])
        else:
            self.project_name = self.get_project_name()

        # The name of the actual AWS Lambda function, ex, 'helloworld-dev'
        # Assume that we already have have validated the name beforehand.
        # Related:  https://github.com/Miserlou/Zappa/pull/664
        #           https://github.com/Miserlou/Zappa/issues/678
        #           And various others from Slack.
        self.lambda_name = slugify.slugify(self.project_name + "-" + self.api_stage)

        # Load stage-specific settings
        self.s3_bucket_name = self.stage_config.get(
            "s3_bucket",
            "zappa-" + "".join(random.choice(string.ascii_lowercase + string.digits) for _ in range(9)),
        )
        self.vpc_config = self.stage_config.get("vpc_config", {})

        # Load and validate EFS configuration
        raw_efs_config = self.stage_config.get("efs_config", None)
        self.efs_config = []

        if raw_efs_config:
            if not self.vpc_config:
                raise ClickException("efs_config requires vpc_config to be set")

            # Normalize configuration to list format
            if raw_efs_config is True:
                # Minimal config: efs_config: true -> single mount at default path
                self.efs_config = [{"LocalMountPath": DEFAULT_EFS_MOUNT_POINT}]
            elif isinstance(raw_efs_config, dict):
                # Single dict config -> wrap in list
                self.efs_config = [raw_efs_config]
            elif isinstance(raw_efs_config, list):
                self.efs_config = raw_efs_config

            # Validate each EFS entry
            mount_paths = set()
            for idx, efs in enumerate(self.efs_config):
                # Default LocalMountPath
                if "LocalMountPath" not in efs: