ansible-playable/helpers/AnsibleAPI.py

#!/usr/bin/env python

import os
import sys
import stat
from collections import namedtuple

from ansible.cli import CLI
from ansible.parsing.dataloader import DataLoader
from ansible.inventory import Inventory
from ansible.executor.playbook_executor import PlaybookExecutor
from ansible.playbook import Playbook

from ansible.playbook.block import Block
from ansible.playbook.play_context import PlayContext
from ansible.utils.vars import load_extra_vars
from ansible.utils.vars import load_options_vars
from ansible.vars import VariableManager

import optparse

import json
import argparse

parser = None

def host_vars():
    parser.add_argument('--inventory_file', help='Inventory Filename', required=True)
    parser.add_argument('--host_name', help='Host or Group name', required=True)
    args= parser.parse_args()

    variable_manager = VariableManager()
    loader = DataLoader()

    inventory = Inventory(loader=loader, variable_manager=variable_manager, host_list=args.inventory_file)
    result = None
    if args.host_name in variable_manager.__getstate__()['host_vars_files']:
        result = variable_manager.__getstate__()['host_vars_files'][args.host_name]
    elif args.host_name in variable_manager.__getstate__()['group_vars_files']:
        result = variable_manager.__getstate__()['group_vars_files'][args.host_name]
    elif args.host_name:
        sys.stderr.write('\nHost %s not found in host vars or group vars files' % args.host_name)
        sys.exit(1)

    print(json.dumps(result))


def play_vars():
    parser.add_argument('--playbook_path', help='Playbook path', required=True)
    parser.add_argument('--play_name', help='Play Name', required=True)
    args = parser.parse_args()

    variable_manager = VariableManager()
    loader = DataLoader()

    # host12_vars = variable_manager.get_vars(loader=loader, host=inventory.parser.groups[args.host_name])

    playbook_path = args.playbook_path

    if not os.path.exists(playbook_path):
        print('[ERROR] The playbook does not exist')
        sys.exit()
    #
    # Options = namedtuple('Options', ['listtags', 'listtasks', 'listhosts', 'syntax', 'connection','module_path', 'forks', 'remote_user', 'private_key_file', 'ssh_common_args', 'ssh_extra_args', 'sftp_extra_args', 'scp_extra_args', 'become', 'become_method', 'become_user', 'verbosity', 'check'])
    # options = Options(listtags=False, listtasks=False, listhosts=False, syntax=False, connection='ssh', module_path=None, forks=100, remote_user='slotlocker', private_key_file=None, ssh_common_args=None, ssh_extra_args=None, sftp_extra_args=None, scp_extra_args=None, become=True, become_method=None, become_user='root', verbosity=None, check=False)
    #
    # # variable_manager.extra_vars = {'hosts': 'mywebserver'} # This can accomodate various other command line arguments.`
    #
    # passwords = {}
    #
    pb = Playbook.load(playbook_path, variable_manager=variable_manager, loader=loader)
    plays = pb.get_plays()
    for play in plays:
        if play._included_path is not None:
            loader.set_basedir(play._included_path)
        else:
            loader.set_basedir(pb._basedir)
        result = variable_manager.get_vars(loader=loader, play=play)

    # pbex = PlaybookExecutor(playbooks=[playbook_path], inventory=inventory, variable_manager=variable_manager, loader=loader, options=options, passwords=passwords)

    # results = pbex.run()

    print(json.dumps(result))

