126 lines
3.2 KiB
Python
126 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
import typing
|
|
|
|
import boto3
|
|
import botocore.client
|
|
import botocore.session
|
|
import rich.console
|
|
import rich.table
|
|
import typer
|
|
|
|
app = typer.Typer()
|
|
stdout = rich.console.Console()
|
|
stderr = rich.console.Console(stderr=True)
|
|
aws_profiles = boto3.session.Session().available_profiles
|
|
|
|
|
|
class SSMPath(pathlib.PurePosixPath):
|
|
def __init__(self, *segments: str | os.PathLike):
|
|
super().__init__("/", *segments)
|
|
|
|
@staticmethod
|
|
def parse(value: str) -> SSMPath:
|
|
return SSMPath(value)
|
|
|
|
|
|
ProfileArg = typing.Annotated[str, typer.Argument(autocompletion=aws_profiles)]
|
|
|
|
|
|
def get_client(profile: str) -> botocore.client.BaseClient:
|
|
try:
|
|
session = boto3.Session(profile_name=profile)
|
|
return session.client("ssm")
|
|
except botocore.session.ProfileNotFound:
|
|
stderr.print(f"Invalid profile '{profile}'", style="bold red")
|
|
sys.exit(1)
|
|
|
|
|
|
@app.command()
|
|
def profiles() -> None:
|
|
"""List available AWS profiles."""
|
|
for profile in aws_profiles:
|
|
print(profile)
|
|
|
|
|
|
@app.command()
|
|
def list(
|
|
profile: ProfileArg,
|
|
path: pathlib.Path,
|
|
recursive: bool = True,
|
|
) -> None:
|
|
"""List parameters and their values at a requested PATH."""
|
|
root = SSMPath(path)
|
|
client = get_client(profile)
|
|
console = rich.console.Console()
|
|
table = rich.table.Table(
|
|
"Name",
|
|
"Value",
|
|
"Description",
|
|
"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"):
|
|
table.add_row(
|
|
str(pathlib.Path(name).relative_to(root)),
|
|
parameter.get("Value"),
|
|
parameter.get("Description"),
|
|
parameter.get("Type"),
|
|
)
|
|
console.print(table)
|
|
|
|
|
|
@app.command()
|
|
def set(
|
|
profile: ProfileArg,
|
|
path: str,
|
|
value: str,
|
|
secure: bool = False,
|
|
overwrite: bool = False,
|
|
description: typing.Annotated[str, typer.Option] = "",
|
|
) -> None:
|
|
"""Set a parameter at PATH to VALUE.
|
|
|
|
If --secure is used, it will be stored as a SecureString.
|
|
"""
|
|
client = get_client(profile)
|
|
try:
|
|
client.put_parameter(
|
|
Name=str(SSMPath(path)),
|
|
Value=value,
|
|
Type="SecureString" if secure else "String",
|
|
Overwrite=overwrite,
|
|
Description=description,
|
|
)
|
|
except client.exceptions.ParameterAlreadyExists:
|
|
stderr.print(
|
|
"Parameter already exists; use --overwrite to replace it.",
|
|
style="bold red",
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
@app.command()
|
|
def unset(profile: ProfileArg, path: str) -> None:
|
|
"""Remove a parameter at PATH."""
|
|
client = get_client(profile)
|
|
try:
|
|
client.delete_parameter(Name=str(SSMPath(path)))
|
|
except client.exceptions.ParameterNotFound:
|
|
stderr.print("Parameter not found", style="yellow")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app()
|