diff --git a/fbpcs/infra/pce_deployment_library/cloud_library/aws/aws.py b/fbpcs/infra/pce_deployment_library/cloud_library/aws/aws.py deleted file mode 100644 index 0a2f8a108..000000000 --- a/fbpcs/infra/pce_deployment_library/cloud_library/aws/aws.py +++ /dev/null @@ -1,203 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict -import logging -import os -from typing import Optional - -import boto3 - -import botocore - -from botocore.exceptions import ClientError, NoCredentialsError -from fbpcs.infra.pce_deployment_library.cloud_library.cloud_base.cloud_base import ( - CloudBase, -) -from fbpcs.infra.pce_deployment_library.cloud_library.defaults import CloudPlatforms -from fbpcs.infra.pce_deployment_library.errors_library.aws_errors import ( - AccessDeniedError, - S3BucketCreationError, - S3BucketDeleteError, - S3BucketDoesntExist, - S3BucketVersioningFailedError, -) - - -class AWS(CloudBase): - def __init__( - self, - aws_access_key_id: Optional[str] = None, - aws_secret_access_key: Optional[str] = None, - aws_session_token: Optional[str] = None, - aws_region: Optional[str] = None, - ) -> None: - - aws_access_key_id = aws_access_key_id or os.environ.get("AWS_ACCESS_KEY_ID") - aws_secret_access_key = aws_secret_access_key or os.environ.get( - "AWS_SECRET_ACCESS_KEY" - ) - aws_session_token = aws_session_token or os.environ.get("AWS_SESSION_TOKEN") - self.aws_region: Optional[str] = aws_region or os.environ.get("AWS_REGION") - - self.log: logging.Logger = logging.getLogger(__name__) - self.__account_id: Optional[str] = None - - try: - self.sts: botocore.client.BaseClient = boto3.client( - "sts", - aws_access_key_id=aws_access_key_id, - aws_secret_access_key=aws_secret_access_key, - aws_session_token=aws_session_token, - ) - self.s3_client: botocore.client.BaseClient = boto3.client( - "s3", - aws_access_key_id=aws_access_key_id, - aws_secret_access_key=aws_secret_access_key, - aws_session_token=aws_session_token, - region_name=aws_region, - ) - - except NoCredentialsError as error: - self.log.error( - f"Error occurred in validating access and secret keys of the aws account.\n" - "Please verify if the correct access and secret key of root user are provided.\n" - "Access and secret key can be passed using:\n" - "1. Passing as variable to class object\n" - "2. Placing keys in ~/.aws/config\n" - "3. Placing keys in ~/.aws/credentials\n" - "4. As environment variables\n" - "\n" - "Please refer to: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html\n" - "\n" - "Following is the error:\n" - f"{error}" - ) - try: - self.log.info("Verifying AWS credentials.") - response = self.sts.get_caller_identity() - - # fetching account ID for the given credentials - self.__account_id = response.get("Account", None) - except NoCredentialsError as error: - self.log.error(f"Couldn't validate the AWS credentials." f"{error}") - - @classmethod - def cloud_type(cls) -> CloudPlatforms: - return CloudPlatforms.AWS - - def check_s3_buckets_exists( - self, s3_bucket_name: str, bucket_version: bool = True - ) -> None: - """ - Checks for the S3 bucket. If not found creates one. - """ - try: - self.log.info(f"Checking if S3 bucket {s3_bucket_name} exists.") - self.s3_client.head_bucket(Bucket=s3_bucket_name) - self.log.info( - f"S3 bucket {s3_bucket_name} already exists in the AWS account." - ) - except ClientError as error: - if error.response["Error"]["Code"] == "404": - # Error reponse was 404 which means bucket doesn't exist. - # In this case creates a new bucket - self.log.info( - f"S3 bucket {s3_bucket_name} deosn't exists in the AWS account." - ) - self.create_s3_bucket( - s3_bucket_name=s3_bucket_name, bucket_version=bucket_version - ) - elif error.response["Error"]["Code"] == "403": - # Error reponse was 403 which means user doesn't have access to this bucket - raise AccessDeniedError("Access denied") from error - else: - raise S3BucketCreationError( - f"Couldn't create bucket {s3_bucket_name}" - ) from error - - def create_s3_bucket( - self, s3_bucket_name: str, bucket_version: bool = True - ) -> None: - bucket_configuration = {"LocationConstraint": self.aws_region} - - try: - self.log.info(f"Creating new S3 bucket {s3_bucket_name}") - self.s3_client.create_bucket( - Bucket=s3_bucket_name, - CreateBucketConfiguration=bucket_configuration, - ) - self.log.info( - f"Create S3 bucket {s3_bucket_name} operation was successful." - ) - except ClientError as error: - error_code = error.response.get("Error", {}).get("Code", None) - raise S3BucketCreationError( - f"Failed to create S3 bucket with error code {error_code}" - ) from error - - if bucket_version: - self.update_bucket_versioning(s3_bucket_name=s3_bucket_name) - - def update_bucket_versioning( - self, s3_bucket_name: str, versioning_status: Optional[str] = "Enabled" - ) -> None: - versioning_configuration = {"Status": versioning_status} - try: - self.log.info("Creating bucket versioning.") - self.s3_client.put_bucket_versioning( - Bucket=s3_bucket_name, VersioningConfiguration=versioning_configuration - ) - self.log.info(f"Bucket {s3_bucket_name} is enabled with versioning.") - except ClientError as error: - if error.response["Error"]["Code"] == "404": - raise S3BucketDoesntExist( - f"S3 bucket {s3_bucket_name} doesn't exist" - ) from error - elif error.response["Error"]["Code"] == "403": - raise AccessDeniedError("Access denied") from error - else: - raise S3BucketVersioningFailedError( - f"Error in versioning S3 bucket {s3_bucket_name}" - ) from error - - def delete_s3_bucket(self, s3_bucket_name: str) -> None: - try: - self.log.info(f"Deleting S3 bucket {s3_bucket_name}") - self.s3_client.delete_bucket(Bucket=s3_bucket_name) - self.log.info( - f"Delete S3 bucket {s3_bucket_name} operation was successful." - ) - except ClientError as error: - raise S3BucketDeleteError( - f"Error in deleting bucket {s3_bucket_name}" - ) from error - - def check_s3_object_exists( - self, s3_bucket_name: str, key_name: str, account_id: Optional[str] = "" - ) -> bool: - account_id = account_id or self.__account_id - try: - self.log.info(f"Checking for file {key_name} in bucket {s3_bucket_name}") - self.s3_client.head_object( - Bucket=s3_bucket_name, Key=key_name, ExpectedBucketOwner=account_id - ) - self.log.info(f"File {key_name} exists.") - return True - except ClientError as error: - if error.response["Error"]["Code"] == "404": - self.log.error( - f"Couldn't find file {key_name} in bucket {s3_bucket_name}" - ) - elif error.response["Error"]["Code"] == "403": - self.log.error( - f"Access denied: failed to access bucket {s3_bucket_name}" - ) - else: - self.log.error( - f"Failed to find file {key_name} in bucket {s3_bucket_name}" - ) - self.log.info(f"File {key_name} doesn't exist.") - return False diff --git a/fbpcs/infra/pce_deployment_library/cloud_library/cloud_base/cloud_base.py b/fbpcs/infra/pce_deployment_library/cloud_library/cloud_base/cloud_base.py deleted file mode 100644 index 1ec8dab69..000000000 --- a/fbpcs/infra/pce_deployment_library/cloud_library/cloud_base/cloud_base.py +++ /dev/null @@ -1,17 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict - -from abc import ABC, abstractmethod - -from fbpcs.infra.pce_deployment_library.cloud_library.defaults import CloudPlatforms - - -class CloudBase(ABC): - @classmethod - @abstractmethod - def cloud_type(cls) -> CloudPlatforms: - pass diff --git a/fbpcs/infra/pce_deployment_library/cloud_library/cloud_factory.py b/fbpcs/infra/pce_deployment_library/cloud_library/cloud_factory.py deleted file mode 100644 index 5c7fe4d4f..000000000 --- a/fbpcs/infra/pce_deployment_library/cloud_library/cloud_factory.py +++ /dev/null @@ -1,34 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict -from typing import Any, Dict, List, Type, Union - -from fbpcs.infra.pce_deployment_library.cloud_library.aws.aws import AWS -from fbpcs.infra.pce_deployment_library.cloud_library.cloud_base.cloud_base import ( - CloudBase, -) -from fbpcs.infra.pce_deployment_library.cloud_library.defaults import CloudPlatforms -from fbpcs.infra.pce_deployment_library.cloud_library.gcp.gcp import GCP - - -class CloudFactory: - CLOUD_TYPES: Dict[CloudPlatforms, Type[Union[AWS, GCP]]] = { - CloudPlatforms.AWS: AWS, - CloudPlatforms.GCP: GCP, - } - - def create_cloud_object( - self, cloud_type: CloudPlatforms, **kwargs: Any - ) -> CloudBase: - supported_cloud_platform = self.get_supported_cloud_platforms() - if self.CLOUD_TYPES.get(cloud_type, None) is None: - raise Exception( - f"{cloud_type} is not a supported cloud platform. Supported platforms are {supported_cloud_platform}" - ) - return self.CLOUD_TYPES[cloud_type](**kwargs) - - def get_supported_cloud_platforms(self) -> List[str]: - return CloudPlatforms.list() diff --git a/fbpcs/infra/pce_deployment_library/cloud_library/defaults.py b/fbpcs/infra/pce_deployment_library/cloud_library/defaults.py deleted file mode 100644 index 10a3e3c11..000000000 --- a/fbpcs/infra/pce_deployment_library/cloud_library/defaults.py +++ /dev/null @@ -1,17 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict -from enum import Enum -from typing import List - - -class CloudPlatforms(str, Enum): - AWS = "aws" - GCP = "gcp" - - @classmethod - def list(cls) -> List[str]: - return [e.value for e in CloudPlatforms] diff --git a/fbpcs/infra/pce_deployment_library/cloud_library/gcp/gcp.py b/fbpcs/infra/pce_deployment_library/cloud_library/gcp/gcp.py deleted file mode 100644 index aee2fc8ec..000000000 --- a/fbpcs/infra/pce_deployment_library/cloud_library/gcp/gcp.py +++ /dev/null @@ -1,16 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict -from fbpcs.infra.pce_deployment_library.cloud_library.cloud_base.cloud_base import ( - CloudBase, -) -from fbpcs.infra.pce_deployment_library.cloud_library.defaults import CloudPlatforms - - -class GCP(CloudBase): - @classmethod - def cloud_type(cls) -> CloudPlatforms: - return CloudPlatforms.GCP diff --git a/fbpcs/infra/pce_deployment_library/deploy_library/deploy_base/deploy_base.py b/fbpcs/infra/pce_deployment_library/deploy_library/deploy_base/deploy_base.py deleted file mode 100644 index c8757f4a6..000000000 --- a/fbpcs/infra/pce_deployment_library/deploy_library/deploy_base/deploy_base.py +++ /dev/null @@ -1,24 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict - -from abc import ABC, abstractmethod - -from fbpcs.infra.pce_deployment_library.deploy_library.models import RunCommandResult - - -class DeployBase(ABC): - @abstractmethod - def create(self) -> RunCommandResult: - pass - - @abstractmethod - def destroy(self) -> RunCommandResult: - pass - - @abstractmethod - def run_command(self) -> RunCommandResult: - pass diff --git a/fbpcs/infra/pce_deployment_library/deploy_library/models.py b/fbpcs/infra/pce_deployment_library/deploy_library/models.py deleted file mode 100644 index 9ef05f48d..000000000 --- a/fbpcs/infra/pce_deployment_library/deploy_library/models.py +++ /dev/null @@ -1,80 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict - -from dataclasses import dataclass -from enum import Enum -from typing import List, Optional - - -@dataclass -class RunCommandResult: - return_code: int - output: Optional[str] - error: Optional[str] - - -@dataclass -class TerraformCliOptions: - state: str = "state" - target: str = "target" - var: str = "var" - var_file: str = "var_file" - parallelism: str = "parallelism" - terraform_input: str = "input" - backend_config: str = "backend_config" - reconfigure: str = "reconfigure" - - -NOT_SUPPORTED_INIT_DEFAULT_OPTIONS: List[str] = [ - TerraformCliOptions.state, - TerraformCliOptions.parallelism, -] - - -class TerraformCommand(str, Enum): - INIT: str = "init" - APPLY: str = "apply" - DESTROY: str = "destroy" - PLAN: str = "plan" - OUTPUT: str = "output" - - -class TerraformOptionFlag: - pass - - -class FlaggedOption(TerraformOptionFlag): - """ - Used to set flag options, eg, `terraform init -reconfigure` - `-reconfigure` is a flagged option here. - - This should not be confused with the options that accept bool values. - In case of options that accept bool values, explicit bool value is passed. - Eg of bool option: `terraform apply -input=false` - - Usage of FlaggedOption: - t = TerraformDeployment() - t.terraform_init(reconfigure=FlaggedOption) - - Results in : `terraform init -reconfigure` - """ - - pass - - -class NotFlaggedOption(TerraformOptionFlag): - """ - Is opposite of the FlaggedOption and is used to unset flag options. - - Usage: - t = TerraformDeployment() - t.terraform_init(reconfigure=NotFlaggedOption) - - Results in : `terraform init` - """ - - pass diff --git a/fbpcs/infra/pce_deployment_library/deploy_library/terraform_library/terraform_deployment.py b/fbpcs/infra/pce_deployment_library/deploy_library/terraform_library/terraform_deployment.py deleted file mode 100644 index 97f706ab9..000000000 --- a/fbpcs/infra/pce_deployment_library/deploy_library/terraform_library/terraform_deployment.py +++ /dev/null @@ -1,241 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict - -import logging -import os -import sys -from subprocess import PIPE, Popen -from typing import Any, Dict, List, Optional, Type - -from fbpcs.infra.pce_deployment_library.deploy_library.deploy_base.deploy_base import ( - DeployBase, -) - -from fbpcs.infra.pce_deployment_library.deploy_library.models import ( - FlaggedOption, - RunCommandResult, - TerraformCliOptions, - TerraformCommand, - TerraformOptionFlag, -) -from fbpcs.infra.pce_deployment_library.deploy_library.terraform_library.terraform_deployment_utils import ( - TerraformDeploymentUtils, -) - - -class TerraformDeployment(DeployBase): - - TERRAFORM_DEFAULT_PARALLELISM = 10 - - def __init__( - self, - state_file_path: Optional[str] = None, - terraform_variables: Optional[Dict[str, str]] = None, - parallelism: Optional[int] = None, - resource_targets: Optional[List[str]] = None, - var_definition_file: Optional[str] = None, - working_directory: Optional[str] = None, - ) -> None: - """ - Accepts options to create Terraform CLIs apply/destroy/plan/init - Args: - state_file_path: Path to store terraform state files - More information about terraform state: https://www.terraform.io/language/state - variables: -var option in terraform CLI. This arguments provides default variables. - These variables can be overwritten by commands also. - More information on terraform vairables: https://www.terraform.io/language/values/variables - parallelism: -parallelism=n option in Terraform CLI. - Limits the number of concurrent operation as Terraform walks the graph - More information on terraform parallelism: https://www.terraform.io/cli/commands/apply#parallelism-n - resource_targets: -target option in Terraform CLI. Used to target specific resource in terraform apply/destroy - More information on terraform targets: https://learn.hashicorp.com/tutorials/terraform/resource-targeting - var_definition_file: -var-file option in Terraform CLI. Used to define terraform variables in bulk though .tfvars file - More information on var_definition_file :https://www.terraform.io/language/values/variables#variable-definitions-tfvars-files - working_directory: Directory in which terraform files are present - """ - self.log: logging.Logger = logging.getLogger(__name__) - self.utils = TerraformDeploymentUtils( - state_file_path=state_file_path, - resource_targets=resource_targets, - terraform_variables=terraform_variables, - parallelism=parallelism, - var_definition_file=var_definition_file, - ) - self._working_directory: str = working_directory or os.path.dirname(__file__) - - @property - def working_directory(self) -> str: - return self._working_directory - - @working_directory.setter - def working_directory(self, working_dir: str) -> None: - self._working_directory = working_dir - - def create( - self, - terraform_input: bool = False, - auto_approve: Type[TerraformOptionFlag] = FlaggedOption, - **kwargs: Dict[str, Any], - ) -> RunCommandResult: - """ - Implements `terraform apply` of terraform CLI. - `terraform apply` command executes the actions proposed in a `terraform plan`. - - More information: https://www.terraform.io/cli/commands/apply - - terraform_input: - Provides `-input=false`. It disables all of Terraform's interactive prompts. - More information: https://www.terraform.io/cli/commands/apply#apply-options - - auto_approve: - Skips interactive approval of plan before applying - More information: https://www.terraform.io/cli/commands/apply#apply-options - """ - options: Dict[str, Any] = kwargs.copy() - options["input"] = terraform_input - options["auto-approve"] = auto_approve # a False value will require an input - options = self.utils.get_default_options(TerraformCommand.APPLY, options) - return self.run_command("terraform apply", **options) - - def destroy( - self, auto_approve: bool = True, **kwargs: Dict[str, Any] - ) -> RunCommandResult: - """ - Implements `terraform destroy` of terraform CLI. - `terraform destroy` destroys all remote objects managed by a particular Terraform configuration. - - More information: https://www.terraform.io/docs/commands/destroy.html - - auto_approve: - Skips interactive approval of plan before applying - More information: https://www.terraform.io/cli/commands/apply#apply-options - - """ - options: Dict[str, Any] = kwargs.copy() - options["auto-approve"] = auto_approve - - options = self.utils.get_default_options(TerraformCommand.DESTROY, options) - return self.run_command("terraform destroy", **options) - - def terraform_init( - self, - backend_config: Optional[Dict[str, str]] = None, - reconfigure: Type[TerraformOptionFlag] = FlaggedOption, - **kwargs: Dict[str, Any], - ) -> RunCommandResult: - """ - Implements `terraform init` of terraform CLI. - `terraform init` command is used to initialize a working directory containing Terraform configuration files. - - More information: https://www.terraform.io/cli/commands/init - - backend_config: - Provides backend config information using key-value pairs. - Usage: - terraform = Terraform() - terraform.terraform_init(backend_config = {"bucket": s3_bucket,"region": region}) - More info: https://www.terraform.io/language/settings/backends/configuration#command-line-key-value-pairs - - reconfigure: - in `terraform init` reconfigure disregards any existing configuration, preventing migration of any existing state. - More information: https://www.terraform.io/cli/commands/init#backend-initialization - """ - options: Dict[str, Any] = kwargs.copy() - options.update( - { - TerraformCliOptions.backend_config: backend_config, - TerraformCliOptions.reconfigure: reconfigure, - } - ) - options = self.utils.get_default_options(TerraformCommand.INIT, options) - return self.run_command("terraform init", **options) - - def plan( - self, - detailed_exitcode: Type[FlaggedOption] = FlaggedOption, - **kwargs: Dict[str, Any], - ) -> RunCommandResult: - """ - Implements `terraform plan` of terraform CLI. - `terraform plan` creates an execution plan, which lets you preview the changes that Terraform plans to make to your infrastructure - - More information: https://www.terraform.io/cli/commands/plan - - detailed_exitcode: - Returns a detailed exit code when the command exits - More information: https://www.terraform.io/cli/commands/plan#other-options - """ - options: Dict[str, Any] = kwargs.copy() - options["detailed_exitcode"] = detailed_exitcode - options = self.utils.get_default_options(TerraformCommand.PLAN, options) - return self.run_command("terraform plan", **options) - - def terraform_output( - self, - json_format: Type[FlaggedOption] = FlaggedOption, - variable: Optional[str] = None, - **kwargs: Dict[str, Any], - ) -> RunCommandResult: - options: Dict[str, Any] = kwargs.copy() - options["json"] = json_format - options["input"] = None - options = self.utils.get_default_options(TerraformCommand.OUTPUT, options) - options_list = (variable,) - return self.run_command("terraform output", *options_list, **options) - - def run_command( - self, - command: str, - *args: Optional[str], - **kwargs: Dict[str, Any], - ) -> RunCommandResult: - """ - Executes Terraform CLIs apply/destroy/init/plan - - Set `dry_run` flag in `kwargs` to test the function. - `dry_run` returns the command that will be run in shell for a given input - """ - options: Dict[str, Any] = kwargs.copy() - dry_run: bool = options.get("dry_run", False) - - capture_output: bool = options.get("capture_output", True) - - if capture_output: - stderr = PIPE - stdout = PIPE - else: - stderr = sys.stderr - stdout = sys.stdout - - command_list = self.utils.get_command_list(command, *args, **options) - command_str = " ".join(command_list) - self.log.info(f"Command: {command_str}") - out, err = None, None - - if dry_run: - # Adding dryrun flag for tests. It returns the command that will be executed for given input - self.log.info("Dry run: will not actually execute command") - self.log.info(f"Running command: {command_str}") - return RunCommandResult( - return_code=0, output=f"Dry run command: {command_str}", error="" - ) - - with Popen( - command_list, stdout=stdout, stderr=stderr, cwd=self.working_directory - ) as p: - out, err = p.communicate() - ret_code = p.returncode - self.log.info(f"output: {out}") - - if capture_output: - out = out.decode() - err = err.decode() - else: - out = None - err = None - - return RunCommandResult(return_code=ret_code, output=out, error=err) diff --git a/fbpcs/infra/pce_deployment_library/deploy_library/terraform_library/terraform_deployment_utils.py b/fbpcs/infra/pce_deployment_library/deploy_library/terraform_library/terraform_deployment_utils.py deleted file mode 100644 index 8ac3ae1ce..000000000 --- a/fbpcs/infra/pce_deployment_library/deploy_library/terraform_library/terraform_deployment_utils.py +++ /dev/null @@ -1,189 +0,0 @@ -# Copyright (c) Meta Platforms, Inc. and affiliates. -# -# This source code is licensed under the MIT license found in the -# LICENSE file in the root directory of this source tree. - -# pyre-strict - -from typing import Any, Dict, List, Optional - -from fbpcs.infra.pce_deployment_library.deploy_library.models import ( - FlaggedOption, - NOT_SUPPORTED_INIT_DEFAULT_OPTIONS, - TerraformCliOptions, - TerraformOptionFlag, -) - - -class TerraformDeploymentUtils: - - TERRAFORM_DEFAULT_PARALLELISM = 10 - - def __init__( - self, - state_file_path: Optional[str] = None, - terraform_variables: Optional[Dict[str, str]] = None, - parallelism: Optional[int] = None, - resource_targets: Optional[List[str]] = None, - var_definition_file: Optional[str] = None, - ) -> None: - """ - Args: - state_file_path: Path to store terraform state files - More information about terraform state: https://www.terraform.io/language/state - variables: -var option in terraform CLI. This arguments provides default variables. - These variables can be overwritten by commands also. - More information on terraform vairables: https://www.terraform.io/language/values/variables - parallelism: -parallelism=n option in Terraform CLI. - Limits the number of concurrent operation as Terraform walks the graph - More information on terraform parallelism: https://www.terraform.io/cli/commands/apply#parallelism-n - resource_targets: -target option in Terraform CLI. Used to target specific resource in terraform apply/destroy - More information on terraform targets: https://learn.hashicorp.com/tutorials/terraform/resource-targeting - var_definition_file: -var-file option in Terraform CLI. Used to define terraform variables in bulk though .tfvars file - More information on var_definition_file :https://www.terraform.io/language/values/variables#variable-definitions-tfvars-files - """ - self.state_file_path = state_file_path - self.resource_targets: Optional[List[str]] = ( - [] if resource_targets is None else resource_targets - ) - self.terraform_variables: Optional[Dict[str, str]] = ( - {} if terraform_variables is None else terraform_variables - ) - self.parallelism = parallelism - self.var_definition_file = var_definition_file - - """ - The -input=false option indicates that Terraform should not attempt to prompt for input, - and instead expect all necessary values to be provided by either configuration files or the command line. - https://learn.hashicorp.com/tutorials/terraform/automate-terraform - """ - self.input = False - - def get_command_list( - self, command: str, *args: Any, **kwargs: Dict[str, Any] - ) -> List[str]: - """ - Converts command string to list and updates commands with terraform options provided through kwargs and args. - """ - - commands_list = command.split() - - type_to_func_dict = { - dict: self.add_dict_options, - list: self.add_list_options, - bool: self.add_bool_options, - type(TerraformOptionFlag): self.add_flagged_option, - } - - for key, value in kwargs.items(): - # terraform CLI accepts options with "-" using "_" will results in error - key = key.replace("_", "-") - - func = type_to_func_dict.get(type(value), self.add_other_options) - - # pyre-fixme - commands_list.extend(func(key, value)) - - # Add args to commands list - # Check if args is None - commands_list.extend([x for x in args if x is not None]) - return commands_list - - def get_default_options( - self, terraform_command: str, input_options: Dict[str, Any] - ) -> Dict[str, Any]: - """ - Returns the terraform configs needed to create terraform cli - """ - - return_dict: Dict[str, Any] = { - TerraformCliOptions.state: self.state_file_path, - TerraformCliOptions.target: self.resource_targets, - TerraformCliOptions.var: self.terraform_variables, - TerraformCliOptions.var_file: self.var_definition_file, - TerraformCliOptions.parallelism: self.parallelism, - TerraformCliOptions.terraform_input: self.input, - **input_options, - } - - if terraform_command == "init": - for default_option in NOT_SUPPORTED_INIT_DEFAULT_OPTIONS: - return_dict.pop(default_option, None) - - return return_dict - - def add_dict_options(self, key: str, value: Dict[str, Any]) -> List[str]: - """ - Adds dict options in Terraform CLI: - Eg: t = TerraformDeploymentUtils() - options = {"backend_config": {"region": "us-west-2", "access_key":"fake_access_key"}} - t.get_command_list("terraform apply") - - Returns: - => ['terraform', 'apply', '-backend-config', 'region=us-west-2', '-backend-config', 'access_key=fake_access_key'] - """ - commands_list: List[str] = [] - for k, v in value.items(): - commands_list.append(f"-{key}") - commands_list.append(f"{k}={v}") - return commands_list - - def add_list_options(self, key: str, value: List[str]) -> List[str]: - """ - Adds list options in Terraform CLI: - Eg: t = TerraformDeploymentUtils() - options = {"target": ["aws_s3_bucket_object.objects[2]", "aws_s3_bucket_object.objects[3]"]} - t.get_command_list("terraform apply") - - Returns: - => ['terraform', 'apply', '-target="aws_s3_bucket_object.objects[2]"', '-target="aws_s3_bucket_object.objects[3]"'] - """ - commands_list: List[str] = [] - for val in value: - commands_list.append(f'-{key}="{val}"') - return commands_list - - def add_bool_options(self, key: str, value: bool) -> List[str]: - """ - Adds bool options in Terraform CLI: - Eg: t = TerraformDeploymentUtils() - options = {"input": False} - t.get_command_list("terraform apply") - - Returns: - => ['terraform', 'apply', '-input false'] - """ - commands_list: List[str] = [] - ret_value: str = str(value).lower() - commands_list.append(f"-{key}={ret_value}") - return commands_list - - def add_flagged_option(self, key: str, value: TerraformOptionFlag) -> List[str]: - """ - Adds flag options in Terraform CLI: - Eg: t = TerraformDeploymentUtils() - options = {"reconfigure": FlaggedOption} - t.get_command_list("terraform init") - - Returns: - => ['terraform', 'init', '-reconfigure'] - """ - commands_list: List[str] = [] - if value == FlaggedOption: - commands_list.append(f"-{key}") - return commands_list - - def add_other_options(self, key: str, value: str) -> List[str]: - """ - Adds default options in Terraform CLI: - Eg: t = TerraformDeploymentUtils() - options = {"target": "aws_s3_bucket_object.objects[2]"} - t.get_command_list("terraform init") - - Returns: - => ['terraform', 'init', '-target="aws_s3_bucket_object.objects[2]"'] - """ - commands_list: List[str] = [] - if value is not None: - commands_list.append(f'-{key}="{value}"') - return commands_list