import os import sys from string import Template # you need to install module `PyYAML` import yaml CONF_FOLDER_PATH = '.' DEFINITION_FOLDER_BASE = '..' CONVERTED_FOLDER_BASE = 'build' CONVERT_CONF = 'convert_config.yaml' CHAR_CODE = 'utf-8' PRD_NAME = 'product' STG_NAME = 'staging' class StateMachineTemplate(Template): """StepFunctionsステートマシンの置き換えるためのテンプレート 生のTemplateクラスはプレースホルダーを表す文字列が`$`で、ステートマシン定義内で常用する文字と被ってしまうため、サブクラス化 """ # `#{プレースホルダー}`という形式が使えるようになる delimiter = "#" def __init__(self, template: str) -> None: super().__init__(template) def main(args=None): """ args1: StepFunctionsステートマシン名 args2: 変換先環境名(product or staging) """ # カレントディレクトリ移動 os.chdir(os.path.dirname(os.path.abspath(__file__))) # 引数チェック state_name, env_name = check_args(args) print('引数確認OK') # ファイル存在チェック check_file_exist(state_name) print('ファイル存在確認OK') # フォルダがなければ作る env_name_folder = f'{DEFINITION_FOLDER_BASE}/{state_name}/{CONVERTED_FOLDER_BASE}/{env_name}' if not os.path.isdir(env_name_folder): os.makedirs(env_name_folder, exist_ok=True) print(f'定義生成用フォルダを作成しました。フォルダ名:{env_name_folder}') # 変換 converted_file_name = convert_definition(state_name, env_name) print(f'変換が完了しました ファイル名:{converted_file_name}') return def check_args(args): try: check_length(args) check_aws_environment(args) return args[1], args[2] except Exception as e: raise Exception(f'引数確認に失敗しました {e}') def check_length(args): if len(args) != 3: raise Exception('引数の数が異常です') return def check_aws_environment(args): if args[2] not in [STG_NAME, PRD_NAME]: raise Exception('第2引数が不正です') return def check_file_exist(state_name): if not os.path.isfile(f'{CONF_FOLDER_PATH}/{CONVERT_CONF}'): raise Exception('変換定義ファイルが存在しません') if not os.path.isfile(f'{DEFINITION_FOLDER_BASE}/{state_name}/{state_name}.json'): raise Exception('変換元のステートマシン定義が存在しません') return def convert_definition(state_name, env_name): try: # 定義フォルダのパス生成 from_folder = f'{DEFINITION_FOLDER_BASE}/{state_name}' to_folder = f'{from_folder}/{CONVERTED_FOLDER_BASE}/{env_name}' # 変換定義の読み込み convert_config = read_env_specific_config(state_name, env_name) # テンプレートとなるファイルを読み込み with open(f'{from_folder}/{state_name}.json', mode='r', encoding=CHAR_CODE) as from_file: state_json = from_file.read() state_json_template = StateMachineTemplate(state_json) # 環境固有の値を置き換え substituted_state_json = state_json_template.substitute(ENV_NAME=env_name, **convert_config) # 書き込みファイルオープン with open(f'{to_folder}/{state_name}.json', mode='w', encoding=CHAR_CODE, newline='\n') as to_file: to_file.write(substituted_state_json) return f'{to_folder}/{state_name}.json' except Exception as e: raise Exception(f'変換に失敗しました {e}') def read_env_specific_config(state_name, env_name): try: # 変換定義ファイルオープン with open(f'{CONF_FOLDER_PATH}/{CONVERT_CONF}', mode='r', encoding=CHAR_CODE) as convert_conf_file: config_yaml = yaml.load(convert_conf_file, yaml.Loader) # 変換定義からステートマシンの環境固有の値を取得 env_specific_config = config_yaml['config'][state_name][env_name] return env_specific_config except Exception as e: Exception('変換定義ファイルの読み込みに失敗しました') if __name__ == '__main__': main(args=sys.argv)