python版本：Python 2.6.6

ansible版本：ansible 2.3.1.0

調用腳本： task_exec_v1.py

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

import os
import sys
import json
import logging
from collections import namedtuple
from ansible.inventory import Inventory
from ansible.vars import VariableManager
from ansible.parsing.dataloader import DataLoader
 
from ansible.executor.playbook_executor import PlaybookExecutor
from ansible.plugins.callback import CallbackBase
from ansible.errors import AnsibleParserError
from optparse import OptionParser

#定義打印日誌
logging.basicConfig(filename=‘task_exec_v3.log‘,               #通過logging.basicConfig函數對日誌的輸出格式及方式做相關配置
                    format
 
=‘%(asctime)s - %(name)s - %(levelname)s -%(module)s:  %(message)s‘,datefmt=‘%Y-%m-%d %H:%M:%S %p‘,
                    level=10)

class MyCallback(CallbackBase):
    #這裏是狀態回調，各種成功失敗的狀態,裏面的各種方法其實都是從寫於CallbackBase父類裏面的，其實還有很多，可以根據需要拿出來用
    def __init__(self,*args):
        super(MyCallback,self).__init__(display
 
=None)
        self.status_ok=json.dumps({})
        self.status_fail=json.dumps({})
        self.status_unreachable=json.dumps({})
        self.status_playbook=‘‘
        self.status_no_hosts=False
        self.host_ok = {}
        self.host_failed={}
        self.host_unreachable={}
    def v2_runner_on_ok(self,result):
        host=result._host.get_name()
        self.runner_on_ok(host, result._result)
        #self.status_ok=json.dumps({host:result._result},indent=4)
        self.host_ok[host] = result
    def v2_runner_on_failed(self, result, ignore_errors=False):
        host = result._host.get_name()
        self.runner_on_failed(host, result._result, ignore_errors)
        #self.status_fail=json.dumps({host:result._result},indent=4)
        self.host_failed[host] = result
    def v2_runner_on_unreachable(self, result):
        host = result._host.get_name()
        self.runner_on_unreachable(host, result._result)
        #self.status_unreachable=json.dumps({host:result._result},indent=4)
        self.host_unreachable[host] = result
    def v2_playbook_on_no_hosts_matched(self):
        self.playbook_on_no_hosts_matched()
        self.status_no_hosts=True
    def v2_playbook_on_play_start(self, play):
        self.playbook_on_play_start(play.name)
        self.playbook_path=play.name

class My_ansible_play():
    #這裏是ansible運行 
    #初始化各項參數，大部分都定義好，只有幾個參數是必須要傳入的
    def __init__(self, playbook, extra_vars={}, 
                        host_list=‘/etc/ansible/hosts‘, 
                        connection=‘ssh‘,
                        become=False,
                        become_user=None,
                        module_path=None,
                        fork=50,
                        ansible_cfg=None,   #os.environ["ANSIBLE_CONFIG"] = None
                        passwords={},
                        check=False):
        self.playbook_path=playbook
        self.passwords=passwords
        self.extra_vars=extra_vars
        Options = namedtuple(‘Options‘,
                   [‘listtags‘, ‘listtasks‘, ‘listhosts‘, ‘syntax‘, ‘connection‘,‘module_path‘,
                   ‘forks‘, ‘private_key_file‘, ‘ssh_common_args‘, ‘ssh_extra_args‘, ‘sftp_extra_args‘,
                      ‘scp_extra_args‘, ‘become‘, ‘become_method‘, ‘become_user‘, ‘verbosity‘, ‘check‘])
        logging.info(Options)
        self.options = Options(listtags=False, listtasks=False, 
                              listhosts=False, syntax=False, 
                              connection=connection, module_path=module_path, 
                              forks=fork, private_key_file=None, 
                              ssh_common_args=None, ssh_extra_args=None, 
                              sftp_extra_args=None, scp_extra_args=None, 
                              become=become, become_method=None, 
                              become_user=become_user, 
                              verbosity=None, check=check)
        logging.info(self.options)
        if ansible_cfg != None:
            os.environ["ANSIBLE_CONFIG"] = ansible_cfg
        self.variable_manager=VariableManager()
        self.variable_manager.extra_vars=self.extra_vars
        self.loader=DataLoader()
        self.inventory=Inventory(loader=self.loader,variable_manager=self.variable_manager,host_list=host_list)

    #定義運行的方法和返回值
    def run(self):
        complex_msg={}
        if not os.path.exists(self.playbook_path):
            code=1000
            results={‘playbook‘:self.playbook_path,‘msg‘:self.playbook_path+‘ playbook is not exist‘,‘flag‘:False}
            logging.info(results)
            #results=self.playbook_path+‘playbook is not existed‘
            #return code,complex_msg,results
        pbex= PlaybookExecutor(playbooks=[self.playbook_path],
                       inventory=self.inventory,
                       variable_manager=self.variable_manager,
                       loader=self.loader,
                       options=self.options,
                       passwords=self.passwords)
        self.results_callback=MyCallback()
        pbex._tqm._stdout_callback=self.results_callback
        try:
            code=pbex.run()
        except AnsibleParserError:
            code=1001
            results={‘playbook‘:self.playbook_path,‘msg‘:self.playbook_path+‘ playbook have syntax error‘,‘flag‘:False}
            #results=‘syntax error in ‘+self.playbook_path #語法錯誤
            return code,results
        if self.results_callback.status_no_hosts:
            code=1002
            results={‘playbook‘:self.playbook_path,‘msg‘:self.results_callback.status_no_hosts,‘flag‘:False,‘executed‘:False}
            return code,results
    def get_result(self):
        self.result_all={‘success‘:{},‘fail‘:{},‘unreachable‘:{}}
        for host, result in self.results_callback.host_ok.items():
            self.result_all[‘success‘][host] = result._result

        for host, result in self.results_callback.host_failed.items():
            if result._result.has_key("msg"):
                self.result_all[‘fail‘][host] = result._result[‘msg‘]

        for host, result in self.results_callback.host_unreachable.items():
            self.result_all[‘unreachable‘][host]= result._result[‘msg‘]
        
        for i in self.result_all[‘success‘].keys():
             print i,self.result_all[‘success‘][i]
        print self.result_all[‘fail‘]
        print self.result_all[‘unreachable‘]


if __name__ ==‘__main__‘:
        play_book=My_ansible_play(‘/etc/ansible/playbooks/test_ping.yml‘)
        logging.info(play_book)
        play_book.run()
        play_book.get_result()

附playbook腳本：test_ping.yml

---
- hosts: test
  gather_facts: False
  tasks:

    - name: test ping
      ping:
    - name: shell commond
      shell: echo "hello world"
      register: result
    - name: show debug info
      debug: var=result.stdout verbosity=0

python3.5調用ansible參考文檔：

　　　　　　　　　　　　　　https://www.cnblogs.com/stones/p/8252731.html

python調用ansible接口API執行命令