@ -3,180 +3,234 @@ import os
import json
import logging
from python_terraform . tfstate import Tfstate
log = logging . getLogger ( __name__ )
class IsFlagged :
pass
class IsNotFlagged :
pass
class Terraform :
def __init__ ( self , targets = None , state = ' terraform.tfstate ' , variables = None ) :
"""
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 . state_filename = state
self . state_data = dict ( )
self . parallelism = 50
def apply ( self , targets = None , variables = None , * * kargs ) :
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
: param variables : variables in dict type
: param targets : targets in list
: 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
"""
variables = self . variables if variables is None else variables
targets = self . targets if targets is None else targets
parameters = [ ]
parameters + = self . _generate_targets ( targets )
parameters + = self . _generate_var_string ( variables )
parameters + = self . _gen_param_string ( kargs )
parameters = \
[ ' terraform ' , ' apply ' , ' -state= %s ' % self . state_filename ] + parameters
cmd = ' ' . join ( parameters )
return self . _run_cmd ( cmd )
def _gen_param_string ( self , kargs ) :
params = [ ]
for key , value in kargs . items ( ) :
if value :
params + = [ ' - %s = %s ' % ( key , value ) ]
else :
params + = [ ' - %s ' % key ]
return params
def _run_cmd ( self , cmd ) :
log . debug ( ' command: ' + cmd )
p = subprocess . Popen (
cmd , stdout = subprocess . PIPE , stderr = subprocess . PIPE , shell = True )
out , err = p . communicate ( )
ret_code = p . returncode
log . debug ( ' output: ' + out )
if ret_code == 0 :
log . debug ( ' error: ' + err )
self . read_state_file ( )
return ret_code , out , err
def destroy ( self , targets = None , variables = None , * * kwargs ) :
variables = self . variables if variables is None else variables
targets = self . targets if targets is None else targets
parameters = [ ]
parameters + = self . _generate_targets ( targets )
parameters + = self . _generate_var_string ( variables )
parameters = \
[ ' terraform ' , ' destroy ' , ' -force ' , ' -state= %s ' % self . state_filename ] + \
parameters
cmd = ' ' . join ( parameters )
return self . _run_cmd ( cmd )
def refresh ( self , targets = None , variables = None ) :
variables = self . variables if variables is None else variables
targets = self . targets if targets is None else targets
parameters = [ ]
parameters + = self . _generate_targets ( targets )
parameters + = self . _generate_var_string ( variables )
parameters = \
[ ' terraform ' , ' refresh ' , ' -state= %s ' % self . state_filename ] + \
parameters
cmd = ' ' . join ( parameters )
return self . _run_cmd ( cmd )
def read_state_file ( self ) :
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 ) :
"""
read . tfstate file
: return : states file in dict type
refer to https : / / www . terraform . io / docs / commands / destroy . html
: raise RuntimeError when return code is not zero
: return : ret_code , stdout , stderr
"""
if os . path . exists ( self . state_filename ) :
with open ( self . state_filename ) as f :
json_data = json . load ( f )
self . state_data = json_data
log . debug ( " state_data= %s " % str ( self . state_data ) )
return json_data
return dict ( )
def is_any_aws_instance_alive ( self ) :
self . refresh ( )
if not os . path . exists ( self . state_filename ) :
log . debug ( " can ' t find %s " % self . state_data )
return False
self . read_state_file ( )
try :
main_module = self . _get_main_module ( )
for resource_key , info in main_module [ ' resources ' ] . items ( ) :
if ' aws_instance ' in resource_key :
log . debug ( " %s is found when read state " % resource_key )
return True
log . debug ( " no aws_instance found in resource key " )
return False
except KeyError as err :
log . debug ( str ( err ) )
return False
except TypeError as err :
log . debug ( str ( err ) )
return False
def _get_main_module ( self ) :
return self . state_data [ ' modules ' ] [ 0 ]
def get_aws_instances ( self ) :
instances = dict ( )
try :
main_module = self . _get_main_module ( )
for resource_key , info in main_module [ ' resources ' ] . items ( ) :
if ' aws_instance ' in resource_key :
instances [ resource_key ] = info
except KeyError :
return instances
except TypeError :
return instances
return instances
def get_aws_instance ( self , resource_name ) :
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 ) :
"""
: param resource_name :
name of terraform resource , make source count is attached
: return : return None if not exist , dict type if exist
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
"""
try :
return self . get_aws_instances ( ) [ resource_name ]
except KeyError :
return None
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
def get_output_value ( self , output_name ) :
# 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 ) )
: param output_name :
: return :
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
"""
try :
main_module = self . _get_main_module ( )
return main_module [ ' outputs ' ] [ output_name ]
except KeyError :
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 ( )
@staticmethod
def _generate_var_string ( d ) :
str_t = [ ]
for k , v in d . iteritems ( ) :
str_t + = [ ' -var ' ] + [ " %s = %s " % ( k , v ) ]
output_dict = json . loads ( out )
return output_dict [ ' value ' ]
return str_t
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
"""
@staticmethod
def _generate_targets ( targets ) :
str_t = [ ]
for t in targets :
str_t + = [ ' -target= %s ' % t ]
return str_t
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 )
Freddy Tan
8 years ago