add typing

alakazam.py

@@ -4,6 +4,7 @@ import os
 import json
 import logging
 from pathlib import Path
+from typing import List, Dict, Any, Optional, Union
 import subprocess
 import re
 
@@ -41,7 +42,7 @@ MySafeConstructor.add_constructor(
     MySafeConstructor.construct_yaml_str)
 
 
-def read_config(filepath):
+def read_config(filepath: str) -> Dict[str, Any]:
     """
     Reads a YAML configuration file from the specified filepath and returns the configuration as a dictionary, including processing any Jinja2 templating specified within the file. It handles the loading of YAML files, processes any Jinja2 templating using the 'GLOBALS' section for variables, and converts the file into a dictionary format.
 
@@ -70,7 +71,7 @@ def read_config(filepath):
     return yaml.load(template.render())
 
 
-def get_value(dict, *keys):
+def get_value(dictionary: Dict[Any, Any], *keys: str) -> Optional[Any]:
     """
     Retrieves a nested value from a dictionary using a series of keys. This function is useful for fetching deep values in a dictionary structure without manually checking each level of the dictionary.
 
@@ -81,7 +82,7 @@ def get_value(dict, *keys):
     Returns:
         The value found at the specified path in the dictionary, or None if the path is not present.
     """
-    _element = dict
+    _element = dictionary
     for key in keys:
         try:
             _element = _element[key]
@@ -90,7 +91,7 @@ def get_value(dict, *keys):
     return _element
 
 
-def merge_dict(dict1, dict2, reverse_list_order=False):
+def merge_dict(dict1: Dict[Any, Any], dict2: Dict[Any, Any], reverse_list_order: bool = False) -> Dict[Any, Any]:
     """
     Merges two dictionaries, combining their key-value pairs. Where keys overlap, values from dict2 will overwrite those from dict1.
     If reverse_list_order is True and the values are lists, the order of elements in lists from dict2 is reversed before merging.
@@ -119,7 +120,7 @@ def merge_dict(dict1, dict2, reverse_list_order=False):
     return merged_dict
 
 
-def merge_pool_configs(dir_path):
+def merge_pool_configs(dir_path: str) -> Dict[str, Dict[str, Any]]:
     """
     Recursively merges 'alaka.yml' files within a specified directory into a single comprehensive configuration dictionary. 
     Configurations at higher directory levels get inherited and potentially overridden by configurations in lower directory levels.
@@ -151,7 +152,7 @@ def merge_pool_configs(dir_path):
             merged_configs[root] = merged_configs.get(os.path.dirname(root))
     return merged_configs
 
-def merge_instance_configs(pool_config, instance_domain, instance_config):
+def merge_instance_configs(pool_config: Dict[str, Any], instance_domain: str, instance_config: Dict[str, Any]) -> Dict[str, Any]:
     """
     Merge instance-specific configurations with pool configurations to produce a consolidated configuration dictionary.
     This considers domain-specific overrides and merges them appropriately with the defaults specified at the pool level.
@@ -184,7 +185,7 @@ def merge_instance_configs(pool_config, instance_domain, instance_config):
     return merged_config
 
 
-def map_subdomain(recipe, instance_domain, app_config):
+def map_subdomain(recipe: str, instance_domain: str, app_config: Dict[str, Any]) -> str:
     """
     Maps a subdomain for an app based on the recipe, instance domain, and specific app configuration.
     It dynamically replaces the placeholder domain with the actual instance domain or constructs
@@ -205,7 +206,7 @@ def map_subdomain(recipe, instance_domain, app_config):
     return domain
 
 
-def get_merged_instance_configs(pool_path, pool_configs):
+def get_merged_instance_configs(pool_path: str, pool_configs: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
     """
     Traverse a directory structure to read and merge all YAML configuration files for each instance.
     This function supports a hierarchical configuration approach by aggregating configurations across directories.
@@ -238,7 +239,7 @@ def get_merged_instance_configs(pool_path, pool_configs):
     return instances
 
 
-def merge_connection_configs(configs):
+def merge_connection_configs(configs: Dict[str, Any]) -> Dict[str, Any]:
     """
     Merge connection configurations from the 'combine.yml' to extend instance configurations with inter-app secrets and settings.
     This involves integrating shared secrets and other connection-specific settings between applications within the same instance.
@@ -261,16 +262,13 @@ def merge_connection_configs(configs):
     return merged_configs
 
 
-def extend_shared_secrets(connection_config):
+def extend_shared_secrets(connection_config: Dict[str, Any]) -> None:
     """
     Extends connection configurations by embedding source app details into the shared secrets configuration.
     This modifies the existing connection configurations in place, adding a layer of source app information to shared secrets.
 
     Args:
         connection_config (dict): Connection configurations which involve shared secrets.