def role_vars():
    parser.add_argument('--playbook_path', help='Test Playbook path for role - usually role/tests/test.yml', required=True)
    parser.add_argument('--vault_password_file', help='Vault password file - usually role/tests/test.yml')
    args = parser.parse_args()

    variable_manager = VariableManager()
    loader = DataLoader()

    if args.vault_password_file:
        # read vault_pass from a file
        vault_pass = CLI.read_vault_password_file(args.vault_password_file, loader=loader)
        loader.set_vault_password(vault_pass)

    playbook_path = args.playbook_path

    if not os.path.exists(playbook_path):
        print('[ERROR] The playbook does not exist')
        sys.exit(1)

    pb = Playbook.load(playbook_path, variable_manager=variable_manager, loader=loader)
    plays = pb.get_plays()

    if len(plays) == 0:
        print('[ERROR] No plays in playbook')
        sys.exit(1)

    first_play = plays[0]
    if first_play._included_path is not None:
        loader.set_basedir(first_play._included_path)
    else:
        loader.set_basedir(pb._basedir)

    result = dict()
    for role in first_play.roles:
        result.update(role.get_default_vars())
        result.update(role.get_vars())

    print(json.dumps(result))


def parse(self):
    # create parser for CLI options
    parser = CLI.base_parser(
        usage="%prog playbook.yml",
        connect_opts=True,
        meta_opts=True,
        runas_opts=True,
        subset_opts=True,
        check_opts=True,
        inventory_opts=True,
        runtask_opts=True,
        vault_opts=True,
        fork_opts=True,
        module_opts=True,
    )

    # ansible playbook specific opts
    parser.add_option('--list-tasks', dest='listtasks', action='store_true',
                      help="list all tasks that would be executed")
    parser.add_option('--list-tags', dest='listtags', action='store_true',
                      help="list all available tags")
    parser.add_option('--step', dest='step', action='store_true',
                      help="one-step-at-a-time: confirm each task before running")
    parser.add_option('--start-at-task', dest='start_at_task',
                      help="start the playbook at the task matching this name")

    self.options, self.args = parser.parse_args(self.args[1:])

    self.parser = parser

    if len(self.args) == 0:
        raise Exception("You must specify a playbook file to run")

    # display.verbosity = self.options.verbosity
    self.validate_conflicts(runas_opts=True, vault_opts=True, fork_opts=True)

