Ansible Hetzner inventory

hetzner.py

#!/usr/bin/env python
# coding: utf-8

# Copyright (c) 2016-2017 eyeo GmbH
#
# This is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This software is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with the software. If not, see <http://www.gnu.org/licenses/>.

'''a dynamic Ansible inventory source based on the Hetzner API; refer to
https://robot.your-server.de/doc/webservice/en.html for more information'''

import argparse
import base64
import json
import os
import sys

try:
    from inspect import getfullargspec as getargspec
except ImportError:
    from inspect import getargspec

try:
    from urllib.request import Request, urlopen
except ImportError:
    from urllib2 import Request, urlopen


def get_data(path, **options):
    '''query any custom path for JSON data'''

    endpoint = options.get('endpoint') or 'https://robot-ws.your-server.de/'
    userinfo = options.get('userinfo') or 'changeme'

    request = Request(str(endpoint) + str(path))

    # https://stackoverflow.com/questions/2407126/
    request_authorization_token = base64.b64encode(userinfo)
    request_authorization = 'Basic {}'.format(request_authorization_token)
    request.add_header('Authorization', request_authorization)

    record = urlopen(request)
    result = json.load(record)
    return result


def get_host(inventory_hostname, **options):
    '''print a single host's record'''
    host_list = get_list(**options)
    return host_list['_meta']['hostvars'][inventory_hostname]


def get_list(**options):
    '''print a list of all host records'''
    server_objects = get_data('/server', **options)

    # https://robot.your-server.de/doc/webservice/en.html#server
    server_decoder = lambda record_envelope: record_envelope['server']
    server_records = map(server_decoder, server_objects)

    # remove servers that are considered inactive, e.g. the ones without a name
    server_has_a_name = lambda record: record.get('server_name', '') != ''
    server_records = filter(server_has_a_name, server_records)

    # remove servers that haven't been made available by Hetzner yet
    server_is_ready = lambda record: record.get('status') == 'ready'
    server_records = filter(server_is_ready, server_records)

    def transform(record):
        server = dict()
        server['ansible_host'] = record['server_ip']
        server['hetzner_server_number'] = record['server_number']
        return record['server_name'], server

    meta = dict(hostvars=dict(map(transform, server_records)))
    result = dict(_meta=meta)
    result[options.get('group')] = meta['hostvars'].keys()
    return result


def _action(callback):
    '''create an argparse.Action derivate around the given callback
    function, using *args values, **kwargs options, and dumping results'''

    def __init__(self, option_strings, dest, **kwargs):
        '''https://docs.python.org/library/argparse.html#action-classes'''
        kwargs.setdefault('help', callback.__doc__.strip())
        kwargs.setdefault('nargs', len(getargspec(callback).args))
        argparse.Action.__init__(self, option_strings, dest, **kwargs)

    def __call__(self, parser, namespace, values, option_string=None):
        '''https://docs.python.org/library/argparse.html#action'''
        kwargs = vars(namespace)
        kwargs = dict((k, v) for k, v in kwargs.iteritems() if v is not None)
        result = callback(*list(values), **kwargs)
        # setattr(namespace, self.dest, result)
        json.dump(result, sys.stdout, indent=2)
        sys.stdout.write('\n')

    cls = type(callback.__name__, (argparse.Action,), locals())
    return cls


if __name__ == '__main__':

    # https://docs.python.org/library/argparse.html
    argument_parser = argparse.ArgumentParser(description=__doc__)
    argument_parser.add_argument(
        '-G',
        help='the group to associate the hosts --list with',
        default=os.getenv('HETZNER_GROUP') or os.path.basename(sys.argv[0]),
        dest='group',
        metavar='$HETZNER_GROUP',
    )

    # https://robot.your-server.de/doc/webservice/en.html#url
    argument_parser.add_argument(
        '-R',
        help='the URL of the Hetzner robot API service',
        default=os.getenv('HETZNER_URL'),
        dest='endpoint',
        metavar='$HETZNER_URL',
    )

    # https://robot.your-server.de/doc/webservice/en.html#general
    argument_parser.add_argument(
        '-U',
        help='the username:password to authenticate with',
        default=os.getenv('HETZNER_USERINFO'),
        dest='userinfo',
        metavar='$HETZNER_USERINFO',
    )

    argument_parser.add_argument('--path', action=_action(get_data))
    argument_parser.add_argument('--host', action=_action(get_host))
    argument_parser.add_argument('--list', action=_action(get_list))
    argument_parser.parse_args()