Finish copy command

This commit is contained in:
Correl Roush 2024-01-23 15:06:29 -05:00
parent 08f55627f2
commit 7e023a5f08

29
ssm.py
View file

@ -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__":