tessl/pypi-ansible-runner

Consistent Ansible Python API and CLI with container and process isolation runtime capabilities

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Configuration and Runner Classes

Name: tessl/pypi-ansible-runner
Author: tessl

Core classes for advanced use cases requiring fine-grained control over execution parameters, event handling, and process management. These classes provide the foundation for all ansible-runner operations and enable custom integration patterns.

Capabilities

Runner Class

The main execution class that handles Ansible operations, event processing, and result management.

class Runner:
    def __init__(
        self,
        config,
        cancel_callback: callable = None,
        remove_partials: bool = True,
        event_handler: callable = None,
        artifacts_handler: callable = None,
        finished_callback: callable = None,
        status_handler: callable = None
    )

Properties:

config: Configuration object (RunnerConfig, CommandConfig, etc.)
status (str): Current execution status ('unstarted', 'running', 'successful', 'failed', 'error', 'canceled')
rc (int): Return code from execution
canceled (bool): Whether execution was canceled
timed_out (bool): Whether execution timed out
errored (bool): Whether execution encountered an error
events: Iterator over execution events
stdout: Iterator over stdout output

Methods:

def run(self) -> str:
    """Execute the configured operation and return final status"""

def cancel(self) -> None:
    """Cancel the running operation"""

def event_callback(self, event_data: dict) -> None:
    """Handle individual event data (internal use)"""

Usage example:

import ansible_runner
from ansible_runner.config.runner import RunnerConfig

# Create configuration
config = RunnerConfig(
    private_data_dir='/tmp/demo',
    playbook='site.yml',
    inventory='hosts'
)

# Create runner instance
runner = Runner(config)

# Execute
status = runner.run()

print(f"Execution completed with status: {status}")
print(f"Return code: {runner.rc}")

# Process events
for event in runner.events:
    if event['event'] == 'playbook_on_task_start':
        print(f"Starting task: {event['event_data']['task']}")

RunnerConfig Class

Configuration class for playbook execution with comprehensive parameter support.

class RunnerConfig:
    def __init__(
        self,
        private_data_dir: str = None,
        playbook: str = None,
        inventory: str = None,
        roles_path: str = None,
        verbosity: int = None,
        quiet: bool = False,
        artifact_dir: str = None,
        project_dir: str = None,
        ident: str = None,
        rotate_artifacts: int = 0,
        json_mode: bool = False,
        check: bool = False,
        tags: str = None,
        skip_tags: str = None,
        limit: str = None,
        module_path: str = None,
        forks: int = None,
        vault_ids: list = None,
        vault_password_file: str = None,
        extravars: dict = None,
        suppress_ansible_output: bool = False,
        ssh_key: str = None,
        process_isolation: bool = False,
        process_isolation_executable: str = None,
        process_isolation_path: str = None,
        process_isolation_hide_paths: list = None,
        process_isolation_show_paths: list = None,
        process_isolation_ro_paths: list = None,
        container_image: str = None,
        container_volume_mounts: list = None,
        container_options: list = None,
        directory_isolation_base_path: str = None,
        envvars: dict = None,
        passwords: dict = None,
        timeout: int = None,
        job_timeout: int = None,
        idle_timeout: int = None,
        pexpect_timeout: int = None,
        pexpect_use_poll: bool = True,
        runner_mode: str = 'pexpect',
        host_cwd: str = None,
        fact_cache: str = None,
        fact_cache_type: str = None,
        cmdline: str = None,
        start_at_task: str = None,
        step: bool = False
    )

Usage example:

from ansible_runner.config.runner import RunnerConfig
from ansible_runner import Runner

# Create detailed configuration
config = RunnerConfig(
    private_data_dir='/path/to/project',
    playbook='deploy.yml',
    inventory='production/hosts',
    extravars={
        'app_version': '2.1.0',
        'environment': 'production'
    },
    verbosity=2,
    limit='webservers',
    tags='deploy,configure',
    process_isolation=True,
    container_image='quay.io/ansible/ansible-runner:latest',
    timeout=3600
)

# Create and run
runner = Runner(config)
status = runner.run()

CommandConfig Class

Configuration class for executing Ansible command-line tools.

class CommandConfig:
    def __init__(
        self,
        executable_cmd: str,
        cmdline_args: list = None,
        private_data_dir: str = None,
        input_fd: object = None,
        output_fd: object = None,
        error_fd: object = None,
        artifact_dir: str = None,
        ident: str = None,
        rotate_artifacts: int = 0,
        quiet: bool = False,
        timeout: int = None,
        runner_mode: str = 'pexpect',
        process_isolation: bool = False,
        process_isolation_executable: str = None,
        process_isolation_path: str = None,
        process_isolation_hide_paths: list = None,
        process_isolation_show_paths: list = None,
        process_isolation_ro_paths: list = None,
        container_image: str = None,
        container_volume_mounts: list = None,
        container_options: list = None,
        directory_isolation_base_path: str = None,
        envvars: dict = None
    )

Usage example:

from ansible_runner.config.command import CommandConfig
from ansible_runner import Runner

# Configure command execution
config = CommandConfig(
    executable_cmd='ansible-playbook',
    cmdline_args=[
        'site.yml',
        '-i', 'inventory',
        '--check',
        '--diff'
    ],
    private_data_dir='/project',
    timeout=1800
)

# Execute command
runner = Runner(config)
status = runner.run()

print(f"Command completed: {status}")
print(f"Return code: {runner.rc}")

InventoryConfig Class

Configuration class for inventory operations.

