143 lines
4.3 KiB
Python
143 lines
4.3 KiB
Python
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)
|
||
print(convert_config)
|
||
|
||
# テンプレートとなるファイルを読み込み
|
||
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)
|