如何通过Ansible API以编程方式运行Ansible Playbook？

嘿，我一眼就瞅出问题所在了——你在定义Options命名元组的时候，把listhosts参数设成了True！这个参数的作用是让Ansible仅列出目标主机，完全跳过任务执行流程，这就是为啥你只得到了包含主机和playbook路径的结果，看不到任何任务执行的记录。

关键修改点 #

把listhosts=True改成listhosts=False，这样Ansible才会真正去执行playbook里的任务。修改后的Options定义如下：

options = Options(connection='local', module_path='%s/' % (ROOT_DIR), forks=100, become=None, become_method=None, become_user=None, check=False, diff=False, listhosts=False, listtasks=False, listtags=False, syntax=False)

额外优化建议 #

为了避免潜在的路径问题，建议把playbook和inventory的路径改成绝对路径（用os.path.join拼接更可靠），同时检查你的hosts文件配置是否正确，确保localhost能正常访问。如果playbook需要sudo权限，记得设置become=True、become_method='sudo'和become_user='root'这类参数。

修改后的完整代码 #

import os
import sys
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
ROOT_DIR = os.path.dirname(SCRIPT_DIR)
import json
from collections import namedtuple
from ansible.parsing.dataloader import DataLoader
from ansible.vars.manager import VariableManager
from ansible.inventory.manager import InventoryManager
from ansible.playbook.play import Play
from ansible.executor.playbook_executor import PlaybookExecutor

def ansible_part():
    # 改用绝对路径，避免相对路径带来的问题
    playbook_path = os.path.join(ROOT_DIR, "playbooks/asg_elb_example.yml")
    inventory_path = os.path.join(ROOT_DIR, "hosts")
    
    Options = namedtuple('Options', ['connection', 'module_path', 'forks', 'become', 'become_method', 'become_user', 'check', 'diff', 'listhosts', 'listtasks', 'listtags', 'syntax'])
    loader = DataLoader()
    # 核心修改：关闭listhosts模式，开启任务执行
    options = Options(connection='local', module_path='%s/' % (ROOT_DIR), forks=100, become=None, become_method=None, become_user=None, check=False, diff=False, listhosts=False, listtasks=False, listtags=False, syntax=False)
    passwords = dict(vault_pass='secret')
    
    inventory = InventoryManager(loader=loader, sources=[inventory_path])
    variable_manager = VariableManager(loader=loader, inventory=inventory)
    
    executor = PlaybookExecutor(
        playbooks=[playbook_path],
        inventory=inventory,
        variable_manager=variable_manager,
        loader=loader,
        options=options,
        passwords=passwords)
    
    results = executor.run()
    print(results)

def main():
    ansible_part()

sys.exit(main())

内容的提问来源于stack exchange，提问作者user982599