From 7e023a5f0852cb6cb1ee0bb2ba5f6afc4a569a6a Mon Sep 17 00:00:00 2001 From: Correl Roush Date: Tue, 23 Jan 2024 15:06:29 -0500 Subject: [PATCH] Finish copy command --- ssm.py | 29 ++++++++++++++++++++++------- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/ssm.py b/ssm.py index aa2e0cc..17768c2 100644 --- a/ssm.py +++ b/ssm.py @@ -43,7 +43,12 @@ def get_client(profile: str) -> botocore.client.BaseClient: def get_parameters( client: botocore.client.BaseClient, path: SSMPath, recursive: bool ) -> typing.Iterable[dict]: - ... + try: + parameter = client.get_parameter(Name=str(path)) + yield parameter.get("Parameter") + return + except client.exceptions.ParameterNotFound: + pass results = client.get_paginator("get_parameters_by_path").paginate( Path=str(path), Recursive=recursive, @@ -63,7 +68,7 @@ def profiles() -> None: print(profile) -@app.command("list") +@app.command("ls") def list_parameters( profile: ProfileArg, path: pathlib.Path, @@ -135,15 +140,15 @@ def unset(profile: ProfileArg, path: str) -> None: sys.exit(1) -@app.command() +@app.command("cp") def copy_tree( source_profile: ProfileArg, source_path: str, - dest_profile: ProfileArg, dest_path: str, + dest_profile: ProfileArg | None = None, replacement_pairs: list[str] = typer.Option([], "--replace", "-r"), recursive: bool = True, -): +) -> None: """Copy parameters from SRC_PATH to DEST_PATH.""" def parse_replacements(values: list[str]) -> list[tuple[str, str]]: @@ -161,13 +166,13 @@ def copy_tree( return value source = get_client(source_profile) - destination = get_client(dest_profile) + destination = get_client(dest_profile or source_profile) sources = { str(pathlib.Path(p["Name"]).relative_to(SSMPath(source_path))): p for p in get_parameters(source, SSMPath(source_path), recursive=recursive) } targets = { - str(pathlib.Path(p["Name"]).relative_to(SSMPath(source_path))): p + str(pathlib.Path(p["Name"]).relative_to(SSMPath(dest_path))): p for p in get_parameters(destination, SSMPath(dest_path), recursive=recursive) } table = rich.table.Table("Path", "Old Value", "New Value") @@ -186,6 +191,16 @@ def copy_tree( if not confirmed: stderr.print("No changes applied", style="yellow italic") sys.exit(1) + for name, param in sources.items(): + new_name = str(SSMPath(dest_path) / pathlib.Path(name)) + stdout.print(f"Writing {new_name}...") + destination.put_parameter( + Name=new_name, + Value=param["Value"], + Type=param["Type"], + Overwrite=True, + Description=param.get("Description", ""), + ) if __name__ == "__main__":