meissa/dda-python-terraform
6
3
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Migrate to Python 3 only

Browse source
This commit is contained in:
aubustou 2020-10-19 23:57:22 +02:00
parent a33aebce1c
commit fc584c2a48
3 changed files with 140 additions and 167 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
.travis.yml
View file

@ -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

260
python_terraform/__init__.py
View file

@ -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

18
python_terraform/tfstate.py
View file

@ -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()
return Tfstate()