From fc584c2a480816bda8c8c65f61f0803fc3c4bb68 Mon Sep 17 00:00:00 2001 From: aubustou Date: Mon, 19 Oct 2020 23:57:22 +0200 Subject: [PATCH] Migrate to Python 3 only --- .travis.yml | 29 +--- python_terraform/__init__.py | 260 +++++++++++++++++------------------ python_terraform/tfstate.py | 18 ++- 3 files changed, 140 insertions(+), 167 deletions(-) diff --git a/.travis.yml b/.travis.yml index bb62e9c..8e95f16 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,8 +1,9 @@ language: python python: -- '2.7' -- '3.5' - '3.6' +- '3.7' +- '3.8' +- '3.9' before_install: sudo apt-get install unzip before_script: - export TFVER=0.10.0 @@ -26,27 +27,3 @@ branches: - master - develop - release/** -deploy: -- provider: pypi - distributions: sdist - server: https://testpypi.python.org/pypi - user: beelit94 - password: - secure: sWxc+p/gdq3k2WbUGNG2F4TukFNkTkvq6OPaiOvyfgWThYNk6/juRkMd8flmTbh0VGhcjFbpDLeSApb2kFhfiokYJSH1hOOcmXf8xzYH8/+R4DDEiGa5Y/pR9TBvYu4S8eJEfFUFfb1BBpapykj7o43hcaqMExBJIdVJU7aeoEAC1jQeTJh8wWwdJKHy2dNSM+6RVhk3e5+b0LfK7Bk5sU5P+YdEMj79MJU450J4OmZXWzJgvBN5/2QfVa5LrUD00nYuGuiBniz2lVevIHWjUYawUzpPsTa7F0s2WemG9YcV7U8u06xNjY9Ce3CTbxNhc7OIKq+TCkOgR3qZFXVJ8A87G+AT2iQ01VslQ4DJCxnJNTnpqojWnwf6MFL9O8ONioWYO32bhQFKOQ806ASHP4lNMRDKqx8hXtP5In7/r0SARKscv6Bas83rp+FESkKD5vWgkZJG+yx96LlwRLUhSVnVyb/nOJ++zt5RR3BvY2O4p9YAZY3Qt8TQihOdBQKnY3UXsMyNaE25+yvyNWpmyJiePRbTUd+cpLnycnqG9Ll8v6TpFXb6ahFMjlAFfJNQYlREfseClTHSRjZNxfsXGQCsJh6TZAq7jOB5hCk3q41eOUFWARxbyj8j59NBV8fSQrrGJJ9/VZKQeYiQlBB9KpK4PrnH84oeQ8i+VSbVr5w= - on: - branch: release/** - tags: false - condition: $TRAVIS_PYTHON_VERSION = "3.5" -- provider: pypi - distributions: sdist - user: beelit94 - password: - secure: QhCiTLrBvw/Uzt3eiLEmvMP3uHnayVCETqEDA+2+Q9vFavqj0CHA76zqYonBFqnh0a3HFCRIVVt+6ynpZ10kpQ3tAObIw+pY39ZPnpAhOjSpFzzMdpIF9Bhv9A93ng2iSESAZPAOwktHzUwjFx0Zvl0lSYD9rutHgttGgdU2CajiUtwTUhCTjOAVdR2Gm+15H808vzKWnMaKflXxZt+fkt279mQTYAtz6eBWtZwIKry/uAJCSrPSWtbi50O0HsWRMXLXWH5Jn/BVjWSDSM92DssUDq0D+tQyp4M5nQXJ9EyAvEdsKNLx3cvNruznh2ohI2jmcoIjwFiS6+wrEmUiXkP86iyzCSqL/EbcOG0xUh3vbfYtMBp7jENgD405+3SEhPY4PlqUmc+HDtB7FUcHz4y7wGWJRGyQzNnjJ6Tv0Ajdz5mfJubWVIvHjcRqkxTVtUKt50o00xZ62M0ZzQkDTIHQEsZly0XeHAgSvNzWkmjt9BiBrZ9OkoWVkRpSrCBy/EcpDNPCTSfSzOQ0Nq1ePFjkkW1n8QWDW9Pdb+/7/P2y9E2S8CT+nXBkRQeQiO86Qf1Ireg7k9TA5VYisVZ6bEXEc9UV0mAojpSsC7zWhVlbAoltN6ZbjKmqy/wqn2QIcJemcSie0JigzKpdw7l8FPT2lCRyTKlYLpRyKXzSkNI= - on: - branch: master - tags: false - condition: $TRAVIS_PYTHON_VERSION = "3.5" -notifications: - email: - recipients: - - beelit94@gmail.com diff --git a/python_terraform/__init__.py b/python_terraform/__init__.py index 33ab114..a0533a7 100644 --- a/python_terraform/__init__.py +++ b/python_terraform/__init__.py @@ -1,6 +1,3 @@ -# -*- coding: utf-8 -*- -# above is for compatibility of python2.7.11 - import subprocess import os import sys @@ -11,17 +8,10 @@ import tempfile from python_terraform.tfstate import Tfstate -try: # Python 2.7+ - from logging import NullHandler -except ImportError: - class NullHandler(logging.Handler): - def emit(self, record): - pass +logger = logging.getLogger(__name__) -log = logging.getLogger(__name__) -log.addHandler(NullHandler()) +COMMAND_WITH_SUBCOMMANDS = {"workspace"} -COMMAND_WITH_SUBCOMMANDS = ['workspace'] class IsFlagged: pass @@ -32,27 +22,29 @@ class IsNotFlagged: 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 + 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 + """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, - ): + 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 @@ -75,8 +67,9 @@ class Terraform(object): 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.terraform_bin_path = ( + terraform_bin_path if terraform_bin_path else "terraform" + ) self.var_file = var_file self.temp_var_files = VariableFiles() @@ -87,17 +80,23 @@ class Terraform(object): def __getattr__(self, item): def wrapper(*args, **kwargs): cmd_name = str(item) - if cmd_name.endswith('_cmd'): + if cmd_name.endswith("_cmd"): cmd_name = cmd_name[:-4] - log.debug('called with %r and %r' % (args, kwargs)) + 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=False, no_color=IsFlagged, - **kwargs): - """ - refer to https://terraform.io/docs/commands/apply.html + def apply( + self, + dir_or_plan=None, + input=False, + skip_plan=False, + 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 @@ -107,58 +106,63 @@ class Terraform(object): :returns return_code, stdout, stderr """ default = kwargs - default['input'] = input - default['no_color'] = no_color - default['auto-approve'] = (skip_plan == True) + default["input"] = input + default["no_color"] = no_color + default["auto-approve"] = skip_plan is True option_dict = self._generate_default_options(default) args = self._generate_default_args(dir_or_plan) - return self.cmd('apply', *args, **option_dict) + 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["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 + """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 + default["force"] = force options = self._generate_default_options(default) args = self._generate_default_args(dir_or_plan) - return self.cmd('destroy', *args, **options) + 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 + """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["detailed_exitcode"] = detailed_exitcode options = self._generate_default_options(options) args = self._generate_default_args(dir_or_plan) - return self.cmd('plan', *args, **options) + 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 + 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. @@ -166,7 +170,7 @@ class Terraform(object): :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', + 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 @@ -174,16 +178,15 @@ class Terraform(object): :return: ret_code, stdout, stderr """ options = kwargs - options['backend_config'] = backend_config - options['reconfigure'] = reconfigure - options['backend'] = backend + 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) + 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 + """For any generate_cmd_string doesn't written as public method of Terraform examples: 1. call import command, @@ -213,50 +216,50 @@ class Terraform(object): cmds.append(subcommand) for option, value in kwargs.items(): - if '_' in option: - option = option.replace('_', '-') + if "_" in option: + option = option.replace("_", "-") if type(value) is list: for sub_v in value: - cmds += ['-{k}={v}'.format(k=option, v=sub_v)] + cmds += [f"-{option}={sub_v}"] continue if type(value) is dict: - if 'backend-config' in option: + if "backend-config" in option: for bk, bv in value.items(): - cmds += ['-backend-config={k}={v}'.format(k=bk, v=bv)] + 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': + 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 += ['-var-file={0}'.format(filename)] + cmds += [f"-var-file={filename}"] continue # simple flag, if value is IsFlagged: - cmds += ['-{k}'.format(k=option)] + 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' + value = "true" if value else "false" - cmds += ['-{k}={v}'.format(k=option, v=value)] + 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 + """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 @@ -281,10 +284,10 @@ class Terraform(object): 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) - + 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 @@ -296,7 +299,7 @@ class Terraform(object): stdout = sys.stdout cmds = self.generate_cmd_string(cmd, *args, **kwargs) - log.debug('command: {c}'.format(c=' '.join(cmds))) + logger.debug("Command: %s", " ".join(cmds)) working_folder = self.working_dir if self.working_dir else None @@ -304,47 +307,46 @@ class Terraform(object): 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) + 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 - log.debug('output: {o}'.format(o=out)) + logger.debug("output: %s", out) if ret_code == 0: self.read_state_file() else: - log.warning('error: {e}'.format(e=err)) + logger.warning("error: %s", err) self.temp_var_files.clean_up() if capture_output is True: - out = out.decode('utf-8') - err = err.decode('utf-8') + out = out.decode() + err = err.decode() else: out = None err = None - if ret_code != 0 and raise_on_error: - raise TerraformCommandError( - ret_code, ' '.join(cmds), out=out, err=err) + 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): - """ - https://www.terraform.io/docs/commands/output.html + """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". + 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, + :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' @@ -357,15 +359,15 @@ class Terraform(object): dict of named dicts each with 'value', 'sensitive', and 'type', if NAME is not provided """ - full_value = kwargs.pop('full_value', False) - name_provided = (len(args) > 0) - kwargs['json'] = IsFlagged - if not kwargs.get('capture_output', True) is True: - raise ValueError('capture_output is required for this method') + full_value = kwargs.pop("full_value", False) + name_provided = bool(len(args)) + 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 != 0: + if ret: return None out = out.lstrip() @@ -373,68 +375,63 @@ class Terraform(object): value = json.loads(out) if name_provided and not full_value: - value = value['value'] + value = value["value"] return value def read_state_file(self, file_path=None): - """ - read .tfstate file + """Read .tfstate file + :param file_path: relative path to working dir :return: states file in dict type """ - working_dir = self.working_dir or '' + working_dir = self.working_dir or "" - file_path = file_path or self.state or '' + file_path = file_path or self.state or "" if not file_path: - backend_path = os.path.join(file_path, '.terraform', - 'terraform.tfstate') + 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(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 + """Set workspace + :param workspace: the desired workspace. :return: status """ - - return self.cmd('workspace', 'select', workspace, *args, **kwargs) + return self.cmd("workspace", "select", workspace, *args, **kwargs) def create_workspace(self, workspace, *args, **kwargs): - """ - create workspace + """Create workspace + :param workspace: the desired workspace. :return: status """ - - return self.cmd('workspace', 'new', workspace, *args, **kwargs) + return self.cmd("workspace", "new", workspace, *args, **kwargs) def delete_workspace(self, workspace, *args, **kwargs): - """ - delete workspace + """Delete workspace + :param workspace: the desired workspace. :return: status """ - - return self.cmd('workspace', 'delete', workspace, *args, **kwargs) + return self.cmd("workspace", "delete", workspace, *args, **kwargs) def show_workspace(self, **kwargs): - """ - show workspace, this command does not need the [DIR] part + """Show workspace, this command does not need the [DIR] part + :return: workspace """ - - return self.cmd('workspace', 'show', **kwargs) + return self.cmd("workspace", "show", **kwargs) def __exit__(self, exc_type, exc_value, traceback): self.temp_var_files.clean_up() @@ -445,11 +442,12 @@ class VariableFiles(object): self.files = [] def create(self, variables): - with tempfile.NamedTemporaryFile('w+t', suffix='.tfvars.json', delete=False) as temp: - log.debug('{0} is created'.format(temp.name)) + with tempfile.NamedTemporaryFile( + "w+t", suffix=".tfvars.json", delete=False + ) as temp: + logger.debug("%s is created", temp.name) self.files.append(temp) - log.debug( - 'variables wrote to tempfile: {0}'.format(str(variables))) + logger.debug("variables wrote to tempfile: %s", variables) temp.write(json.dumps(variables)) file_name = temp.name diff --git a/python_terraform/tfstate.py b/python_terraform/tfstate.py index 497e757..2ecd64b 100644 --- a/python_terraform/tfstate.py +++ b/python_terraform/tfstate.py @@ -1,14 +1,11 @@ -# -*- coding: utf-8 -*- -# above is for compatibility of python2.7.11 - import json import os import logging -log = logging.getLogger(__name__) +logger = logging.getLogger(__name__) -class Tfstate(object): +class Tfstate: def __init__(self, data=None): self.tfstate_file = None self.native_data = data @@ -17,10 +14,11 @@ class Tfstate(object): @staticmethod def load_file(file_path): + """Read the tfstate file and load its contents. + + Parses then as JSON and put the result into the object. """ - Read the tfstate file and load its contents, parses then as JSON and put the result into the object - """ - log.debug('read data from {0}'.format(file_path)) + logger.debug('read data from %s', file_path) if os.path.exists(file_path): with open(file_path) as f: json_data = json.load(f) @@ -29,6 +27,6 @@ class Tfstate(object): tf_state.tfstate_file = file_path return tf_state - log.debug('{0} is not exist'.format(file_path)) + logger.debug('%s is not exist', file_path) - return Tfstate() \ No newline at end of file + return Tfstate()