Add copy-tree command

ssm.py

@@ -10,6 +10,7 @@ import botocore.client
 import botocore.session
 import rich.console
 import rich.table
+import rich.text
 import typer
 
 app = typer.Typer()
@@ -39,6 +40,22 @@ def get_client(profile: str) -> botocore.client.BaseClient:
         sys.exit(1)
 
 
+def get_parameters(
+    client: botocore.client.BaseClient, path: SSMPath, recursive: bool
+) -> typing.Iterable[dict]:
+    ...
+    results = client.get_paginator("get_parameters_by_path").paginate(
+        Path=str(path),
+        Recursive=recursive,
+        WithDecryption=True,
+    )
+    for parameter in (
+        result for chunk in results for result in chunk.get("Parameters")
+    ):
+        if parameter.get("Name"):
+            yield parameter
+
+
 @app.command()
 def profiles() -> None:
     """List available AWS profiles."""
@@ -46,8 +63,8 @@ def profiles() -> None:
         print(profile)
 
 
-@app.command()
-def list(
+@app.command("list")
+def list_parameters(
     profile: ProfileArg,
     path: pathlib.Path,
     recursive: bool = True,
@@ -63,21 +80,17 @@ def list(
         "Type",
         title=f"{root} ({profile})",
     )
-    results = client.get_paginator("get_parameters_by_path").paginate(
-        Path=str(root),
-        Recursive=recursive,
-        WithDecryption=True,
-    )
-    for parameter in (
-        result for chunk in results for result in chunk.get("Parameters")
-    ):
-        if name := parameter.get("Name"):
+    try:
+        for parameter in get_parameters(client, root, recursive):
             table.add_row(
-                str(pathlib.Path(name).relative_to(root)),
+                str(pathlib.Path(parameter["Name"]).relative_to(root)),
                 parameter.get("Value"),
                 parameter.get("Description"),
                 parameter.get("Type"),
             )
+    except client.exceptions.ClientError as e:
+        stderr.print(str(e), style="bold red")
+        sys.exit(1)
     console.print(table)
 
 
@@ -122,5 +135,58 @@ def unset(profile: ProfileArg, path: str) -> None:
         sys.exit(1)
 
 
+@app.command()
+def copy_tree(
+    source_profile: ProfileArg,
+    source_path: str,
+    dest_profile: ProfileArg,
+    dest_path: str,
+    replacement_pairs: list[str] = typer.Option([], "--replace", "-r"),
+    recursive: bool = True,
+):
+    """Copy parameters from SRC_PATH to DEST_PATH."""
+
+    def parse_replacements(values: list[str]) -> list[tuple[str, str]]:
+        pairs: list[tuple[str, str]] = []
+        for value in values:
+            a, b = value.split("=", 1)
+            pairs.append((a, b))
+        return pairs
+
+    replacements = parse_replacements(replacement_pairs)
+
+    def replace(value: str) -> str:
+        for old, new in replacements:
+            value = value.replace(old, new)
+        return value
+
+    source = get_client(source_profile)
+    destination = get_client(dest_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
+        for p in get_parameters(destination, SSMPath(dest_path), recursive=recursive)
+    }
+    table = rich.table.Table("Path", "Old Value", "New Value")
+    for name, param in sources.items():
+        old = targets[name]["Value"] if name in targets else None
+        new = replace(param["Value"])
+        table.add_row(
+            name,
+            rich.text.Text(old, style="red")
+            if old
+            else rich.text.Text("Not defined", style="bright_black italic"),
+            rich.text.Text(new, style="green"),
+        )
+    stdout.print(table)
+    confirmed = typer.confirm("Are you sure you want to apply the above changes?")
+    if not confirmed:
+        stderr.print("No changes applied", style="yellow italic")
+        sys.exit(1)
+
+
 if __name__ == "__main__":
     app()