def list_tags():
    # ansible playbook specific opts
    # parser.add_argument('--list-tasks', dest='listtasks', action='store_true',
    #                   help="list all tasks that would be executed")
    # parser.add_argument('--list-tags', dest='listtags', action='store_true',
    #                   help="list all available tags")
    # parser.add_argument('--step', dest='step', action='store_true',
    #                   help="one-step-at-a-time: confirm each task before running")
    # parser.add_argument('--start-at-task', dest='start_at_task',
    #                   help="start the playbook at the task matching this name")
    #
    # parser.add_argument('--inventory_file', help='Inventory Filename', required=True)
    # parser.add_argument('--playbook_path', help='Inventory Filename', required=True)

    parser = CLI.base_parser(
        usage="%prog playbook.yml",
        connect_opts=True,
        meta_opts=True,
        runas_opts=True,
        subset_opts=True,
        check_opts=True,
        inventory_opts=True,
        runtask_opts=True,
        vault_opts=True,
        fork_opts=True,
        module_opts=True,
    )

    # ansible playbook specific opts
    parser.add_option('--list-tasks', dest='listtasks', action='store_true',
                      help="list all tasks that would be executed")
    parser.add_option('--list-tags', dest='listtags', action='store_true',
                      help="list all available tags")
    parser.add_option('--step', dest='step', action='store_true',
                      help="one-step-at-a-time: confirm each task before running")
    parser.add_option('--start-at-task', dest='start_at_task',
                      help="start the playbook at the task matching this name")

    options, args = parser.parse_args(args[1:])

    # options = parser.parse_args()

    inventory_path = options.inventory_file
    playbook_path = options.playbook_path

    if not os.path.exists(playbook_path):
        print('[ERROR] The playbook does not exist')
        sys.exit(1)

    loader = DataLoader()

    # initial error check, to make sure all specified playbooks are accessible
    # before we start running anything through the playbook executor
    # for playbook in playbooks:
    #     if not os.path.exists(playbook):
    #         raise Exception("the playbook: %s could not be found" % playbook)
    #     if not (os.path.isfile(playbook) or stat.S_ISFIFO(os.stat(playbook).st_mode)):
    #         raise Exception("the playbook: %s does not appear to be a file" % playbook)

    # create the variable manager, which will be shared throughout
    # the code, ensuring a consistent view of global variables
    variable_manager = VariableManager()
    # variable_manager.extra_vars = load_extra_vars(loader=loader, options=options)

    # variable_manager.options_vars = load_options_vars(options)

    # create the inventory, and filter it based on the subset specified (if any)
    inventory = Inventory(loader=loader, variable_manager=variable_manager, host_list=inventory_path)
    variable_manager.set_inventory(inventory)

    # (which is not returned in list_hosts()) is taken into account for
    # warning if inventory is empty.  But it can't be taken into account for
    # checking if limit doesn't match any hosts.  Instead we don't worry about
    # limit if only implicit localhost was in inventory to start with.
    #
    # Fix this when we rewrite inventory by making localhost a real host (and thus show up in list_hosts())
    no_hosts = False
    if len(inventory.list_hosts()) == 0:
        # Empty inventory
        print("provided hosts list is empty, only localhost is available")
        no_hosts = True
    # inventory.subset(options.subset)
    if len(inventory.list_hosts()) == 0 and no_hosts is False:
        # Invalid limit
        raise Exception("Specified --limit does not match any hosts")

    # flush fact cache if requested
    # if options.flush_cache:
    #     for host in inventory.list_hosts():
    #         variable_manager.clear_facts(host)

    passwords = {}

    # create the playbook executor, which manages running the plays via a task queue manager
    pbex = PlaybookExecutor(playbooks=[playbook_path], inventory=inventory, variable_manager=variable_manager,
                            loader=loader, options=options, passwords=passwords)

    results = pbex.run()

    if isinstance(results, list):
        for p in results:

            print('\nplaybook: %s' % p['playbook'])
            for idx, play in enumerate(p['plays']):
                msg = "\n  play #%d (%s): %s" % (idx + 1, ','.join(play.hosts), play.name)
                mytags = set(play.tags)
                msg += '\tTAGS: [%s]' % (','.join(mytags))

                if options.listhosts:
                    playhosts = set(inventory.get_hosts(play.hosts))
                    msg += "\n    pattern: %s\n    hosts (%d):" % (play.hosts, len(playhosts))
                    for host in playhosts:
                        msg += "\n      %s" % host

                print(msg)

                all_tags = set()
                if options.listtags or options.listtasks:
                    taskmsg = ''
                    if options.listtasks:
                        taskmsg = '    tasks:\n'

                    def _process_block(b):
                        taskmsg = ''
                        for task in b.block:
                            if isinstance(task, Block):
                                taskmsg += _process_block(task)
                            else:
                                if task.action == 'meta':
                                    continue

                                all_tags.update(task.tags)
                                if options.listtasks:
                                    cur_tags = list(mytags.union(set(task.tags)))
                                    cur_tags.sort()
                                    if task.name:
                                        taskmsg += "      %s" % task.get_name()
                                    else:
                                        taskmsg += "      %s" % task.action
                                    taskmsg += "\tTAGS: [%s]\n" % ', '.join(cur_tags)

                        return taskmsg

                    all_vars = variable_manager.get_vars(loader=loader, play=play)
                    play_context = PlayContext(play=play, options=options)
                    for block in play.compile():
                        block = block.filter_tagged_tasks(play_context, all_vars)
                        if not block.has_tasks():
                            continue
                        taskmsg += _process_block(block)

                    if options.listtags:
                        cur_tags = list(mytags.union(all_tags))
                        cur_tags.sort()
                        taskmsg += "      TASK TAGS: [%s]\n" % ', '.join(cur_tags)

                    print(taskmsg)

def main():
    global parser
    parser = argparse.ArgumentParser()
    parser.add_argument('command', help='Command Type - [host_vars, play_vars]')
    args, unknown = parser.parse_known_args()

    if args.command == 'host_vars':
        host_vars()
    elif args.command == 'play_vars':
        play_vars()
    elif args.command == 'role_vars':
        role_vars()
    elif args.command == 'list':
        list_tags()


if __name__ == '__main__':
    main()