swf-domains.py

"""AWS SWF domains stuff."""
import argparse
import boto3


def get_swf_domains(client=boto3.client("swf"), **parameters):
    """Get AWS SWF domain names."""
    paginator = client.get_paginator("list_domains")

    for response in paginator.paginate(**parameters):
        for domain in response["domainInfos"]:
            yield domain["name"]


def list_domains(arguments):
    """Print AWS SWF domain names."""
    client = boto3.client("swf", region_name=arguments.region_name)
    parameters = {
        "registrationStatus": "REGISTERED",
        "PaginationConfig": {
            "MaxItems": arguments.max_items,
            "PageSize": arguments.page_size
        }
    }

    for domain_name in get_swf_domains(client, **parameters):
        print(domain_name)


def register_domain(arguments):
    """Register an AWS SWF domain name."""
    client = boto3.client("swf", region_name=arguments.region_name)
    parameters = {
        "name": arguments.name,
        "description": arguments.description,
        "workflowExecutionRetentionPeriodInDays": arguments.retention_period
    }

    client.register_domain(**parameters)


def parse_paginator_arguments(parser):
    """Add the paginator argument(s) to the argument parser."""
    parser.add_argument(
        "--max-items",
        help="Paginator max items",
        type=int,
        default=None
    )
    parser.add_argument(
        "--page-size",
        help="Paginator page size",
        type=int,
        default=None
    )


def parse_default_arguments(parser):
    """Add the default argument(s) to the argument parser."""
    parser.add_argument(
        "-r", "--region-name",
        help='AWS region name',
        default=None
    )


def parse_list_domains(subparsers):
    """Add the list-domain command."""
    parser = subparsers.add_parser('list-domains')

    parse_default_arguments(parser)
    parse_paginator_arguments(parser)

    parser.set_defaults(func=list_domains)


def parse_register_domain(subparsers):
    """Add the list-domain command."""
    parser = subparsers.add_parser('register-domain')

    parse_default_arguments(parser)
    parser.add_argument(
        "name",
        help='AWS SWF domain name'
    )
    parser.add_argument(
        "-d", "--description",
        help='AWS SWF domain description',
        default=None
    )
    parser.add_argument(
        "retention_period",
        help='AWS SWF workflow execution retention period in days',
        default="NONE"
    )

    parser.set_defaults(func=register_domain)


def parse_args():
    """Parse input arguments."""
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers()
    subparsers.required = True
    subparsers.dest = 'command'

    parse_list_domains(subparsers)
    parse_register_domain(subparsers)

    return parser.parse_args()


if __name__ == "__main__":
    arguments = parse_args()

    arguments.func(arguments)