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

Add typing

Browse source
This commit is contained in:
Francois Lebreau 2021-01-21 01:01:34 +00:00
parent b373f85311
commit bcd523ff67
6 changed files with 484 additions and 456 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -14,6 +14,7 @@ Icon
# virtualenv
.virtualenv/
venv/
# Intellij

1
.pre-commit-config.yaml
View file

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

1
.travis.yml
View file

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

454
python_terraform/__init__.py
View file

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

476
python_terraform/terraform.py Normal file
View file

@ -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 = []

7
python_terraform/tfstate.py
View file

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