参考

AnsibleAPI 开发
Python3 调用 Ansible2.x API

ansible 封装基础类

import json
import shutil
from ansible.module_utils.common.collections import ImmutableDict  #用于添加选项。比如: 指定远程用户remote_user=None
from ansible.parsing.dataloader import DataLoader                  #读取 json/ymal/ini 格式的文件的数据解析器
from ansible.vars.manager import VariableManager                   #管理主机和主机组的变量管理器
from ansible.inventory.manager import InventoryManager             #管理资源库的，可以指定一个 inventory 文件等
from ansible.playbook.play import Play                             #用于执行 Ad-hoc 的类 ,需要传入相应的参数
from ansible.executor.task_queue_manager import TaskQueueManager   #ansible 底层用到的任务队列管理器
from ansible.plugins.callback import CallbackBase                  #处理任务执行后返回的状态,用于处理执行结果的。 后面我们可以继承改写这个类用作回调插件，以便满足我们的需求。
from ansible import context                                        #上下文管理器，他就是用来接收 ImmutableDict 的示例对象
import ansible.constants as C                                      #用于获取 ansible 产生的临时文档
from ansible.executor.playbook_executor import PlaybookExecutor  # 执行 playbook 的核心类
from ansible.inventory.host import Group                           #对 主机组 执行操作 ，可以给组添加变量等操作，扩展
from ansible.inventory.host import Host                            #对 主机 执行操作 ，可以给主机添加变量等操作，扩展


import time
import datetime

