Script for dumping RabbitMQ definitions with deterministic sorting

rmq-sorted-definitions.py

#!/usr/bin/env python3
import argparse
import json
import requests
import sys


def dict_list_key(item):
    if 'vhost' in item and 'name' in item:
        return item['vhost'], item['name']
    elif 'vhost' in item and 'source' in item and 'destination' in item:
        return item['vhost'], item['source'], item['destination']
    for key in {'name', 'user'}:
        if key in item:
            return item[key]
    return item


def nested_sort(value):
    if isinstance(value, dict):
        return dict([(k, nested_sort(v)) for k, v in value.items()])
    elif isinstance(value, list):
        if all([isinstance(i, dict) for i in value]):
            return sorted(value, key=dict_list_key)
        return sorted(value)
    return value


def main():
    parser = argparse.ArgumentParser(
        'rmq-sorted-definitions',
        description='Provides deterministic sorting for RabbitMQ definitions '
                    'provided by the Management API')
    parser.add_argument(
        '--url', default='http://localhost:15672',
        help='The RabbitMQ Management API base URL. Default: %(default)s')
    parser.add_argument(
        '-u', '--username', default='guest',
        help='The RabbitMQ Management API username. Default: %(default)s')
    parser.add_argument(
        '-p', '--password', default='guest',
        help='The RabbitMQ Management API password. Default: %(default)s')
    parser.add_argument(
        'file', default=sys.stdout, type=argparse.FileType('w'), nargs='?',
        metavar='DESTINATION',
        help='Location to write the definitions to. Default: STDOUT')
    args = parser.parse_args()

    try:
        response = requests.get(
            '{}/api/definitions'.format(args.url.rstrip('/')),
            auth=(args.username, args.password))
    except requests.RequestException as error:
        sys.stderr.write('ERROR: {}'.format(error))
        sys.exit(1)

    if not response.ok:
        error = response.json()
        sys.stderr.write(
            'Error: {}\n'.format(error.get('reason', error.get('error'))))
        sys.exit(1)

    with args.file as handle:
        json.dump(nested_sort(response.json()), handle,
                  sort_keys=True, indent=2)
        handle.write('\n')


if __name__ == '__main__':
    main()