From bcd523ff67385c7bb24d6273699993aa58f1b4de Mon Sep 17 00:00:00 2001 From: Francois Lebreau Date: Thu, 21 Jan 2021 01:01:34 +0000 Subject: [PATCH] Add typing --- .gitignore | 1 + .pre-commit-config.yaml | 1 - .travis.yml | 1 + python_terraform/__init__.py | 454 +------------------------------- python_terraform/terraform.py | 476 ++++++++++++++++++++++++++++++++++ python_terraform/tfstate.py | 7 +- 6 files changed, 484 insertions(+), 456 deletions(-) create mode 100644 python_terraform/terraform.py diff --git a/.gitignore b/.gitignore index a99ba67..b62d9be 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ Icon # virtualenv .virtualenv/ +venv/ # Intellij diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 8dbbe15..4c7fa54 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,6 +1,5 @@ default_language_version: python: python3.6 -fail_fast: true repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v3.1.0 # v2.1.0 diff --git a/.travis.yml b/.travis.yml index 39c16d6..7ab948e 100644 --- a/.travis.yml +++ b/.travis.yml @@ -3,6 +3,7 @@ python: - '3.6' - '3.7' - '3.8' +- '3.9' before_install: sudo apt-get install unzip before_script: - export TFVER=0.13.4 diff --git a/python_terraform/__init__.py b/python_terraform/__init__.py index 664ed8f..c9852e5 100644 --- a/python_terraform/__init__.py +++ b/python_terraform/__init__.py @@ -1,452 +1,2 @@ -import json -import logging -import os -import subprocess -import sys -import tempfile - -from python_terraform.tfstate import Tfstate - -logger = logging.getLogger(__name__) - -COMMAND_WITH_SUBCOMMANDS = {"workspace"} - - -class IsFlagged: - pass - - -class IsNotFlagged: - pass - - -class TerraformCommandError(subprocess.CalledProcessError): - def __init__(self, ret_code, cmd, out, err): - super(TerraformCommandError, self).__init__(ret_code, cmd) - self.out = out - self.err = err - - -class Terraform(object): - """Wrapper of terraform command line tool. - - https://www.terraform.io/ - """ - - def __init__( - self, - working_dir=None, - targets=None, - state=None, - variables=None, - parallelism=None, - var_file=None, - terraform_bin_path=None, - is_env_vars_included=True, - ): - """ - :param working_dir: the folder of the working folder, if not given, - will be current working folder - :param targets: list of target - as default value of apply/destroy/plan command - :param state: path of state file relative to working folder, - as a default value of apply/destroy/plan command - :param variables: default variables for apply/destroy/plan command, - will be override by variable passing by apply/destroy/plan method - :param parallelism: default parallelism value for apply/destroy command - :param var_file: passed as value of -var-file option, - could be string or list, list stands for multiple -var-file option - :param terraform_bin_path: binary path of terraform - :type is_env_vars_included: bool - :param is_env_vars_included: included env variables when calling terraform cmd - """ - self.is_env_vars_included = is_env_vars_included - self.working_dir = working_dir - self.state = state - self.targets = [] if targets is None else targets - self.variables = dict() if variables is None else variables - self.parallelism = parallelism - self.terraform_bin_path = ( - terraform_bin_path if terraform_bin_path else "terraform" - ) - self.var_file = var_file - self.temp_var_files = VariableFiles() - - # store the tfstate data - self.tfstate = None - self.read_state_file(self.state) - - def __getattr__(self, item): - def wrapper(*args, **kwargs): - cmd_name = str(item) - if cmd_name.endswith("_cmd"): - cmd_name = cmd_name[:-4] - logger.debug("called with %r and %r", args, kwargs) - return self.cmd(cmd_name, *args, **kwargs) - - return wrapper - - def apply( - self, - dir_or_plan=None, - input=False, - skip_plan=True, - no_color=IsFlagged, - **kwargs, - ): - """Refer to https://terraform.io/docs/commands/apply.html - - no-color is flagged by default - :param no_color: disable color of stdout - :param input: disable prompt for a missing variable - :param dir_or_plan: folder relative to working folder - :param skip_plan: force apply without plan (default: false) - :param kwargs: same as kwags in method 'cmd' - :returns return_code, stdout, stderr - """ - if not skip_plan: - return self.plan(dir_or_plan=dir_or_plan, **kwargs) - default = kwargs - default["input"] = input - default["no_color"] = no_color - default["auto-approve"] = True - option_dict = self._generate_default_options(default) - args = self._generate_default_args(dir_or_plan) - return self.cmd("apply", *args, **option_dict) - - def _generate_default_args(self, dir_or_plan): - return [dir_or_plan] if dir_or_plan else [] - - def _generate_default_options(self, input_options): - option_dict = dict() - option_dict["state"] = self.state - option_dict["target"] = self.targets - option_dict["var"] = self.variables - option_dict["var_file"] = self.var_file - option_dict["parallelism"] = self.parallelism - option_dict["no_color"] = IsFlagged - option_dict["input"] = False - option_dict.update(input_options) - return option_dict - - def destroy(self, dir_or_plan=None, force=IsFlagged, **kwargs): - """Refer to https://www.terraform.io/docs/commands/destroy.html - - force/no-color option is flagged by default - :return: ret_code, stdout, stderr - """ - default = kwargs - default["force"] = force - options = self._generate_default_options(default) - args = self._generate_default_args(dir_or_plan) - return self.cmd("destroy", *args, **options) - - def plan(self, dir_or_plan=None, detailed_exitcode=IsFlagged, **kwargs): - """Refer to https://www.terraform.io/docs/commands/plan.html - - :param detailed_exitcode: Return a detailed exit code when the command exits. - :param dir_or_plan: relative path to plan/folder - :param kwargs: options - :return: ret_code, stdout, stderr - """ - options = kwargs - options["detailed_exitcode"] = detailed_exitcode - options = self._generate_default_options(options) - args = self._generate_default_args(dir_or_plan) - return self.cmd("plan", *args, **options) - - def init( - self, - dir_or_plan=None, - backend_config=None, - reconfigure=IsFlagged, - backend=True, - **kwargs, - ): - """Refer to https://www.terraform.io/docs/commands/init.html - - By default, this assumes you want to use backend config, and tries to - init fresh. The flags -reconfigure and -backend=true are default. - - :param dir_or_plan: relative path to the folder want to init - :param backend_config: a dictionary of backend config options. eg. - t = Terraform() - t.init(backend_config={'access_key': 'myaccesskey', - 'secret_key': 'mysecretkey', 'bucket': 'mybucketname'}) - :param reconfigure: whether or not to force reconfiguration of backend - :param backend: whether or not to use backend settings for init - :param kwargs: options - :return: ret_code, stdout, stderr - """ - options = kwargs - options["backend_config"] = backend_config - options["reconfigure"] = reconfigure - options["backend"] = backend - options = self._generate_default_options(options) - args = self._generate_default_args(dir_or_plan) - return self.cmd("init", *args, **options) - - def generate_cmd_string(self, cmd, *args, **kwargs): - """For any generate_cmd_string doesn't written as public method of Terraform - - examples: - 1. call import command, - ref to https://www.terraform.io/docs/commands/import.html - --> generate_cmd_string call: - terraform import -input=true aws_instance.foo i-abcd1234 - --> python call: - tf.generate_cmd_string('import', 'aws_instance.foo', 'i-abcd1234', input=True) - - 2. call apply command, - --> generate_cmd_string call: - terraform apply -var='a=b' -var='c=d' -no-color the_folder - --> python call: - tf.generate_cmd_string('apply', the_folder, no_color=IsFlagged, var={'a':'b', 'c':'d'}) - - :param cmd: command and sub-command of terraform, seperated with space - refer to https://www.terraform.io/docs/commands/index.html - :param args: arguments of a command - :param kwargs: same as kwags in method 'cmd' - :return: string of valid terraform command - """ - cmds = cmd.split() - cmds = [self.terraform_bin_path] + cmds - if cmd in COMMAND_WITH_SUBCOMMANDS: - args = list(args) - subcommand = args.pop(0) - cmds.append(subcommand) - - for option, value in kwargs.items(): - if "_" in option: - option = option.replace("_", "-") - - if type(value) is list: - for sub_v in value: - cmds += [f"-{option}={sub_v}"] - continue - - if type(value) is dict: - if "backend-config" in option: - for bk, bv in value.items(): - cmds += [f"-backend-config={bk}={bv}"] - continue - - # since map type sent in string won't work, create temp var file for - # variables, and clean it up later - elif option == "var": - # We do not create empty var-files if there is no var passed. - # An empty var-file would result in an error: An argument or block definition is required here - if value: - filename = self.temp_var_files.create(value) - cmds += [f"-var-file={filename}"] - - continue - - # simple flag, - if value is IsFlagged: - cmds += ["-{k}".format(k=option)] - continue - - if value is None or value is IsNotFlagged: - continue - - if type(value) is bool: - value = "true" if value else "false" - - cmds += [f"-{option}={value}"] - - cmds += args - return cmds - - def cmd(self, cmd, *args, **kwargs): - """Run a terraform command, if success, will try to read state file - - :param cmd: command and sub-command of terraform, seperated with space - refer to https://www.terraform.io/docs/commands/index.html - :param args: arguments of a command - :param kwargs: any option flag with key value without prefixed dash character - if there's a dash in the option name, use under line instead of dash, - ex. -no-color --> no_color - if it's a simple flag with no value, value should be IsFlagged - ex. cmd('taint', allow＿missing=IsFlagged) - if it's a boolean value flag, assign True or false - if it's a flag could be used multiple times, assign list to it's value - if it's a "var" variable flag, assign dictionary to it - if a value is None, will skip this option - if the option 'capture_output' is passed (with any value other than - True), terraform output will be printed to stdout/stderr and - "None" will be returned as out and err. - if the option 'raise_on_error' is passed (with any value that evaluates to True), - and the terraform command returns a nonzerop return code, then - a TerraformCommandError exception will be raised. The exception object will - have the following properties: - returncode: The command's return code - out: The captured stdout, or None if not captured - err: The captured stderr, or None if not captured - :return: ret_code, out, err - """ - capture_output = kwargs.pop("capture_output", True) - raise_on_error = kwargs.pop("raise_on_error", False) - synchronous = kwargs.pop("synchronous", True) - - if capture_output is True: - stderr = subprocess.PIPE - stdout = subprocess.PIPE - elif capture_output == "framework": - stderr = None - stdout = None - else: - stderr = sys.stderr - stdout = sys.stdout - - cmds = self.generate_cmd_string(cmd, *args, **kwargs) - logger.debug("Command: %s", " ".join(cmds)) - - working_folder = self.working_dir if self.working_dir else None - - environ_vars = {} - if self.is_env_vars_included: - environ_vars = os.environ.copy() - - p = subprocess.Popen( - cmds, stdout=stdout, stderr=stderr, cwd=working_folder, env=environ_vars - ) - - if not synchronous: - return p, None, None - - out, err = p.communicate() - ret_code = p.returncode - logger.debug("output: %s", out) - - if ret_code == 0: - self.read_state_file() - else: - logger.warning("error: %s", err) - - self.temp_var_files.clean_up() - if capture_output is True: - out = out.decode() - err = err.decode() - else: - out = None - err = None - - if ret_code and raise_on_error: - raise TerraformCommandError(ret_code, " ".join(cmds), out=out, err=err) - - return ret_code, out, err - - def output(self, *args, **kwargs): - """Refer https://www.terraform.io/docs/commands/output.html - - Note that this method does not conform to the (ret_code, out, err) return - convention. To use the "output" command with the standard convention, - call "output_cmd" instead of "output". - - :param args: Positional arguments. There is one optional positional - argument NAME; if supplied, the returned output text - will be the json for a single named output value. - :param kwargs: Named options, passed to the command. In addition, - 'full_value': If True, and NAME is provided, then - the return value will be a dict with - "value', 'type', and 'sensitive' - properties. - :return: None, if an error occured - Output value as a string, if NAME is provided and full_value - is False or not provided - Output value as a dict with 'value', 'sensitive', and 'type' if - NAME is provided and full_value is True. - dict of named dicts each with 'value', 'sensitive', and 'type', - if NAME is not provided - """ - kwargs["json"] = IsFlagged - if not kwargs.get("capture_output", True) is True: - raise ValueError("capture_output is required for this method") - - ret, out, err = self.output_cmd(*args, **kwargs) - - if ret: - return None - - return json.loads(out.lstrip()) - - def read_state_file(self, file_path=None): - """Read .tfstate file - - :param file_path: relative path to working dir - :return: states file in dict type - """ - - working_dir = self.working_dir or "" - - file_path = file_path or self.state or "" - - if not file_path: - backend_path = os.path.join(file_path, ".terraform", "terraform.tfstate") - - if os.path.exists(os.path.join(working_dir, backend_path)): - file_path = backend_path - else: - file_path = os.path.join(file_path, "terraform.tfstate") - - file_path = os.path.join(working_dir, file_path) - - self.tfstate = Tfstate.load_file(file_path) - - def set_workspace(self, workspace, *args, **kwargs): - """Set workspace - - :param workspace: the desired workspace. - :return: status - """ - return self.cmd("workspace", "select", workspace, *args, **kwargs) - - def create_workspace(self, workspace, *args, **kwargs): - """Create workspace - - :param workspace: the desired workspace. - :return: status - """ - return self.cmd("workspace", "new", workspace, *args, **kwargs) - - def delete_workspace(self, workspace, *args, **kwargs): - """Delete workspace - - :param workspace: the desired workspace. - :return: status - """ - return self.cmd("workspace", "delete", workspace, *args, **kwargs) - - def show_workspace(self, **kwargs): - """Show workspace, this command does not need the [DIR] part - - :return: workspace - """ - return self.cmd("workspace", "show", **kwargs) - - def __exit__(self, exc_type, exc_value, traceback): - self.temp_var_files.clean_up() - - -class VariableFiles(object): - def __init__(self): - self.files = [] - - def create(self, variables): - with tempfile.NamedTemporaryFile( - "w+t", suffix=".tfvars.json", delete=False - ) as temp: - logger.debug("%s is created", temp.name) - self.files.append(temp) - logger.debug("variables wrote to tempfile: %s", variables) - temp.write(json.dumps(variables)) - file_name = temp.name - - return file_name - - def clean_up(self): - for f in self.files: - os.unlink(f.name) - - self.files = [] +from .terraform import Terraform, VariableFiles +from .tfstate import Tfstate diff --git a/python_terraform/terraform.py b/python_terraform/terraform.py new file mode 100644 index 0000000..cd4b712 --- /dev/null +++ b/python_terraform/terraform.py @@ -0,0 +1,476 @@ +import json +import logging +import os +import subprocess +import sys +import tempfile +from typing import Callable, Dict, List, Optional, Sequence, Tuple, Type, TypeVar, Union + +from python_terraform.tfstate import Tfstate + +logger = logging.getLogger(__name__) + +COMMAND_WITH_SUBCOMMANDS = {"workspace"} + + +class TerraformFlag: + pass + + +class IsFlagged(TerraformFlag): + pass + + +class IsNotFlagged(TerraformFlag): + pass + + +CommandOutput = Tuple[Optional[int], Optional[str], Optional[str]] + + +class TerraformCommandError(subprocess.CalledProcessError): + def __init__(self, ret_code: int, cmd: str, out: Optional[str], err: Optional[str]): + super(TerraformCommandError, self).__init__(ret_code, cmd) + self.out = out + self.err = err + + +class Terraform: + """Wrapper of terraform command line tool. + + https://www.terraform.io/ + """ + + def __init__( + self, + working_dir: Optional[str] = None, + targets: Optional[Sequence[str]] = None, + state: Optional[str] = None, + variables: Optional[Sequence[str]] = None, + parallelism: Optional[str] = None, + var_file: Optional[str] = None, + terraform_bin_path: Optional[str] = None, + is_env_vars_included: bool = True, + ): + """ + :param working_dir: the folder of the working folder, if not given, + will be current working folder + :param targets: list of target + as default value of apply/destroy/plan command + :param state: path of state file relative to working folder, + as a default value of apply/destroy/plan command + :param variables: default variables for apply/destroy/plan command, + will be override by variable passing by apply/destroy/plan method + :param parallelism: default parallelism value for apply/destroy command + :param var_file: passed as value of -var-file option, + could be string or list, list stands for multiple -var-file option + :param terraform_bin_path: binary path of terraform + :type is_env_vars_included: bool + :param is_env_vars_included: included env variables when calling terraform cmd + """ + self.is_env_vars_included = is_env_vars_included + self.working_dir = working_dir + self.state = state + self.targets = [] if targets is None else targets + self.variables = dict() if variables is None else variables + self.parallelism = parallelism + self.terraform_bin_path = ( + terraform_bin_path if terraform_bin_path else "terraform" + ) + self.var_file = var_file + self.temp_var_files = VariableFiles() + + # store the tfstate data + self.tfstate = None + self.read_state_file(self.state) + + def __getattr__(self, item: str) -> Callable: + def wrapper(*args, **kwargs): + cmd_name = str(item) + if cmd_name.endswith("_cmd"): + cmd_name = cmd_name[:-4] + logger.debug("called with %r and %r", args, kwargs) + return self.cmd(cmd_name, *args, **kwargs) + + return wrapper + + def apply( + self, + dir_or_plan: Optional[str] = None, + input: bool = False, + skip_plan: bool = True, + no_color: Type[TerraformFlag] = IsFlagged, + **kwargs, + ) -> CommandOutput: + """Refer to https://terraform.io/docs/commands/apply.html + + no-color is flagged by default + :param no_color: disable color of stdout + :param input: disable prompt for a missing variable + :param dir_or_plan: folder relative to working folder + :param skip_plan: force apply without plan (default: false) + :param kwargs: same as kwags in method 'cmd' + :returns return_code, stdout, stderr + """ + if not skip_plan: + return self.plan(dir_or_plan=dir_or_plan, **kwargs) + default = kwargs + default["input"] = input + default["no_color"] = no_color + default["auto-approve"] = True + option_dict = self._generate_default_options(default) + args = self._generate_default_args(dir_or_plan) + return self.cmd("apply", *args, **option_dict) + + def _generate_default_args(self, dir_or_plan) -> Sequence[str]: + return [dir_or_plan] if dir_or_plan else [] + + def _generate_default_options(self, input_options): + option_dict = dict() + option_dict["state"] = self.state + option_dict["target"] = self.targets + option_dict["var"] = self.variables + option_dict["var_file"] = self.var_file + option_dict["parallelism"] = self.parallelism + option_dict["no_color"] = IsFlagged + option_dict["input"] = False + option_dict.update(input_options) + return option_dict + + def destroy( + self, + dir_or_plan: Optional[str] = None, + force: Type[TerraformFlag] = IsFlagged, + **kwargs, + ) -> CommandOutput: + """Refer to https://www.terraform.io/docs/commands/destroy.html + + force/no-color option is flagged by default + :return: ret_code, stdout, stderr + """ + default = kwargs + default["force"] = force + options = self._generate_default_options(default) + args = self._generate_default_args(dir_or_plan) + return self.cmd("destroy", *args, **options) + + def plan( + self, + dir_or_plan: Optional[str] = None, + detailed_exitcode: Type[TerraformFlag] = IsFlagged, + **kwargs, + ) -> CommandOutput: + """Refer to https://www.terraform.io/docs/commands/plan.html + + :param detailed_exitcode: Return a detailed exit code when the command exits. + :param dir_or_plan: relative path to plan/folder + :param kwargs: options + :return: ret_code, stdout, stderr + """ + options = kwargs + options["detailed_exitcode"] = detailed_exitcode + options = self._generate_default_options(options) + args = self._generate_default_args(dir_or_plan) + return self.cmd("plan", *args, **options) + + def init( + self, + dir_or_plan: Optional[str] = None, + backend_config: Optional[Dict[str, str]] = None, + reconfigure: Type[TerraformFlag] = IsFlagged, + backend: bool = True, + **kwargs, + ) -> CommandOutput: + """Refer to https://www.terraform.io/docs/commands/init.html + + By default, this assumes you want to use backend config, and tries to + init fresh. The flags -reconfigure and -backend=true are default. + + :param dir_or_plan: relative path to the folder want to init + :param backend_config: a dictionary of backend config options. eg. + t = Terraform() + t.init(backend_config={'access_key': 'myaccesskey', + 'secret_key': 'mysecretkey', 'bucket': 'mybucketname'}) + :param reconfigure: whether or not to force reconfiguration of backend + :param backend: whether or not to use backend settings for init + :param kwargs: options + :return: ret_code, stdout, stderr + """ + options = kwargs + options["backend_config"] = backend_config + options["reconfigure"] = reconfigure + options["backend"] = backend + options = self._generate_default_options(options) + args = self._generate_default_args(dir_or_plan) + return self.cmd("init", *args, **options) + + def generate_cmd_string(self, cmd: str, *args, **kwargs) -> List[str]: + """For any generate_cmd_string doesn't written as public method of Terraform + + examples: + 1. call import command, + ref to https://www.terraform.io/docs/commands/import.html + --> generate_cmd_string call: + terraform import -input=true aws_instance.foo i-abcd1234 + --> python call: + tf.generate_cmd_string('import', 'aws_instance.foo', 'i-abcd1234', input=True) + + 2. call apply command, + --> generate_cmd_string call: + terraform apply -var='a=b' -var='c=d' -no-color the_folder + --> python call: + tf.generate_cmd_string('apply', the_folder, no_color=IsFlagged, var={'a':'b', 'c':'d'}) + + :param cmd: command and sub-command of terraform, seperated with space + refer to https://www.terraform.io/docs/commands/index.html + :param args: arguments of a command + :param kwargs: same as kwags in method 'cmd' + :return: string of valid terraform command + """ + cmds = cmd.split() + cmds = [self.terraform_bin_path] + cmds + if cmd in COMMAND_WITH_SUBCOMMANDS: + args = list(args) + subcommand = args.pop(0) + cmds.append(subcommand) + + for option, value in kwargs.items(): + if "_" in option: + option = option.replace("_", "-") + + if isinstance(value, list): + for sub_v in value: + cmds += [f"-{option}={sub_v}"] + continue + + if isinstance(value, dict): + if "backend-config" in option: + for bk, bv in value.items(): + cmds += [f"-backend-config={bk}={bv}"] + continue + + # since map type sent in string won't work, create temp var file for + # variables, and clean it up later + elif option == "var": + # We do not create empty var-files if there is no var passed. + # An empty var-file would result in an error: An argument or block definition is required here + if value: + filename = self.temp_var_files.create(value) + cmds += [f"-var-file={filename}"] + + continue + + # simple flag, + if value is IsFlagged: + cmds += [f"-{option}"] + continue + + if value is None or value is IsNotFlagged: + continue + + if isinstance(value, bool): + value = "true" if value else "false" + + cmds += [f"-{option}={value}"] + + cmds += args + return cmds + + def cmd( + self, + cmd: str, + *args, + capture_output: Union[bool, str] = True, + raise_on_error: bool = False, + synchronous: bool = True, + **kwargs, + ) -> CommandOutput: + """Run a terraform command, if success, will try to read state file + + :param cmd: command and sub-command of terraform, seperated with space + refer to https://www.terraform.io/docs/commands/index.html + :param args: arguments of a command + :param kwargs: any option flag with key value without prefixed dash character + if there's a dash in the option name, use under line instead of dash, + ex. -no-color --> no_color + if it's a simple flag with no value, value should be IsFlagged + ex. cmd('taint', allow＿missing=IsFlagged) + if it's a boolean value flag, assign True or false + if it's a flag could be used multiple times, assign list to it's value + if it's a "var" variable flag, assign dictionary to it + if a value is None, will skip this option + if the option 'capture_output' is passed (with any value other than + True), terraform output will be printed to stdout/stderr and + "None" will be returned as out and err. + if the option 'raise_on_error' is passed (with any value that evaluates to True), + and the terraform command returns a nonzerop return code, then + a TerraformCommandError exception will be raised. The exception object will + have the following properties: + returncode: The command's return code + out: The captured stdout, or None if not captured + err: The captured stderr, or None if not captured + :return: ret_code, out, err + """ + if capture_output is True: + stderr = subprocess.PIPE + stdout = subprocess.PIPE + elif capture_output == "framework": + stderr = None + stdout = None + else: + stderr = sys.stderr + stdout = sys.stdout + + cmds = self.generate_cmd_string(cmd, *args, **kwargs) + logger.debug("Command: %s", " ".join(cmds)) + + working_folder = self.working_dir if self.working_dir else None + + environ_vars = {} + if self.is_env_vars_included: + environ_vars = os.environ.copy() + + p = subprocess.Popen( + cmds, stdout=stdout, stderr=stderr, cwd=working_folder, env=environ_vars + ) + + if not synchronous: + return None, None, None + + out, err = p.communicate() + ret_code = p.returncode + logger.debug("output: %s", out) + + if ret_code == 0: + self.read_state_file() + else: + logger.warning("error: %s", err) + + self.temp_var_files.clean_up() + if capture_output is True: + out = out.decode() + err = err.decode() + else: + out = None + err = None + + if ret_code and raise_on_error: + raise TerraformCommandError(ret_code, " ".join(cmds), out=out, err=err) + + return ret_code, out, err + + def output( + self, *args, capture_output: bool = True, **kwargs + ) -> Union[None, str, Dict[str, str], Dict[str, Dict[str, str]]]: + """Refer https://www.terraform.io/docs/commands/output.html + + Note that this method does not conform to the (ret_code, out, err) return + convention. To use the "output" command with the standard convention, + call "output_cmd" instead of "output". + + :param args: Positional arguments. There is one optional positional + argument NAME; if supplied, the returned output text + will be the json for a single named output value. + :param kwargs: Named options, passed to the command. In addition, + 'full_value': If True, and NAME is provided, then + the return value will be a dict with + "value', 'type', and 'sensitive' + properties. + :return: None, if an error occured + Output value as a string, if NAME is provided and full_value + is False or not provided + Output value as a dict with 'value', 'sensitive', and 'type' if + NAME is provided and full_value is True. + dict of named dicts each with 'value', 'sensitive', and 'type', + if NAME is not provided + """ + kwargs["json"] = IsFlagged + if capture_output is False: + raise ValueError("capture_output is required for this method") + + ret, out, _ = self.output_cmd(*args, **kwargs) + + if ret: + return None + + return json.loads(out.lstrip()) + + def read_state_file(self, file_path=None) -> None: + """Read .tfstate file + + :param file_path: relative path to working dir + :return: states file in dict type + """ + + working_dir = self.working_dir or "" + + file_path = file_path or self.state or "" + + if not file_path: + backend_path = os.path.join(file_path, ".terraform", "terraform.tfstate") + + if os.path.exists(os.path.join(working_dir, backend_path)): + file_path = backend_path + else: + file_path = os.path.join(file_path, "terraform.tfstate") + + file_path = os.path.join(working_dir, file_path) + + self.tfstate = Tfstate.load_file(file_path) + + def set_workspace(self, workspace, *args, **kwargs) -> CommandOutput: + """Set workspace + + :param workspace: the desired workspace. + :return: status + """ + return self.cmd("workspace", "select", workspace, *args, **kwargs) + + def create_workspace(self, workspace, *args, **kwargs) -> CommandOutput: + """Create workspace + + :param workspace: the desired workspace. + :return: status + """ + return self.cmd("workspace", "new", workspace, *args, **kwargs) + + def delete_workspace(self, workspace, *args, **kwargs) -> CommandOutput: + """Delete workspace + + :param workspace: the desired workspace. + :return: status + """ + return self.cmd("workspace", "delete", workspace, *args, **kwargs) + + def show_workspace(self, **kwargs) -> CommandOutput: + """Show workspace, this command does not need the [DIR] part + + :return: workspace + """ + return self.cmd("workspace", "show", **kwargs) + + def __exit__(self, exc_type, exc_value, traceback) -> None: + self.temp_var_files.clean_up() + + +class VariableFiles: + def __init__(self): + self.files = [] + + def create(self, variables: Dict[str, str]) -> str: + with tempfile.NamedTemporaryFile( + "w+t", suffix=".tfvars.json", delete=False + ) as temp: + logger.debug("%s is created", temp.name) + self.files.append(temp) + logger.debug("variables wrote to tempfile: %s", variables) + temp.write(json.dumps(variables)) + file_name = temp.name + + return file_name + + def clean_up(self): + for f in self.files: + os.unlink(f.name) + + self.files = [] diff --git a/python_terraform/tfstate.py b/python_terraform/tfstate.py index e16cf36..20061b9 100644 --- a/python_terraform/tfstate.py +++ b/python_terraform/tfstate.py @@ -1,19 +1,20 @@ import json import logging import os +from typing import Dict, Optional logger = logging.getLogger(__name__) class Tfstate: - def __init__(self, data=None): - self.tfstate_file = None + def __init__(self, data: Optional[Dict[str, str]] = None): + self.tfstate_file: Optional[str] = None self.native_data = data if data: self.__dict__ = data @staticmethod - def load_file(file_path): + def load_file(file_path: str) -> "Tfstate": """Read the tfstate file and load its contents. Parses then as JSON and put the result into the object.