import subprocess import os import json import logging from python_terraform.tfstate import Tfstate log = logging.getLogger(__name__) class IsFlagged: pass class IsNotFlagged: pass class Terraform: """ 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): """ :param working_dir: the folder of the working folder, if not given, will be where python :param targets: list of target :param state: path of state file relative to working folder :param variables: variables for apply/destroy/plan command :param parallelism: parallelism for apply/destroy command :param var_file: if specified, variables will not be used :param terraform_bin_path: binary path of terraform """ 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 # store the tfstate data self.tfstate = dict() def apply(self, dir=None, is_no_color=True, is_input=False, **kwargs): """ refer to https://terraform.io/docs/commands/apply.html :raise RuntimeError when return code is not zero :param is_no_color: if True, add flag -no-color :param is_input: if True, add option -input=true :param dir: folder relative to working folder :param kwargs: same as kwags in method 'cmd' :returns return_code, stdout, stderr """ args, option_dict = self._create_cmd_args(is_input, is_no_color, dir, kwargs) return self.cmd('apply', *args, **option_dict) def _create_cmd_args(self, is_input, is_no_color, dir, kwargs): 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 if is_no_color: option_dict['no_color'] = IsFlagged option_dict['input'] = is_input option_dict.update(kwargs) args = [dir] if dir else [] return args, option_dict def destroy(self, working_dir=None, is_force=True, is_no_color=True, is_input=False, **kwargs): """ refer to https://www.terraform.io/docs/commands/destroy.html :raise RuntimeError when return code is not zero :return: ret_code, stdout, stderr """ args, option_dict = self._create_cmd_args(is_input, is_no_color, working_dir, kwargs) if is_force: option_dict['force'] = IsFlagged return self.cmd('destroy', *args, **option_dict) 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 for k, v in kwargs.items(): if '_' in k: k = k.replace('_', '-') if type(v) is list: for sub_v in v: cmds += ['-{k}={v}'.format(k=k, v=sub_v)] continue if type(v) is dict: for sub_k, sub_v in v.items(): cmds += ["-{k}='{var_k}={var_v}'".format(k=k, var_k=sub_k, var_v=sub_v)] continue # simple flag, if v is IsFlagged: cmds += ['-{k}'.format(k=k)] continue if v is IsNotFlagged: continue if not v: continue if type(v) is bool: v = 'true' if v else 'false' cmds += ['-{k}={v}'.format(k=k, v=v)] cmds += args cmd = ' '.join(cmds) return cmd 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 :return: ret_code, out, err """ cmd_string = self.generate_cmd_string(cmd, *args, **kwargs) log.debug('command: {c}'.format(c=cmd_string)) working_folder = self.working_dir if self.working_dir else None p = subprocess.Popen(cmd_string, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, cwd=working_folder) out, err = p.communicate() ret_code = p.returncode log.debug('output: {o}'.format(o=out)) if ret_code == 0: self.read_state_file() else: log.warn('error: {e}'.format(e=err)) return ret_code, out.decode('utf-8'), err.decode('utf-8') def output(self, name): """ https://www.terraform.io/docs/commands/output.html :param name: name of output :return: output value """ ret, out, err = self.cmd('output', name, json=IsFlagged) log.debug('output raw string: {0}'.format(out)) if ret != 0: return None out = out.lstrip() output_dict = json.loads(out) return output_dict['value'] 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 """ if not file_path: file_path = self.state if not file_path: file_path = 'terraform.tfstate' if self.working_dir: file_path = os.path.join(self.working_dir, file_path) self.tfstate = Tfstate.load_file(file_path)