ansible-core

import json import shutil import os import ansible.constants as C from ansible.executor.task_queue_manager import TaskQueueManager 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.vars.manager import VariableManager from ansible import context # NOTE: The Ansible Python API is for internal use and not officially supported for external applications. # For robust programmatic execution of playbooks and modules, consider using 'ansible-runner'. # --- Minimal example to run a simple ad-hoc command --- # Configure Ansible context (optional, but good practice) context.CLIARGS = ['ansible', 'all', '-m', 'ping', '-i', 'localhost,'] # Mimics command line args class ResultsCollectorJSONCallback(CallbackBase): def __init__(self): self.host_ok = {} self.host_failed = {} self.host_unreachable = {} def v2_runner_on_ok(self, result, **kwargs): self.host_ok[result._host.get_name()] = result._result def v2_runner_on_failed(self, result, **kwargs): self.host_failed[result._host.get_name()] = result._result def v2_runner_on_unreachable(self, result, **kwargs): self.host_unreachable[result._host.get_name()] = result._result # initialize needed objects loader = DataLoader() inv_data = 'localhost ansible_connection=local' inventory = InventoryManager(loader=loader, sources=[inv_data]) variable_manager = VariableManager(loader=loader, inventory=inventory) # instantiate our callback plugin results_callback = ResultsCollectorJSONCallback() # create play with tasks play_source = dict( name="Ansible Ad-hoc Ping", hosts='all', gather_facts='no', tasks=[ dict(action=dict(module='ping'), register='ping_result') ] ) play = Play().load(play_source, variable_manager=variable_manager, loader=loader) # run it t = TaskQueueManager( inventory=inventory, variable_manager=variable_manager, loader=loader, options=context.CLIARGS, passwords={}, stdout_callback=results_callback, ) try: status = t.run(play) finally: # cleanup after ourselves if t is not None: t.cleanup() if loader: # DataLoader may delete tmpdir on exit, ensure it's removed if exists if hasattr(loader, '_tempdir') and os.path.exists(loader._tempdir): shutil.rmtree(loader._tempdir) print("\n--- Ad-hoc Ping Results ---") print(f"OK: {json.dumps(results_callback.host_ok, indent=4)}") print(f"Failed: {json.dumps(results_callback.host_failed, indent=4)}") print(f"Unreachable: {json.dumps(results_callback.host_unreachable, indent=4)}")