Method create_efs

zappa/core.py:1028–1127

Create EFS file system, mount targets, and access point for Lambda. Returns the access point ARN.

(
        self,
        lambda_name: str,
        vpc_config: dict,
        mount_path: str = DEFAULT_EFS_MOUNT_POINT,
        throughput_mode: str = "bursting",
        performance_mode: str = "generalPurpose",
    )

    ##

    def create_efs(
        self,
        lambda_name: str,
        vpc_config: dict,
        mount_path: str = DEFAULT_EFS_MOUNT_POINT,
        throughput_mode: str = "bursting",
        performance_mode: str = "generalPurpose",
    ) -> str:
        """
        Create EFS file system, mount targets, and access point for Lambda.
        Returns the access point ARN.
        """
        # Generate unique name based on mount path (e.g., /mnt/data -> data, /mnt/ -> default)
        mount_suffix = mount_path.replace("/mnt/", "").rstrip("/") or "default"
        efs_name = f"{lambda_name}-efs-{mount_suffix}"

        # Check if EFS already exists (by name tag)
        existing = self.efs_client.describe_file_systems()
        file_system_id = None
        for fs in existing["FileSystems"]:
            tags = {t["Key"]: t["Value"] for t in fs.get("Tags", [])}
            if tags.get("Name") == efs_name:
                file_system_id = fs["FileSystemId"]
                logger.info(f"Using existing EFS: {file_system_id}")
                break

        if file_system_id is None:
            # Create new file system
            logger.info(f"Creating EFS file system: {efs_name}")
            response = self.efs_client.create_file_system(
                CreationToken=efs_name,
                PerformanceMode=performance_mode,
                ThroughputMode=throughput_mode,
                Encrypted=True,
                Tags=[
                    {"Key": "Name", "Value": efs_name},
                    {"Key": "CreatedBy", "Value": "Zappa"},
                ],
            )
            file_system_id = response["FileSystemId"]

            # Wait for file system to be available
            while True:
                fs_response = self.efs_client.describe_file_systems(FileSystemId=file_system_id)
                if fs_response["FileSystems"][0]["LifeCycleState"] == "available":
                    break
                time.sleep(5)

        # Create mount targets for each subnet
        security_group_ids = vpc_config.get("SecurityGroupIds", [])
        for subnet_id in vpc_config.get("SubnetIds", []):
            try:
                self.efs_client.create_mount_target(
                    FileSystemId=file_system_id,
                    SubnetId=subnet_id,
                    SecurityGroups=security_group_ids,
                )
                logger.info(f"Created mount target in subnet: {subnet_id}")