|
import json |
|
import os |
|
import re |
|
from copy import deepcopy |
|
from distutils.util import strtobool |
|
from pathlib import Path |
|
|
|
import yaml |
|
|
|
|
|
|
|
def build_from_file(file_path: str): |
|
path = Path(file_path) |
|
workers_path = path / "workers" |
|
if not workers_path.exists(): |
|
return [] |
|
|
|
|
|
worker_configs = {} |
|
other_configs = {} |
|
file_names = {} |
|
|
|
for file in path.rglob("*"): |
|
if file.suffix not in [".json", ".yaml", ".yml"]: |
|
continue |
|
|
|
key = file.name.split(".", 1)[0] |
|
|
|
if key in file_names: |
|
raise Exception( |
|
f"Duplicate file name [{key}] found:\n" |
|
f"File 1: {file_names[key]}\n" |
|
f"File 2: {file}" |
|
) |
|
file_names[key] = file |
|
|
|
try: |
|
with open(file, "r") as f: |
|
if file.suffix == ".json": |
|
content = json.load(f) |
|
else: |
|
content = yaml.load(f, Loader=yaml.FullLoader) |
|
except Exception as e: |
|
raise Exception(f"Error loading file {file}: {str(e)}") |
|
|
|
if workers_path in file.parents: |
|
worker_configs[key] = content |
|
else: |
|
other_configs[key] = content |
|
|
|
for conf in worker_configs.values(): |
|
prep_config(conf, other_configs, []) |
|
|
|
worker_configs_list = [] |
|
for worker_config in worker_configs.values(): |
|
if isinstance(worker_config, list): |
|
worker_configs_list.extend(worker_config) |
|
else: |
|
worker_configs_list.append(worker_config) |
|
|
|
return worker_configs_list |
|
|
|
|
|
def prep_config(sub_config: dict | list, config: dict, forbid_keys: list): |
|
if isinstance(sub_config, dict): |
|
for key, conf in sub_config.items(): |
|
if isinstance(conf, str): |
|
if match := re.search(r"\$\{sub\|([^}]+)\}", conf): |
|
module_key = match.group(1).strip() |
|
if module_key not in config: |
|
raise Exception( |
|
"Incomplete configuration, lack module [{}]".format( |
|
module_key |
|
) |
|
) |
|
elif module_key in forbid_keys: |
|
raise Exception( |
|
"Can't reference submodule recursively. [{}]".format( |
|
module_key |
|
) |
|
) |
|
sub_config[key] = deepcopy(config[module_key]) |
|
prep_config(sub_config[key], config, forbid_keys + [module_key]) |
|
elif match := re.search(r"\$\{env\|([^,}]+)(?:,([^}]+))?\}", conf): |
|
env_key = match.group(1).strip() |
|
default_value = match.group(2) |
|
env_value = os.getenv(env_key) |
|
if env_value: |
|
sub_config[key] = env_value |
|
elif not env_value and default_value: |
|
sub_config[key] = default_value.strip() |
|
if sub_config[key] == "null" or sub_config[key] == "~": |
|
sub_config[key] = None |
|
else: |
|
raise ValueError( |
|
f"Environmental variable {env_key} need to be set." |
|
) |
|
|
|
elif isinstance(conf, dict): |
|
prep_config(sub_config[key], config, forbid_keys) |
|
|
|
elif isinstance(conf, list): |
|
for i, item in enumerate(conf): |
|
if isinstance(item, dict): |
|
prep_config(sub_config[key][i], config, forbid_keys) |
|
elif isinstance(sub_config, list): |
|
for item in sub_config: |
|
prep_config(item, config, forbid_keys) |
|
|