-
-    Returns:
-        None
     """
     for _, source_apps in connection_config.items(): 
         for source_app, target_conf in source_apps.items():
@@ -278,7 +276,7 @@ def extend_shared_secrets(connection_config):
                 target_conf['shared_secrets'] = {source_app: shared_secrets}
 
 
-def abra(*args, machine_output=False, ignore_error=False):
+def abra(*args: str, machine_output: bool = False, ignore_error: bool = False) -> Union[str,Dict]:
     """
     Execute the 'abra' command with the specified arguments. This function acts as a wrapper around the 'abra' CLI tool. It allows for capturing the output and optionally returning it as machine-readable JSON.
 
@@ -311,7 +309,7 @@ def abra(*args, machine_output=False, ignore_error=False):
     return process.stdout.decode()
 
 
-def write_env_header(path):
+def write_env_header(path: str) -> None:
     """
     Writes a header comment at the top of an environment file to indicate that it is generated automatically and should not be manually edited.
     This function ensures that anyone modifying the .env file is aware that changes might be overwritten by subsequent automated processes.
@@ -331,7 +329,7 @@ def write_env_header(path):
         file.write(header + old_content)
 
 
-def new_app(recipe, domain, server, version):
+def new_app(recipe: str, domain: str, server: str, version: str) -> None:
     """
     Generate a new application .env by calling the 'abra' command with the specified parameters.
     If the app .env already exists, it is removed before the new app is created.
@@ -360,7 +358,7 @@ def new_app(recipe, domain, server, version):
         logging.info(f'{recipe} created on {server} at {domain}')
 
 
-def update_configs(path, config):
+def update_configs(path: str, config: Dict[str, Any]) -> None:
     """
     Update the .env configuration files at the specified path according to the provided configuration dictionary.
     This function manages the commenting, uncommenting, and setting of environment variables as specified in the config dictionary.
@@ -380,7 +378,7 @@ def update_configs(path, config):
             dotenv.set_key(path, key, value, quote_mode="never")
 
 
-def generate_all_secrets(domain):
+def generate_all_secrets(domain: str) -> None:
     """
     Generates all secrets for the app specified by its domain using the 'abra' command.
 
@@ -398,11 +396,11 @@ def generate_all_secrets(domain):
         print(generated_secrets)
 
 
-def get_env_path(server, domain):
+def get_env_path(server: str, domain: str) -> str:
     return Path(f"~/.abra/servers/{server}/{domain}.env").expanduser()
 
 
-def exchange_secrets(app1, instance_config, apps):
+def exchange_secrets(app1: str, instance_config: Dict[str, Any], apps: List[str]) -> None:
     """
     Facilitates the exchange of shared secrets between apps within the same instance based on the configuration.
     This function checks for shared secrets configurations and applies them to ensure that secrets are synchronized between apps.
@@ -429,10 +427,10 @@ def exchange_secrets(app1, instance_config, apps):
             share_secrets(app2_domain, app1_domain, app2_shared_secrets)
 
 
-def str2bool(value):
+def str2bool(value: str) -> bool:
   return value.lower() in ("yes", "true", "t", "1")
 
-def share_secrets(app1_domain, app2_domain, secrets):
+def share_secrets(app1_domain: str, app2_domain: str, secrets: Dict[str, str]) -> None:
     """
     Facilitates the sharing of secrets between two applications.
     This function checks and transfers secrets from one applications to another if one applications possesses a secret that the other lacks, ensuring both applications maintain synchronized secret configurations.
@@ -470,7 +468,7 @@ def share_secrets(app1_domain, app2_domain, secrets):
             insert_secret(app2_domain, app2_secret, secret)
 
 
-def get_secret(domain, secret_name):
+def get_secret(domain: str, secret_name: str) -> str:
     """
     Retrieves a specified secret from the application specified by its domain using the 'abra' command.
     This function is useful for fetching secrets that need to be shared or updated between different applications.
@@ -486,7 +484,7 @@ def get_secret(domain, secret_name):
     secret = abra("app", "run", domain, "worker", "cat", f"/var/run/secrets/{secret_name}")
     return secret
 
-def generate_secret(domain, secret_name):
+def generate_secret(domain: str, secret_name: str) -> str:
     """
     Generates a new secret with a specified name for a given app domain. This function is utilized when a required secret is missing or needs to be regenerated.
 