class InventoryConfig:
    def __init__(
        self,
        action: str,
        inventories: list,
        response_format: str = None,
        host: str = None,
        playbook_dir: str = None,
        private_data_dir: str = None,
        artifact_dir: str = None,
        ident: str = None,
        rotate_artifacts: int = 0,
        timeout: int = None,
        process_isolation: bool = False,
        process_isolation_executable: str = None,
        container_image: str = None,
        envvars: dict = None
    )

Usage example:

from ansible_runner.config.inventory import InventoryConfig
from ansible_runner import Runner

# Configure inventory listing
config = InventoryConfig(
    action='list',
    inventories=['production/hosts', 'production/groups'],
    response_format='json'
)

runner = Runner(config)
status = runner.run()

if status == 'successful':
    # Process inventory data from artifacts
    print("Inventory retrieved successfully")

DocConfig Class

Configuration class for plugin documentation operations.

class DocConfig:
    def __init__(
        self,
        plugin_names: list,
        plugin_type: str = None,
        response_format: str = None,
        snippet: bool = False,
        playbook_dir: str = None,
        module_path: str = None,
        private_data_dir: str = None,
        artifact_dir: str = None,
        ident: str = None,
        rotate_artifacts: int = 0,
        timeout: int = None,
        process_isolation: bool = False,
        process_isolation_executable: str = None,
        container_image: str = None,
        envvars: dict = None
    )

Usage example:

from ansible_runner.config.doc import DocConfig
from ansible_runner import Runner

# Configure plugin documentation retrieval
config = DocConfig(
    plugin_names=['copy', 'template', 'file'],
    plugin_type='module',
    response_format='json',
    snippet=False
)

runner = Runner(config)
status = runner.run()

AnsibleCfgConfig Class

Configuration class for Ansible configuration operations.

class AnsibleCfgConfig:
    def __init__(
        self,
        action: str,
        config_file: str = None,
        only_changed: bool = None,
        private_data_dir: str = None,
        artifact_dir: str = None,
        ident: str = None,
        rotate_artifacts: int = 0,
        timeout: int = None,
        process_isolation: bool = False,
        process_isolation_executable: str = None,
        container_image: str = None,
        envvars: dict = None
    )

Usage example:

from ansible_runner.config.ansible_cfg import AnsibleCfgConfig
from ansible_runner import Runner

# Configure ansible configuration dump
config = AnsibleCfgConfig(
    action='dump',
    only_changed=True
)

runner = Runner(config)
status = runner.run()

Advanced Usage Patterns

Custom Event Handling

class CustomRunner:
    def __init__(self, config):
        self.events = []
        self.runner = Runner(
            config,
            event_handler=self.handle_event,
            status_handler=self.handle_status,
            finished_callback=self.handle_finished
        )
    
    def handle_event(self, event):
        # Custom event processing
        self.events.append(event)
        if event['event'] == 'runner_on_failed':
            print(f"Task failed: {event['event_data']}")
        return True
    
    def handle_status(self, data, config):
        print(f"Status changed: {data['status']}")
    
    def handle_finished(self, runner):
        print(f"Execution finished: {runner.status}")
        self.analyze_results()
    
    def analyze_results(self):
        failed_tasks = [e for e in self.events if e['event'] == 'runner_on_failed']
        print(f"Found {len(failed_tasks)} failed tasks")
    
    def run(self):
        return self.runner.run()

# Usage
config = RunnerConfig(
    private_data_dir='/project',
    playbook='site.yml'
)
custom_runner = CustomRunner(config)
status = custom_runner.run()

Configuration Chaining

def create_production_config(playbook, **overrides):
    """Create standardized production configuration"""
    base_config = {
        'private_data_dir': '/production/ansible',
        'inventory': 'production/hosts',
        'verbosity': 1,
        'process_isolation': True,
        'container_image': 'quay.io/ansible/ansible-runner:latest',
        'timeout': 3600,
        'envvars': {
            'ANSIBLE_HOST_KEY_CHECKING': 'False',
            'ANSIBLE_GATHERING': 'smart'
        }
    }
    
    # Apply overrides
    base_config.update(overrides)
    base_config['playbook'] = playbook
    
    return RunnerConfig(**base_config)

# Usage
deploy_config = create_production_config(
    'deploy.yml',
    limit='webservers',
    extravars={'version': '2.1.0'}
)

maintenance_config = create_production_config(
    'maintenance.yml',
    tags='maintenance',
    check=True
)

Result Processing

def process_runner_results(runner):
    """Comprehensive result processing"""
    results = {
        'status': runner.status,
        'return_code': runner.rc,
        'events': [],
        'failed_hosts': set(),
        'changed_hosts': set(),
        'unreachable_hosts': set()
    }
    
    # Process events
    for event in runner.events:
        event_type = event['event']
        results['events'].append(event)
        
        if event_type == 'runner_on_failed':
            results['failed_hosts'].add(event['event_data']['host'])
        elif event_type == 'runner_on_ok' and event['event_data'].get('changed'):
            results['changed_hosts'].add(event['event_data']['host'])
        elif event_type == 'runner_on_unreachable':
            results['unreachable_hosts'].add(event['event_data']['host'])
    
    return results

# Usage
config = RunnerConfig(private_data_dir='/project', playbook='site.yml')
runner = Runner(config)
status = runner.run()
results = process_runner_results(runner)

print(f"Execution summary:")
print(f"  Status: {results['status']}")
print(f"  Failed hosts: {len(results['failed_hosts'])}")
print(f"  Changed hosts: {len(results['changed_hosts'])}")
print(f"  Unreachable hosts: {len(results['unreachable_hosts'])}")

Install with Tessl CLI