Finish copy command

poetry.lock

@@ -42,17 +42,17 @@ uvloop = ["uvloop (>=0.15.2)"]
 
 [[package]]
 name = "boto3"
-version = "1.33.5"
+version = "1.33.6"
 description = "The AWS SDK for Python"
 optional = false
 python-versions = ">= 3.7"
 files = [
-    {file = "boto3-1.33.5-py3-none-any.whl", hash = "sha256:fcc24f62a1f512dd9b4a7a8af6f5fbfb3d69842a92aa2e79c2ca551ac49a4757"},
-    {file = "boto3-1.33.5.tar.gz", hash = "sha256:6a1d938bbf11518b1d17ca8186168f3ba2a0e8b2bf3c82cdd810ecb884627d2a"},
+    {file = "boto3-1.33.6-py3-none-any.whl", hash = "sha256:b88f0f305186c5fd41f168e006baa45b7002a33029aec8e5bef373237a172fca"},
+    {file = "boto3-1.33.6.tar.gz", hash = "sha256:4f62fc1c7f3ea2d22917aa0aa07b86f119abd90bed3d815e4b52fb3d84773e15"},
 ]
 
 [package.dependencies]
-botocore = ">=1.33.5,<1.34.0"
+botocore = ">=1.33.6,<1.34.0"
 jmespath = ">=0.7.1,<2.0.0"
 s3transfer = ">=0.8.2,<0.9.0"
 
@@ -61,13 +61,13 @@ crt = ["botocore[crt] (>=1.21.0,<2.0a0)"]
 
 [[package]]
 name = "botocore"
-version = "1.33.5"
+version = "1.33.6"
 description = "Low-level, data-driven core of boto 3."
 optional = false
 python-versions = ">= 3.7"
 files = [
-    {file = "botocore-1.33.5-py3-none-any.whl", hash = "sha256:c165207fb33e8352191d6a2770bce9f9bf01c62f5149824c4295d7f49bf96746"},
-    {file = "botocore-1.33.5.tar.gz", hash = "sha256:aa4a5c7cf78a403280e50daba8966479e23577b4a5c20165f71fab7a9b405e99"},
+    {file = "botocore-1.33.6-py3-none-any.whl", hash = "sha256:14282cd432c0683770eee932c43c12bb9ad5730e23755204ad102897c996693a"},
+    {file = "botocore-1.33.6.tar.gz", hash = "sha256:938056bab831829f90e09ecd70dd6b295afd52b1482f5582ee7a11d8243d9661"},
 ]
 
 [package.dependencies]
@@ -230,13 +230,13 @@ files = [
 
 [[package]]
 name = "platformdirs"
-version = "4.0.0"
+version = "4.1.0"
 description = "A small Python package for determining appropriate platform-specific dirs, e.g. a \"user data dir\"."
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.8"
 files = [
-    {file = "platformdirs-4.0.0-py3-none-any.whl", hash = "sha256:118c954d7e949b35437270383a3f2531e99dd93cf7ce4dc8340d3356d30f173b"},
-    {file = "platformdirs-4.0.0.tar.gz", hash = "sha256:cb633b2bcf10c51af60beb0ab06d2f1d69064b43abf4c185ca6b28865f3f9731"},
+    {file = "platformdirs-4.1.0-py3-none-any.whl", hash = "sha256:11c8f37bcca40db96d8144522d925583bdb7a31f7b0e37e3ed4318400a8e2380"},
+    {file = "platformdirs-4.1.0.tar.gz", hash = "sha256:906d548203468492d432bcb294d4bc2fff751bf84971fbb2c10918cc206ee420"},
 ]
 
 [package.extras]

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,27 @@ 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]:
+    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,
+        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 +68,8 @@ def profiles() -> None:
         print(profile)
 
 
-@app.command()
-def list(
+@app.command("ls")
+def list_parameters(
     profile: ProfileArg,
     path: pathlib.Path,
     recursive: bool = True,
@@ -63,21 +85,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 +140,68 @@ def unset(profile: ProfileArg, path: str) -> None:
         sys.exit(1)
 
 
+@app.command("cp")
+def copy_tree(
+    source_profile: ProfileArg,
+    source_path: str,
+    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]]:
+        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 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(dest_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)
+    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__":
     app()