class ResultCallback(CallbackBase):
    """
    重写callbackBase类的部分方法,可以获得执行后返回的结果和状态
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        #参考的博客中原本定义的，信息不准确,{ip:result}
        #以ip为key，result信息为value，存进字典中，但如果执行了多个任务只能收集到最后执行的任务的结果，之前的执行信息会被覆盖,所以后面我改了下收集信息的方式
        self.host_ok = {}
        self.host_unreachable = {}
        self.host_failed = {}

        #重新定义的收集结果信息的方法，可以收集到所有的执行信息，按执行的顺序放入列表，但缺点是没有按ip来进行分类
        self.host_all_info=[]   #存放所有的执行结果信息的列表[{reslut1},{reslut2}]
        self.host_failed_iplist=[]
        self.host_unreachabled_iplist=[]

    #机器不可达的时候,result参数里是执行的结果信息等
    def v2_runner_on_unreachable(self, result):
        #以ip为key，result信息为value，存进字典中，但如果执行了多个任务只能收集到最后执行的任务的结果，所以后面我改了下收集信息的方式
        self.host_unreachable[result._host.get_name()] = result

        exec_time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')  #执行时间
        host_ip=result._host.get_name()    #主机ip
        task_name=result.task_name         #任务名称
        result_info=result._result         #执行的结果信息

        #获取任务执行的相关信息，放入字典
        result_dict={"exec_time":exec_time,"host_ip":host_ip,"task_name":task_name,"status":"unreachable","result_info":result_info}
        #print(result_dict)
        #然后再讲字典放入存放全部信息的列表中
        self.host_all_info.append(result_dict)

    #任务成功的时候回调函数
    #在ad-hoc下每个ip成功，都会执行这个函数，而在playbook状态下，每个action都会执行这个函数，例如，全部ip第一个action执行完才会继续执行下一个action
    def v2_runner_on_ok(self, result, **kwargs):
        # 以ip为key，result信息为value，存进字典中，但如果执行了多个任务只能收集到最后执行的任务的结果，所以后面我改了下收集信息的方式
        self.host_ok[result._host.get_name()] = result


        exec_time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        host_ip=result._host.get_name()
        task_name=result.task_name
        result_info=result._result
        # 获取任务执行的相关信息，放入字典
        result_dict={"exec_time":exec_time,"host_ip":host_ip,"task_name":task_name,"status":"success","result_info":result_info}
        #print(result_dict)
        # 然后再讲字典放入存放全部信息的列表中
        self.host_all_info.append(result_dict)


    #任务执行失败时的回调函数
    def v2_runner_on_failed(self, result, ignore_errors=False,**kwargs):

        self.host_failed[result._host.get_name()] = result

        exec_time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        host_ip=result._host.get_name()
        task_name=result.task_name
        result_info=result._result
        result_dict={"exec_time":exec_time,"host_ip":host_ip,"task_name":task_name,"status":"failed","result_info":result_info}
        #print(result_dict)
        self.host_all_info.append(result_dict)


        # print("failed:",end="")
        # print(result._host.get_name(),end=" ")
        # print(result.task_name,end=" ")
        # print(result._result)

    #任务被skip时的回调函数
    def v2_runner_on_skipped(self, result,**kwargs):
        exec_time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        host_ip=result._host.get_name()
        task_name=result.task_name
        result_info=result._result
        result_dict={"exec_time":exec_time,"host_ip":host_ip,"task_name":task_name,"status":"skip","result_info":result_info}
        #print(result_dict)
        self.host_all_info.append(result_dict)

    #task中每个action都会执行这个函数，可以用task.name获取action的名称,只执行一次
    def v2_playbook_on_task_start(self, task, is_conditional):
        pass
        #print(task.name)
        #print(task._uuid)

    def v2_playbook_on_play_start(self, play):
        pass
        #print(play.name)

    #获取playbook执行的状态，在playbook执行完后执行该函数，可以获取该playbook中执行的主机，以及每个主机执行该playbook中action成功失败的个数
    def v2_playbook_on_stats(self, stats):
        #print(stats.processed)   #{'10.104.114.197': 1, '10.104.113.129': 1, '10.104.113.130': 1}
        #print(stats.processed.keys())             #dict_keys(['10.104.114.197', '10.104.113.129', '10.104.113.130'])
        hosts = sorted(stats.processed.keys())
        for h in hosts:
            t = stats.summarize(h)                #{'ok': 1, 'failures': 0, 'unreachable': 0, 'changed': 0, 'skipped': 0, 'rescued': 0, 'ignored': 0}
            if t["failures"]!=int(0):
                self.host_failed_iplist.append(h)
            if t["unreachable"]!=int(0):
                self.host_unreachabled_iplist.append(h)
            #print(t)

    # task中每个action都会执行这个函数，可以用task.name获取action的名称，每个ip都执行一次
    def v2_runner_on_start(self, host, task):
        pass
        #print(task.name)


    # def v2_playbook_on_stats(self, stats):
    #     print(stats.processed.keys())
    #     print(stats.processed)
    #     print(sorted(stats.processed.keys()))
    #     print(stats.summarize(sorted(stats.processed.keys())[0]))

    # def v2_playbook_on_play_start(self, play):
    #     print(play.name)


class MyAnsiable():

    #自定义类的一些初始化信息,在下面的context.CLIARGS初始化函数中使用这些初始化属性
    # 在初始化的这个类时候可以传参，以便覆盖默认选项的值,我们可以自己传参数，否则使用定义的默认值
    def __init__(self,
                 connection='ssh',  # 连接方式 local 本地方式，smart ssh方式
                 remote_user="None",  # 远程用户
                 ack_pass=None,  # 提示输入密码
                 sudo=None, sudo_user=None, ask_sudo_pass=None,
                 module_path=None,  # 模块路径，可以指定一个自定义模块的路径
                 become=None,  # 是否提权
                 become_method=None,  # 提权方式 默认 sudo 可以是 su
                 become_user=None,  # 提权后，要成为的用户，并非登录用户
                 check=False, diff=False,
                 listhosts=None, listtasks=None, listtags=None,
                 forks=5,      #同时执行的主机数量
                 tags=[],      #执行的tags列表
                 skip_tags=[], #skip跳过的tags列表
                 verbosity=3,
                 syntax=None,
                 start_at_task=None,
                 inventory=None,
                 passwords=None):


        #上下文管理器，用来接收 ImmutableDict 的示例对象,ImmutableDict用于添加选项
        #2.7 和 2.8 版本有一些差异： 2.7 使用了 Python 标准库里的 命名元组来初始化选项，而 2.8 是 Ansible 自己封装了一个 ImmutableDict,要和 context 结合使用的
        context.CLIARGS = ImmutableDict(
            connection=connection,
            remote_user=remote_user,
            ack_pass=ack_pass,
            sudo=sudo,
            sudo_user=sudo_user,
            ask_sudo_pass=ask_sudo_pass,
            module_path=module_path,
            become=become,
            become_method=become_method,
            become_user=become_user,
            verbosity=verbosity,
            listhosts=listhosts,
            listtasks=listtasks,
            listtags=listtags,
            forks=forks,
            tags=tags,
            skip_tags=skip_tags,
            syntax=syntax,
            start_at_task=start_at_task,
        )

        # 三元表达式，假如没有传递 inventory文件, 就使用 "localhost,"
        # 这里需要注意如果不使用host文件，即inventory未传入，直接动态传入host列表的情况，这里会将localhost加入主机中，所以执行all主机组也会执行本机，所以要么改为未传入就为“”，要么就设置其他的主机组名称，不使用all执行全部
        # inventory 就是平时用到的存放主机ip以及变量的资源库文件，-i 参数后面跟的文件
        self.inventory = inventory if inventory else "localhost,"

        # 实例化数据解析器,用于解析 存放主机列表的资源库文件 （比如： /etc/ansible/hosts） 中的数据和变量数据的
        self.loader = DataLoader()

        # 实例化 资产配置对象,InventoryManager管理资源库，可以指定一个loader数据解析器和一个inventory文件
        self.inv_obj = InventoryManager(loader=self.loader, sources=self.inventory)

        # 设置密码，可以为空字典，但必须有此参数
        self.passwords = {"conn_pass":passwords}

        # 实例化回调插件对象
        self.results_callback = ResultCallback()

        # 变量管理器,假如有变量，所有的变量应该交给他管理。 这里他会从 inventory 对象中获取到所有已定义好的变量。 这里也需要数据解析器。
        self.variable_manager = VariableManager(self.loader, self.inv_obj)


    #用来执行 Ad-hoc 的方法
    def run(self, hosts='all', gether_facts="no", module="ping", args=''):

        #定义一个字典，来设置主机组、是否获取机器信息、以及执行的模块及参数,
        #参数可以在执行run()函数的时候传入
        play_source = dict(
            name="Ad-hoc",
            hosts=hosts,
            gather_facts=gether_facts,
            tasks=[
                # 这里每个 task 就是这个列表中的一个元素，格式是嵌套的字典
                # 也可以作为参数传递过来，这里就简单化了。
                {"action": {"module": module, "args": args}},
            ])

        #Play()是用于执行 Ad-hoc 的类 ,这里传入一个上面的play_source字典参数 VariableManager变量管理器  DataLoader数据解析器
        play = Play().load(play_source, variable_manager=self.variable_manager, loader=self.loader)

        # 先定义一个值，防止代码出错后，   `finally` 语句中的 `tqm` 未定义。
        tqm = None
        try:
            #TaskQueueManager是底层用到的任务队列管理器
            #要想执行 Ad-hoc ，需要把上面的 play 对象交给任务队列管理器的 run 方法去运行
            tqm = TaskQueueManager(
                inventory=self.inv_obj,
                variable_manager=self.variable_manager,
                loader=self.loader,
                passwords=self.passwords,
                stdout_callback=self.results_callback)

            result = tqm.run(play)
        finally:
            if tqm is not None:
                tqm.cleanup()
            shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)

    #用来执行 playbook 的方法
    def playbook(self, playbooks,extra_vars={}):

        #自定义变量是用一个字典保存的
        #使用变量管理器VariableManager().extra_vars.update(extra_vars_dict)的方法进行添加
        self.variable_manager.extra_vars.update(extra_vars)

        #PlaybookExecutor是执行playbook 的核心类,这里实例化一个PlaybookExecutor对象
        playbook = PlaybookExecutor(playbooks=playbooks,  #playbook yaml文件列表,注意这里是一个列表
                                    inventory=self.inv_obj,  # InventoryManager资产配置对象
                                    variable_manager=self.variable_manager,  #VariableManager变量管理器
                                    loader=self.loader,     #DataLoader数据解析器
                                    passwords=self.passwords)   #密码

        # 使用回调函数
        playbook._tqm._stdout_callback = self.results_callback

        #执行PlaybookExecutor类的run方法
        result = playbook.run()

    #这个获取执行状态和信息的函数是不准确的，如果一个playbook中有多个action，
    #用ip做key，key值是唯一的，但value可以改变，获取到的信息只能是最后一个，所以不准确
    def get_result(self):
        result_raw = {'success': {}, 'failed': {}, 'unreachable': {}}

        # print(self.results_callback.host_ok)
        for host, result in self.results_callback.host_ok.items():
            result_raw['success'][host] = result._result
        for host, result in self.results_callback.host_failed.items():
            result_raw['failed'][host] = result._result
        for host, result in self.results_callback.host_unreachable.items():
            result_raw['unreachable'][host] = result._result
        return result_raw

        # 最终打印结果，并且使用 JSON 继续格式化
        #print(json.dumps(result_raw, indent=4))


    #动态添加主机，传入主机列表和组名
    def add_dynamic_hosts(self, hostip_list, groupname=None, groupvars=None):
        """
            add hosts to a group
        """
        #如果有传入组名，则添加组，并创建Group实例
        if groupname:
            self.inv_obj.add_group(groupname)
            my_group = Group(name=groupname)
        #如果有传入组变量，则将组变量设置给上面创建的Group实例
        if groupvars:
            for key, value in groupvars.iteritems():
                my_group.set_variable(key, value)

        # add hosts to group
        # 如果有传入组名，则遍历主机列表，添加主机并且设置组
        if groupname:
            for hostip in hostip_list:
                self.inv_obj.add_host(host=hostip,group=groupname)
        #如果没有传入组名，则遍历主机列表，只添加主机
        else:
            for hostip in hostip_list:
                self.inv_obj.add_host(host=hostip)

Ad-hoc 模式到 API 的映射

二、验证

示例一、验证ansible-plook 的执行

import sys
from myansible import MyAnsiable


if __name__ == '__main__':
    ##实例化
    ansible3 = MyAnsiable(inventory='/etc/ansible/hosts', connection='ssh',remote_user="root",forks=1,tags=["test1","test2","test3"],
                          passwords="")
    #传入playbooks 的参数，需要是一个列表的数据类型，这里是使用的相对路径
    #相对路径是相对于执行脚本的当前用户的家目录
    print("playbook指定tags以及传递自定义变量")
    ansible3.playbook(playbooks=['test_set_fact.yml'],extra_vars={"testvars":"testvars1"})
    print(ansible3.results_callback.host_unreachable)

cat >test_set_fact.yml<<EOF
---
- hosts: tx_151
  remote_user: root
  vars: 
    testvar1: test1
  tasks:
  - set_fact:
     testvar2: "test2"
  - debug:
      msg: "{{testvars}} {{testvar1}} {{testvar2}} "
EOF
ansible-playbook test_set_fact.yml -v

示例二、验证ansible的执行

if __name__ == '__main__':
    #实例化
    ansible1 = MyAnsiable(inventory='/etc/ansible/hosts', remote_user="root",connection='ssh',forks=1,passwords="")
    #执行 ad-hoc
    print("执行ad-hoc的结果")
    ansible1.run(hosts= "owner", module="shell", args='pwd')
    #打印结果
    print(ansible1.get_result())
    print(ansible1.results_callback.host_all_info)

实例三、添加group和host及定义主机变量

    host_list=["10.104.113.129","10.104.113.130"]
    ansible4 = MyAnsiable(connection='ssh',remote_user="root",forks=1,tags=["test4"],
                           passwords="123456")
    #创建一个组 remote
    ansible4.inv_obj.add_group('remote')
    #向 remote 组中添加主机
    for host in host_list:
        ansible4.inv_obj.add_host(host,group="remote")
        ansible4.inv_obj.get_host(host).vars['ansible_ssh_host'] = host

    print("playbook添加group和host及定义主机变量")
    ansible4.playbook(playbooks=['test.yml'])

    print(ansible4.get_result())
    print(ansible4.results_callback.host_all_info)

实例四、动态添加主机，传入主机列表和组名

 
    ansible5=MyAnsiable(connection='ssh',remote_user="root",forks=1,tags=["test4"],
                           passwords="123456")
    host_list = ["10.104.116.15"]

    #使用上面类中设置的add_dynamic_hosts方法
    ansible5.add_dynamic_hosts(hostip_list=host_list,groupname="remote")
    print("动态添加主机，传入主机列表和组名")
    ansible5.playbook(playbooks=["test.yml"])
    print(ansible5.get_result())
    print(ansible5.results_callback.host_all_info)

实例五 ignore_errors忽略错误测试

    ansible6=MyAnsiable(connection='ssh',remote_user="root",forks=1,tags=["ignore"],
                            passwords="123456")
    host_list = ["10.104.116.15"]
    ansible6.add_dynamic_hosts(hostip_list=host_list,groupname="remote")
    print("ignore_errors忽略错误测试")
    ansible6.playbook(playbooks=["test.yml"])
    print(ansible6.get_result())
    print(ansible6.results_callback.host_all_info)

test.yml

---
- hosts: owner
  remote_user: root
  sudo: true
  gather_facts: no
  tasks:
    - name: test1
      sudo: yes
      shell: id
      tags:
        - test1
    - name: test2
      sudo: yes
      shell: pwd
      tags:
        - test2
    - name: test3
      sudo: yes
      debug: msg={{ testvars }}
      tags:
        - test3
    - name: test4
      sudo: yes
      debug: msg={{ ansible_ssh_host }}
      tags:
        - test4

    - name: ignore_faild
      sudo: yes
      shell: "123"
      ignore_errors: True
      tags:
        - ignore