Ansible Runner lib

README.md

>>> import ansible
>>> ansible.run_module("ping", "*").host_results
{'h2.pdp7.net': {'ping': 'pong', 'invocation': {'module_args': {'data': 'pong'}}, 'ansible_facts': {'discovered_interpreter_python': '/usr/bin/python3'}, '_ansible_no_log': None, 'changed': False}, 'h1.pdp7.net': {'ping': 'pong', 'invocation': {'module_args': {'data': 'pong'}}, 'ansible_facts': {'discovered_interpreter_python': '/usr/bin/python3'}, '_ansible_no_log': None, 'changed': False}}
>>> ansible.command("*", "uptime")
{'h1.pdp7.net': CommandResult(host='h1.pdp7.net', stdout=' 14:44:29 up 10 days,  2:59,  1 user,  load average: 0.75, 0.90, 0.72', stderr='', rc=0), 'h2.pdp7.net': CommandResult(host='h2.pdp7.net', stdout=' 14:44:29 up 209 days, 23:39,  1 user,  load average: 0.09, 0.11, 0.12', stderr='', rc=0)}
>>> ansible.command("*", "false")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/acorcole/tmp/20231018-ansiblepy/ansible.py", line 29, in command
    raise CommandError(result)
ansible.CommandError: ('error running command', {'h1.pdp7.net': CommandResult(host='h1.pdp7.net', stdout='', stderr='', rc=1), 'h2.pdp7.net': CommandResult(host='h2.pdp7.net', stdout='', stderr='', rc=1)})

ansible.py

import ansible_runner
import dataclasses
import json


def run_module(module, pattern, **module_args):
    events = []
    args = ["-m", module, pattern]
    if module_args:
        args += ["-a", json.dumps(module_args)]
    ansible_runner.run_command(
        "ansible",
        args,
        event_handler=events.append,
        envvars={"ANSIBLE_CALLBACKS_ENABLED": "json", "ANSIBLE_LOAD_CALLBACK_PLUGINS": "1"},
        quiet=True)
    return AnsibleResult(events)


def command(pattern, command, check=True):
    r = run_module("command", pattern, cmd=command)
    result = {}
    for host, ansible_result in r.host_results.items():
        result[host] = CommandResult(host=host, stdout=ansible_result["stdout"], stderr=ansible_result["stderr"], rc=ansible_result["rc"])
    if check:
        failed = any([r.rc != 0 for r in result.values()])
        if failed:
            raise CommandError(result)
    return result

@dataclasses.dataclass
class CommandResult:
    host: str
    stdout: str
    stderr: str
    rc: int


class CommandError(Exception):
    def __init__(self, result: CommandResult):
        super().__init__("error running command", result)


class AnsibleResult:
    def __init__(self, events):
        self.events = events

    @property
    def host_results(self):
        runner_on_events = [e for e in self.events if e["event"] in ("runner_on_ok", "runner_on_failed")]
        return dict([(e["event_data"]["host"], e["event_data"]["res"]) for e in runner_on_events])