Python API
注意
此 API 旨在供 Ansible 内部使用。Ansible 可能会随时更改此 API，这可能会破坏与旧版 API 的向后兼容性。因此，Ansible 不支持外部使用。如果您想仅使用 Python API 执行剧本或模块，请首先考虑使用 ansible-runner。
从 API 角度来看，有几种方法可以使用 Ansible。您可以使用 Ansible Python API 控制节点，可以扩展 Ansible 以响应各种 Python 事件，可以编写插件，也可以从外部数据源插入清单数据。本文档提供了 Ansible 执行和剧本 API 的基本概述和示例。
如果您想从除 Python 之外的其他语言以编程方式使用 Ansible，异步触发事件或具有访问控制和日志记录需求，请参阅 AWX 项目。
注意
由于 Ansible 依赖于分叉进程，因此此 API 不是线程安全的。
Python API 示例
此示例是一个简单的演示，展示了如何以最小的方式运行几个任务。
#!/usr/bin/env python
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import json
import shutil
import ansible.constants as C
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.module_utils.common.collections import ImmutableDict
from ansible.inventory.manager import InventoryManager
from ansible.parsing.dataloader import DataLoader
from ansible.playbook.play import Play
from ansible.plugins.callback import CallbackBase
from ansible.plugins.loader import init_plugin_loader
from ansible.vars.manager import VariableManager
from ansible import context
# Create a callback plugin so we can capture the output
class ResultsCollectorJSONCallback(CallbackBase):
"""A sample callback plugin used for performing an action as results come in.
If you want to collect all results into a single object for processing at
the end of the execution, look into utilizing the ``json`` callback plugin
or writing your own custom callback plugin.
"""
def __init__(self, *args, **kwargs):
super(ResultsCollectorJSONCallback, self).__init__(*args, **kwargs)
self.host_ok = {}
self.host_unreachable = {}
self.host_failed = {}
def v2_runner_on_unreachable(self, result):
host = result._host
self.host_unreachable[host.get_name()] = result
def v2_runner_on_ok(self, result, *args, **kwargs):
"""Print a json representation of the result.
Also, store the result in an instance attribute for retrieval later
"""
host = result._host
self.host_ok[host.get_name()] = result
print(json.dumps({host.name: result._result}, indent=4))
def v2_runner_on_failed(self, result, *args, **kwargs):
host = result._host
self.host_failed[host.get_name()] = result
def main():
init_plugin_loader()
host_list = ['localhost', 'www.example.com', 'www.google.com']
# since the API is constructed for CLI it expects certain options to always be set in the context object
context.CLIARGS = ImmutableDict(module_path=['/to/mymodules', '/usr/share/ansible'], forks=10, become=None,
become_method=None, become_user=None, check=False, diff=False, verbosity=0)
# required for
# https://github.com/ansible/ansible/blob/devel/lib/ansible/inventory/manager.py#L204
sources = ','.join(host_list)
if len(host_list) == 1:
sources += ','
# initialize needed objects
loader = DataLoader() # Takes care of finding and reading yaml, json and ini files
passwords = dict(vault_pass='secret')
# Instantiate our ResultsCollectorJSONCallback for handling results as they come in. Ansible expects this to be one of its main display outlets
results_callback = ResultsCollectorJSONCallback()
# create inventory, use path to host config file as source or hosts in a comma separated string
inventory = InventoryManager(loader=loader, sources=sources)
# variable manager takes care of merging all the different sources to give you a unified view of variables available in each context
variable_manager = VariableManager(loader=loader, inventory=inventory)
# instantiate task queue manager, which takes care of forking and setting up all objects to iterate over host list and tasks
# IMPORTANT: This also adds library dirs paths to the module loader
# IMPORTANT: and so it must be initialized before calling `Play.load()`.
tqm = TaskQueueManager(
inventory=inventory,
variable_manager=variable_manager,
loader=loader,
passwords=passwords,
stdout_callback=results_callback, # Use our custom callback instead of the ``default`` callback plugin, which prints to stdout
)
# create data structure that represents our play, including tasks, this is basically what our YAML loader does internally.
play_source = dict(
name="Ansible Play",
hosts=host_list,
gather_facts='no',
tasks=[
dict(action=dict(module='shell', args='ls'), register='shell_out'),
dict(action=dict(module='debug', args=dict(msg='{{shell_out.stdout}}'))),
dict(action=dict(module='command', args=dict(cmd='/usr/bin/uptime'))),
]
)
# Create play object, playbook objects use .load instead of init or new methods,
# this will also automatically create the task objects from the info provided in play_source
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
# Actually run it
try:
result = tqm.run(play) # most interesting data for a play is actually sent to the callback's methods
finally:
# we always need to cleanup child procs and the structures we use to communicate with them
tqm.cleanup()
if loader:
loader.cleanup_all_tmp_files()
# Remove ansible tmpdir
shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)
print("UP ***********")
for host, result in results_callback.host_ok.items():
print('{0} >>> {1}'.format(host, result._result['stdout']))
print("FAILED *******")
for host, result in results_callback.host_failed.items():
print('{0} >>> {1}'.format(host, result._result['msg']))
print("DOWN *********")
for host, result in results_callback.host_unreachable.items():
print('{0} >>> {1}'.format(host, result._result['msg']))
if __name__ == '__main__':
main()
注意
Ansible 通过显示对象发出警告和错误，该对象直接打印到 stdout、stderr 和 Ansible 日志。
ansible 命令行工具（lib/ansible/cli/）的源代码 在 GitHub 上可用。