@@ -501,7 +499,7 @@ def generate_secret(domain, secret_name):
     return secret[0]['value']
 
 
-def insert_secrets_from_conf(domain, config):
+def insert_secrets_from_conf(domain: str, config: Dict[str, Any]) -> None:
     """
     Inserts secrets into the specified app based on a configuration dictionary.
 
@@ -515,7 +513,7 @@ def insert_secrets_from_conf(domain, config):
             insert_secret(domain, secret_name, secret)
 
 
-def unquote_strings(s):
+def unquote_strings(s: str) -> str:
     """
     Removes surrounding single or double quotes from a string if present.
     This is used to clean the quotes from strings extracted from configurations or secrets, ensuring that the values can be used directly without formatting issues.
@@ -534,7 +532,7 @@ def unquote_strings(s):
         return s
 
 
-def insert_secret(domain, secret_name, secret):
+def insert_secret(domain: str, secret_name: str, secret: str) -> None:
     """
     Inserts a secret for a specific app only if the secret does not already exists.
     This function interacts with the 'abra' command to manage secrets on the server.
@@ -553,7 +551,7 @@ def insert_secret(domain, secret_name, secret):
         abra("app", "secret", "insert", domain, secret_name, "v1", secret)
 
 
-def uncomment(keys, path, match_all=False):
+def uncomment(keys: List[str], path: str, match_all: bool = False) -> None:
     """
     Uncomments lines in a configuration file that contain specified keys.
     If 'match_all' is True, it matches against the entire line, otherwise, it matches only against the key.
@@ -577,7 +575,7 @@ def uncomment(keys, path, match_all=False):
             file.write(line)
 
 
-def comment(keys, path, match_all=False):
+def comment(keys: List[str], path: str, match_all: bool = False) -> None:
     """
     Comments lines in a configuration file that contain specified keys.
     If 'match_all' is True, it matches against the entire line, otherwise, it matches only against the key.
@@ -601,7 +599,7 @@ def comment(keys, path, match_all=False):
             file.write(line)
 
 
-def exchange_domains(instance_domain, instance_config, path):
+def exchange_domains(instance_domain: str, instance_config: Dict[str, Any], path: str) -> None:
     """
     Replaces all domain references in the specified .env configuration file based on the instance configuration. It is used to ensure that all references to an application's domain are consistent across various configuration files.
 
@@ -623,7 +621,7 @@ def exchange_domains(instance_domain, instance_config, path):
 
 
 
-def replace_domains(path, old_domain, new_domain):
+def replace_domains(path: str, old_domain: str, new_domain: str) -> None:
     """
     Replaces occurrences of an old domain with a new domain in a configuration file.
 
@@ -640,7 +638,7 @@ def replace_domains(path, old_domain, new_domain):
         file.write(content)
 
 
-def list_commands(app_config):
+def list_commands(app_config: Dict[str, Any]) -> None:
     """
     Lists all post-deployment commands for an app based on the app configuration.
     This can help in verifying which commands are set to run after deployment.
@@ -661,7 +659,7 @@ def list_commands(app_config):
         print(f"{domain}:{container} --> '{cmd}'")
 
 
-def execute_cmds(app_config):
+def execute_cmds(app_config: Dict[str, Any]) -> None:
     """
     Execute post-deployment commands for an application based on the provided configuration.
     This can include running scripts or commands inside the application's environment.
@@ -690,7 +688,7 @@ def execute_cmds(app_config):
 @click.option('-l', '--log', 'loglevel')
 @click.option('-p', '--pool_path', 'pool_path')
 @click.option('-c', '--config_path', 'config_path', default=".")
-def cli(loglevel, pool_path, config_path):
+def cli(loglevel: str, pool_path: str, config_path: str) -> None:
     """
     Command-line interface setup function for the Alakazam application. It configures logging levels, loads configuration files, and merges configuration settings from specified paths.
     This function is the entry point for CLI commands provided by the Alakazam tool.
@@ -716,7 +714,7 @@ def cli(loglevel, pool_path, config_path):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def config(apps):
+def config(apps: List[str]) -> None:
     """
     Generates and updates .env configuration files for a specified list of applications.
     This function reads the necessary configurations from the global settings and applies these to create or update .env files for each app.
@@ -750,7 +748,7 @@ def config(apps):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def secrets(apps):
+def secrets(apps: List[str]) -> None:
     """
     Generates and inserts secrets for specified apps.
     This function handles the generation of new secrets, the syncronisation of shared secrets and the insertion of existing secrets from the configuration into the appropriate locations for each application.
