Fix some types
This commit is contained in:
57
alakazam.py
57
alakazam.py
@ -4,7 +4,7 @@ import os
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any, Optional, Union, Set
|
||||
from typing import List, Dict, Any, Optional, Union, Set, Tuple
|
||||
import subprocess
|
||||
import re
|
||||
|
||||
@ -367,7 +367,7 @@ def abra(*args: str, machine_output: bool = False, ignore_error: bool = False) -
|
||||
return process.stdout.decode()
|
||||
|
||||
|
||||
def write_env_header(path: str) -> None:
|
||||
def write_env_header(path: Path) -> 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.
|
||||
@ -416,7 +416,7 @@ def new_app(recipe: str, domain: str, server: str, version: str) -> None:
|
||||
logging.info(f'{recipe} created on {server} at {domain}')
|
||||
|
||||
|
||||
def update_configs(path: str, config: Dict[str, Any]) -> None:
|
||||
def update_configs(path: Path, 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.
|
||||
@ -454,11 +454,11 @@ def generate_all_secrets(domain: str) -> None:
|
||||
print(generated_secrets)
|
||||
|
||||
|
||||
def get_env_path(server: str, domain: str) -> str:
|
||||
def get_env_path(server: str, domain: str) -> Path:
|
||||
return Path(f"~/.abra/servers/{server}/{domain}.env").expanduser()
|
||||
|
||||
|
||||
def exchange_secrets(app1: str, instance_config: Dict[str, Any], apps: List[str]) -> None:
|
||||
def exchange_secrets(app1: str, instance_config: Dict[str, Any], apps: Tuple[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.
|
||||
@ -559,9 +559,10 @@ def get_secret(domain: str, secret_name: str, containers: List) -> str:
|
||||
for container in containers:
|
||||
try:
|
||||
secret = abra("app", "run", domain, container, "cat", f"/var/run/secrets/{secret_name}")
|
||||
return secret
|
||||
return str(secret)
|
||||
except RuntimeError:
|
||||
continue
|
||||
raise RuntimeError(f"{secret_name} not found for {domain}")
|
||||
|
||||
def generate_secret(domain: str, secret_name: str) -> str:
|
||||
"""
|
||||
@ -677,7 +678,7 @@ def comment(keys: List[str], path: str, match_all: bool = False) -> None:
|
||||
file.write(line)
|
||||
|
||||
|
||||
def exchange_domains(instance_domain: str, instance_config: Dict[str, Any], path: str) -> None:
|
||||
def exchange_domains(instance_domain: str, instance_config: Dict[str, Any], path: Path) -> 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.
|
||||
|
||||
@ -699,7 +700,7 @@ def exchange_domains(instance_domain: str, instance_config: Dict[str, Any], path
|
||||
|
||||
|
||||
|
||||
def replace_domains(path: str, old_domain: str, new_domain: str) -> None:
|
||||
def replace_domains(path: Path, old_domain: str, new_domain: str) -> None:
|
||||
"""
|
||||
Replaces occurrences of an old domain with a new domain in a configuration file.
|
||||
|
||||
@ -766,7 +767,7 @@ def execute_cmds(app_config: Dict[str, Any]) -> None:
|
||||
@click.option('-l', '--log', 'loglevel', help='Desired logging level ("debug", "info", "warning", "error", "critical")')
|
||||
@click.option('-e', '--exclude', help='Path to a directory that contains a group of instance configurations to be excluded.', multiple=True, type=click.Path(exists=True))
|
||||
@click.argument('group_path', type=click.Path(exists=True))
|
||||
def cli(loglevel: str, group_path: str, exclude:List[str]) -> None:
|
||||
def cli(loglevel: str, group_path: str, exclude:Tuple[str]) -> None:
|
||||
"""
|
||||
Alakazam is a meta-configuration app-connector and an abra wrapper, designed as a proof-of-concept to simplify the management of environment configuration files across multiple instances.
|
||||
|
||||
@ -796,7 +797,7 @@ def cli(loglevel: str, group_path: str, exclude:List[str]) -> None:
|
||||
|
||||
@cli.command()
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
def show_config(apps: List[str]) -> None:
|
||||
def show_config(apps: Tuple[str]) -> None:
|
||||
"""
|
||||
Show the merged instance configurations.
|
||||
|
||||
@ -812,7 +813,7 @@ def show_config(apps: List[str]) -> None:
|
||||
|
||||
@cli.command()
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
def config(apps: List[str]) -> None:
|
||||
def config(apps: Tuple[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.
|
||||
@ -846,7 +847,7 @@ def config(apps: List[str]) -> None:
|
||||
|
||||
@cli.command()
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
def secrets(apps: List[str]) -> None:
|
||||
def secrets(apps: Tuple[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.
|
||||
@ -869,7 +870,7 @@ def secrets(apps: List[str]) -> None:
|
||||
generate_all_secrets(domain)
|
||||
|
||||
|
||||
def get_deployed_apps(apps: List[str]) -> Dict[str, str]:
|
||||
def get_deployed_apps(apps: Tuple[str]) -> Dict[str, str]:
|
||||
"""
|
||||
Retrieves a list of deployed apps and their versions.
|
||||
This function utilizes the 'abra' command-line tool to fetch deployment information.
|
||||
@ -903,7 +904,7 @@ def get_deployed_apps(apps: List[str]) -> Dict[str, str]:
|
||||
@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: List[str], run_cmds: bool, force: bool, converge_checks: bool) -> None:
|
||||
def deploy(apps: Tuple[str], run_cmds: bool, force: bool, converge_checks: bool) -> None:
|
||||
"""
|
||||
Deploys applications as specified in the configuration.
|
||||
|
||||
@ -950,7 +951,7 @@ def deploy(apps: List[str], run_cmds: bool, force: bool, converge_checks: bool)
|
||||
@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: List[str], run_cmds: bool, dry_run: bool) -> None:
|
||||
def upgrade(apps: Tuple[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.
|
||||
@ -1007,13 +1008,13 @@ def upgrade(apps: List[str], run_cmds: bool, dry_run: bool) -> None:
|
||||
for app_config, cmd in upgrade_cmds:
|
||||
print(abra("app", *cmd))
|
||||
if run_cmds:
|
||||
logging.info(f'execute commands for {domain}')
|
||||
execute_cmds(app_config)
|
||||
logging.info(f'execute commands for {app_config.get("subdomain")}')
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
def undeploy(apps: List[str]) -> None:
|
||||
def undeploy(apps: Tuple[str]) -> None:
|
||||
"""
|
||||
Undeploys multiple applications at once.
|
||||
|
||||
@ -1089,7 +1090,7 @@ def list_cmds(apps: List[str]) -> None:
|
||||
|
||||
@cli.command()
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
def purge(apps: List[str]) -> None:
|
||||
def purge(apps: Tuple[str]) -> None:
|
||||
"""
|
||||
Completely removes applications and their configurations. This function is used to clean up all traces of an application from the server.
|
||||
|
||||
@ -1115,7 +1116,7 @@ def purge(apps: List[str]) -> None:
|
||||
@click.option('-d', '--deployed', is_flag=True)
|
||||
@click.option('-u', '--undeployed', is_flag=True)
|
||||
@click.option('--domains', is_flag=True)
|
||||
def ls(apps: List[str], deployed: bool, undeployed: bool, domains: bool) -> None:
|
||||
def ls(apps: Tuple[str], deployed: bool, undeployed: bool, domains: bool) -> None:
|
||||
"""
|
||||
Lists all selected applications along with their domains.
|
||||
|
||||
@ -1151,7 +1152,7 @@ def print_all_apps(instance_apps: Dict[str, List[List]]) -> None:
|
||||
print(tabulate(apps, headers=["",instance, "", ""], tablefmt='rounded_outline'))
|
||||
print()
|
||||
|
||||
def get_apps(apps: Optional[List[str]] = None) -> Dict[str, List[List]]:
|
||||
def get_apps(apps: Optional[Tuple[str]] = None) -> Dict[str, List[List]]:
|
||||
"""
|
||||
Retrieves a dict of applications and their associated domains from the configuration.
|
||||
This function provides an organized view of the applications within their respective instances.
|
||||
@ -1178,7 +1179,7 @@ def get_apps(apps: Optional[List[str]] = None) -> Dict[str, List[List]]:
|
||||
return instance_apps
|
||||
|
||||
|
||||
def get_server(apps: Optional[List[str]] = None) -> Set[str]:
|
||||
def get_server(apps: Optional[Tuple[str]] = None) -> Set[str]:
|
||||
"""
|
||||
Retrieves a list of server for the selected apps.
|
||||
|
||||
@ -1200,7 +1201,7 @@ def get_server(apps: Optional[List[str]] = None) -> Set[str]:
|
||||
return server
|
||||
|
||||
|
||||
def get_apps_by_deployment(apps: List[str], deployed: bool = True) -> Dict[str, List[List]]:
|
||||
def get_apps_by_deployment(apps: Tuple[str], deployed: bool = True) -> Dict[str, List[List]]:
|
||||
"""
|
||||
Retrieves a dict of deployed/undeployed apps and their associated domains.
|
||||
|
||||
@ -1241,7 +1242,7 @@ def get_latest_version(app: str) -> str:
|
||||
@cli.command()
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
@click.option('-s', '--secret', multiple=False)
|
||||
def purge_secrets(apps: List[str], secret:str) -> None:
|
||||
def purge_secrets(apps: Tuple[str], secret:str) -> None:
|
||||
"""
|
||||
Purges all secrets associated with specified applications.
|
||||
|
||||
@ -1265,7 +1266,7 @@ def purge_secrets(apps: List[str], secret:str) -> None:
|
||||
|
||||
@cli.command()
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
def purge_volumes(apps: List[str]) -> None:
|
||||
def purge_volumes(apps: Tuple[str]) -> None:
|
||||
"""
|
||||
Purges all volumes associated with specified applications, ensuring that all data stored within the app's volumes is cleanly removed.
|
||||
|
||||
@ -1287,7 +1288,7 @@ def purge_volumes(apps: List[str]) -> None:
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
@click.option('-w', '--watch', is_flag=True)
|
||||
@click.option('-v', '--verbose', is_flag=True)
|
||||
def ps(apps: List[str], watch: bool, verbose: bool) -> None:
|
||||
def ps(apps: Tuple[str], watch: bool, verbose: bool) -> None:
|
||||
"""
|
||||
Check the health status of all apps inside the selected group.
|
||||
|
||||
@ -1334,7 +1335,7 @@ def install() -> None:
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
@click.option('-m', '--message')
|
||||
@click.option('-p', '--push', is_flag=True)
|
||||
def commit(apps: List[str], message: str, push: bool = False) -> None:
|
||||
def commit(apps: Tuple[str], message: str, push: bool = False) -> None:
|
||||
"""
|
||||
Commit the changes to the .env repositories inside ~/.abra/servers and push them to the remote.
|
||||
|
||||
@ -1371,7 +1372,7 @@ def commit(apps: List[str], message: str, push: bool = False) -> None:
|
||||
|
||||
@cli.command()
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
def diff(apps: List[str]) -> None:
|
||||
def diff(apps: Tuple[str]) -> None:
|
||||
"""
|
||||
Show the changes in the .env repositories inside ~/.abra/servers.
|
||||
|
||||
@ -1397,7 +1398,7 @@ def diff(apps: List[str]) -> None:
|
||||
@click.option('-a', '--apps', multiple=True)
|
||||
@click.option('-u', '--update', is_flag=True)
|
||||
@click.option('missing', '-m', '--get-missing', is_flag=True)
|
||||
def uptime(apps: List[str], update: bool, missing: bool) -> None:
|
||||
def uptime(apps: Tuple[str], update: bool, missing: bool) -> None:
|
||||
"""
|
||||
Add the app domains to an uptime kuma instance.
|
||||
|
||||
|
Reference in New Issue
Block a user