import os from actions_logging.app_logging import logger from env_files.constants import DEFAULT_ENV_FILES_PATH, GLOBAL_FILE, KNOWN_FE_BUILD_PREFIXES from env_files.utils import get_sub_dir_for_env_file from env_files.create_dot_env import create_dot_env_from_json, create_dotenv_data from github.env import exit_on_error_and_write_summary, get_required_env_var, write_github_env from github.get_file_from_github import get_file_from_github def write_to_github_env_from_env_file(file_path): """ Reads environment variables from a specified file and writes them to the GitHub environment. Args: file_path (str): The path to the environment file containing key-value pairs. Behavior: - Reads the file line by line, expecting each line to be in the format `KEY=VALUE`. - Splits each line into a key and value. If a key exists without a value, the value is set to an empty string. - Skips setting the `ENV_NAME` variable in the GitHub environment if its value is empty. - Logs debug messages for each key-value pair being written to the GitHub environment. - Calls `write_github_env(value, key)` to write each key-value pair to the GitHub environment. Exceptions: - If the file is not found, it exits the program and writes an error summary. Note: This function assumes the existence of `logger.debug`, `write_github_env`, and `exit_on_error_and_write_summary` functions for logging, writing to the GitHub environment, and handling errors, respectively. """ logger.info(f"Starting to process environment file: {file_path}") try: with open(file_path, "r") as file: data = file.readlines() logger.debug(f"Read {len(data)} lines from {file_path}") for line in data: kv = line.strip().split("=") key = kv[0] value = "" if len(kv) > 1: value = "=".join(kv[1:]) if key == "ENV_NAME" and not value: logger.debug("Empty ENV_NAME, skipping setting it in GitHub environment") continue logger.debug(f"Writing {key}:{value} to GitHub environment") write_github_env(value, key) logger.info(f"Successfully processed environment file: {file_path}") except FileNotFoundError: logger.error(f"Environment file not found: {file_path}") exit_on_error_and_write_summary(f"{file_path} file not found") except Exception as e: logger.error(f"An error occurred while processing the environment file: {e}") raise def get_env_file_path( env_name: str, svc_name: str, is_multi_service_repo: bool = False, env_files_path: str = DEFAULT_ENV_FILES_PATH ) -> str: """ Constructs the path to the environment file. Args: env_name (str): The name of the environment (e.g., "dev", "prod"). svc_name (str): The name of the service. is_multi_service_repo (bool): Indicates if the repository contains multiple services. env_files_path (str): The base path for environment files. Returns: str: The path to the environment file. """ logger.info( f"getting env file path for env: {env_name}, service: {svc_name}, multi-service repo: {is_multi_service_repo}" ) env_level = get_sub_dir_for_env_file(env_name) logger.debug(f"Determined environment level: {env_level}") env_file_path = os.path.join(env_files_path, env_level, f"{env_name}.json") logger.debug(f"Initial environment file path: {env_file_path}") if is_multi_service_repo: env_file_path = os.path.join(env_files_path, svc_name, env_level, f"{env_name}.json") logger.debug(f"Updated environment file path for multi-service repo: {env_file_path}") logger.info(f"Final environment file path: {env_file_path}") return env_file_path def get_global_file_path( svc_name: str, is_multi_service_repo: bool = False, env_files_path: str = DEFAULT_ENV_FILES_PATH ) -> str: """ Constructs the path to the global environment file. Args: svc_name (str): The name of the service. is_multi_service_repo (bool): Indicates if the repository contains multiple services. env_files_path (str): The base path for environment files. Returns: str: The path to the global environment file. """ logger.info(f"Constructing global file path for service: {svc_name}, multi-service repo: {is_multi_service_repo}") global_file_path = os.path.join(env_files_path, GLOBAL_FILE) logger.debug(f"Initial global file path: {global_file_path}") if is_multi_service_repo: global_file_path = os.path.join(env_files_path, svc_name, GLOBAL_FILE) logger.debug(f"Updated global file path for multi-service repo: {global_file_path}") logger.info(f"Final global file path: {global_file_path}") return global_file_path def update_env_files_from_version( repo_name: str, svc_name: str, env_name: str, git_env_version: str, is_multi_service_repo: bool = False ) -> tuple[str, str]: """ Updates environment files by fetching them from a specific Git version or using local files. Args: repo_name (str): The name of the repository where the environment files are stored. svc_name (str): The name of the service for which the environment files are being updated. env_name (str): The name of the environment (e.g., "dev", "prod"). git_env_version (str): The Git version (branch, tag, or commit hash) to fetch the environment files from. If empty, local environment files will be used. is_multi_service_repo (bool, optional): Indicates whether the repository contains multiple services. Defaults to False. Returns: tuple[str, str]: A tuple containing the paths to the environment file and the global file. """ env_file_path = get_env_file_path(env_name, svc_name, is_multi_service_repo) global_file_path = get_global_file_path(svc_name, is_multi_service_repo) if git_env_version: logger.info_green(f"fetching {env_file_path} and {global_file_path} from {git_env_version}") get_file_from_github(repo_name, env_file_path, os.path.dirname(env_file_path), git_env_version) get_file_from_github(repo_name, global_file_path, os.path.dirname(global_file_path), git_env_version) else: logger.info_blue("no ENV_FILE_VERSION provided, using local env files to build .env") return env_file_path, global_file_path def create_and_load_dotenv( repo_name: str, svc_name: str, env_name: str, git_env_version: str, is_multi_service_repo: bool = False ): """ Creates and loads a `.env` file for a given repository, service, and environment. This function updates environment files based on a specified version, generates a `.env` file from the updated environment JSON, logs its contents for debugging purposes, and writes the environment variables to the GitHub Actions environment. **Warning**: Logging the contents of the `.env` file can expose sensitive information, such as TLS certificates, which may not always be recognized as secrets. Use caution when running this function in debug mode, and ensure that builds using this function are deleted afterward. Args: repo_name (str): The name of the repository. svc_name (str): The name of the service. env_name (str): The name of the environment (e.g., "dev", "prod"). git_env_version (str): The version of the environment configuration in the repository. is_multi_service_repo (bool, optional): Indicates if the repository contains multiple services. Defaults to False. Raises: Exception: If any errors occur during the creation or loading of the `.env` file. Returns: None """ env_file_path, _ = update_env_files_from_version( repo_name, svc_name, env_name, git_env_version, is_multi_service_repo ) dotenv_file = create_dot_env_from_json(env_file_path, env_name, svc_name, is_multi_service_repo) with open(dotenv_file, "r") as f: logger.debug(f.readlines()) write_to_github_env_from_env_file(dotenv_file) def replace_values_to_placeholders( input_data: dict, output_data: dict, prefix: str, placeholder: str = "PLACEHOLDER_", ) -> dict: """ Replaces values in a dictionary with placeholder strings for keys that start with a specified prefix. Args: input_data (dict): The input dictionary containing key-value pairs to process. output_data (dict): The output dictionary where the processed key-value pairs will be stored. prefix (str): The prefix to identify keys whose values should be replaced with placeholders. placeholder (str, optional): The placeholder string to use as a prefix for replaced values. Defaults to "PLACEHOLDER_". Returns: dict: The updated output dictionary with values replaced by placeholders for matching keys. Raises: RuntimeError: If an error occurs during the replacement process. Example: input_data = {"ENV_VAR1": "value1", "PREFIX_VAR2": "value2"} output_data = {} result = replace_values_to_placeholders(input_data, output_data, prefix="PREFIX_") # result: {"ENV_VAR1": "value1", "PREFIX_VAR2": "PLACEHOLDER_PREFIX_VAR2"} """ try: logger.info(f"replacing values in flat k/v data with placeholders for keys starting with {prefix}") for key, value in input_data.items(): key_parts = key.split(prefix, 1) if len(key_parts) > 1: value = f"{placeholder}{prefix}{key_parts[-1]}" output_data[key] = value logger.info(f"replaced {len(output_data)} values with placeholders") return output_data except Exception as e: raise RuntimeError(f"Error while replacing values in json with placeholders: {e}") def get_key_prefixes_from_key_value(json_data: str, prefix_list: list[str] = KNOWN_FE_BUILD_PREFIXES) -> set[str]: """ Extracts known prefixes from the keys of a JSON-like dictionary. Args: json_data (str): The input dictionary containing key-value pairs. prefix_list (list[str]): A list of known prefixes to search for in the keys. Defaults to KNOWN_FE_BUILD_PREFIXES. Returns: set[str]: A set of prefixes found in the keys of the input dictionary. Raises: RuntimeError: If an error occurs during the extraction process. Example: json_data = {"PREFIX_KEY1": "value1", "OTHER_KEY2": "value2"} prefix_list = ["PREFIX_", "OTHER_"] result = get_key_prefixes_from_key_value(json_data, prefix_list) # result: {"PREFIX_", "OTHER_"} """ prefixes = set() try: logger.debug(f"Starting to extract prefixes from keys in json_data with prefix_list: {prefix_list}") for key in json_data.keys(): logger.debug(f"Processing key: {key}") prefix = None for p in prefix_list: if key.startswith(p): prefix = p logger.debug(f"Key {key} starts with prefix {p}") break if prefix: logger.info(f"Found prefix {prefix} in key {key}") prefixes.add(prefix) if len(prefixes) == len(KNOWN_FE_BUILD_PREFIXES): logger.debug("All known prefixes have been found, stopping search") break logger.info(f"Found {len(prefixes)} prefixes: {prefixes}") return prefixes except Exception as e: logger.error(f"Error while extracting known prefixes from json keys: {e}") raise RuntimeError(f"Error while extracting known prefixes from json keys: {e}") def load_dotenv_with_placeholders( repo_name: str, svc_name: str, env_name: str, git_env_version: str, placeholder: str = "PLACEHOLDER_", is_multi_service_repo: bool = False, ): """ Creates and loads a dotenv file with placeholder values for a given service and environment. This function generates a dotenv file with placeholder values based on the provided repository, service, and environment details. It replaces specific values in the dotenv data with placeholders and writes the resulting data to GitHub environment variables. Args: repo_name (str): The name of the repository. svc_name (str): The name of the service. env_name (str): The name of the environment (e.g., "dev", "prod"). git_env_version (str): The version of the environment configuration in Git. placeholder (str, optional): The placeholder prefix to use for replacing values. Defaults to "PLACEHOLDER_". is_multi_service_repo (bool, optional): Indicates if the repository contains multiple services. Defaults to False. Raises: RuntimeError: If an error occurs during the creation or loading of the dotenv file. Logs: - Info: Logs the start and completion of the dotenv file creation process. - Warning: Logs if no known prefixes are found in the dotenv keys. - Error: Logs any exceptions encountered during the process. """ try: logger.info(f"creating dot env file with placeholder values for {svc_name}") env_file_path, _ = update_env_files_from_version( repo_name, svc_name, env_name, git_env_version, is_multi_service_repo ) dotenv_data = create_dotenv_data(env_file_path, env_name, svc_name, is_multi_service_repo) prefixes = get_key_prefixes_from_key_value(dotenv_data) dotenv_with_placholders = None if not prefixes: logger.warning("no known prefixes found in dotenv keys, skipping placeholder replacement") dotenv_with_placholders = {} for prefix in prefixes: dotenv_with_placholders = replace_values_to_placeholders( dotenv_data, dotenv_with_placholders, prefix, placeholder ) [write_github_env(value, key) for key, value in dotenv_with_placholders.items()] logger.info_green("done injecting dotenv data with placeholder values") except Exception as e: raise RuntimeError(f"Error creating env file with placeholder values: {e}") if __name__ == "__main__": svc_name = get_required_env_var("SVC_NAME") env_name = get_required_env_var("ENV_NAME") repo_name = get_required_env_var("GITHUB_REPOSITORY") is_multi_service_repo = os.getenv("IS_MULTI_SERVICE_REPO", "false") == "true" git_env_version = os.getenv("ENV_FILE_VERSION") create_and_load_dotenv(repo_name, svc_name, env_name, git_env_version, is_multi_service_repo)