@@ -758,7 +756,7 @@ def secrets(apps):
     Args:
         apps (list): A list of application names for which secrets are to be generated and managed.
     """
-    for instance, instance_config in CONFIGS.items():
+    for _, instance_config in CONFIGS.items():
         instance_apps = instance_config.keys()
         if apps:
             selected_apps = [app for app in apps if app in instance_config.keys()]
@@ -773,7 +771,7 @@ def secrets(apps):
             generate_all_secrets(domain)
 
 
-def get_deployed_apps(apps):
+def get_deployed_apps(apps: List[str]) -> Dict[str, Any]:
     """
     Retrieves a list of deployed apps and their versions.
     This function utilizes the 'abra' command-line tool to fetch deployment information.
@@ -807,7 +805,7 @@ def get_deployed_apps(apps):
 @click.option('-r', '--run-cmds', is_flag=True)
 @click.option('-f', '--force', is_flag=True)
 @click.option('-c', '--converge-checks', is_flag=True)
-def deploy(apps, run_cmds, force, converge_checks):
+def deploy(apps: List[str], run_cmds: bool, force: bool, converge_checks: bool) -> None:
     """
     Deploys applications as specified in the configuration.
 
@@ -859,7 +857,7 @@ def deploy(apps, run_cmds, force, converge_checks):
 @click.option('-a', '--apps', multiple=True)
 @click.option('-r', '--run-cmds', is_flag=True)
 @click.option('-d', '--dry-run', is_flag=True)
-def upgrade(apps, run_cmds, dry_run):
+def upgrade(apps: List[str], run_cmds: bool, dry_run: bool) -> None:
     """
     Upgrades specified applications by executing the upgrade commands via the 'abra' command-line interface.
     It checks the current deployment status of the apps and performs upgrades only where necessary, with options to execute additional commands or perform a dry run. It either took the target version from the configuration or it uses the latest available version.
@@ -912,7 +910,7 @@ def upgrade(apps, run_cmds, dry_run):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def undeploy(apps):
+def undeploy(apps: List[str]) -> None:
     """
     Undeploys multiple applications at once.
 
@@ -940,7 +938,7 @@ def undeploy(apps):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def cmds(apps):
+def cmds(apps: List[str]) -> None:
     """
     Execute post-deployment commands for all specified applications based on the provided configuration.
 
@@ -967,7 +965,7 @@ def cmds(apps):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def list_cmds(apps):
+def list_cmds(apps: List[str]) -> None:
     """
     Lists all post-deployment commands for specified applications.
     This function helps in reviewing which commands would run after the deployment of the specified apps.
@@ -987,7 +985,7 @@ def list_cmds(apps):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def purge(apps):
+def purge(apps: List[str]) -> None:
     """
     Completely removes applications and their configurations. This function is used to clean up all traces of an application from the server.
 
@@ -1009,7 +1007,7 @@ def purge(apps):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def ls(apps):
+def ls(apps: List[str]) -> None:
     """
     Lists all selected applications along with their domains.
 
@@ -1019,7 +1017,7 @@ def ls(apps):
     print_all_apps(apps)
 
 
-def print_all_apps(apps):
+def print_all_apps(apps: List[str]) -> Dict[str, Any]:
     """
     Prints a detailed list of all specified applications including their instances and domains. 
 
@@ -1036,7 +1034,7 @@ def print_all_apps(apps):
         print()
     return pool_apps
 
-def list_apps(apps):
+def list_apps(apps: Optional[List[str]] = None) -> Dict[str, List[tuple]]:
     """
     Retrieves a list of applications and their associated domains from the configuration.
     This function provides an organized view of the applications within their respective instances.
@@ -1065,7 +1063,7 @@ def list_apps(apps):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def purge_secrets(apps):
+def purge_secrets(apps: List[str]) -> None:
     """
     Purges all secrets associated with specified applications.
 
@@ -1083,7 +1081,7 @@ def purge_secrets(apps):
 
 @cli.command()
 @click.option('-a', '--apps', multiple=True)
-def purge_volumes(apps):
+def purge_volumes(apps: List[str]) -> None:
     """
     Purges all volumes associated with specified applications, ensuring that all data stored within the app's volumes is cleanly removed.