ansible-runner

import ansible_runner import os import tempfile import shutil # Create a temporary directory for ansible-runner's private_data_dir with tempfile.TemporaryDirectory() as private_data_dir: # Create project/ and inventory/ subdirectories project_dir = os.path.join(private_data_dir, 'project') inventory_dir = os.path.join(private_data_dir, 'inventory') os.makedirs(project_dir, exist_ok=True) os.makedirs(inventory_dir, exist_ok=True) # Write a simple Ansible playbook playbook_content = ''' - name: Test playbook hosts: localhost connection: local tasks: - name: Print a message ansible.builtin.debug: msg: "Hello from Ansible Runner!" - name: Verify an environment variable (optional) ansible.builtin.debug: msg: "TEST_VAR is {{ lookup('env', 'TEST_VAR') }}" when: lookup('env', 'TEST_VAR') | length > 0 ''' with open(os.path.join(project_dir, 'test_playbook.yml'), 'w') as f: f.write(playbook_content) # Write a simple inventory file inventory_content = ''' localhost ansible_connection=local ''' with open(os.path.join(inventory_dir, 'hosts'), 'w') as f: f.write(inventory_content) # Set an optional environment variable for the playbook to consume os.environ['TEST_VAR'] = os.environ.get('MY_ANSIBLE_RUNNER_TEST_VAR', 'DefaultValueFromRunner') print(f"Running Ansible playbook in: {private_data_dir}") # Run the playbook using ansible-runner r = ansible_runner.run( private_data_dir=private_data_dir, playbook='test_playbook.yml', quiet=False # Set to True to suppress stdout from Ansible process ) print(f"\nAnsible Runner Status: {r.status}") print(f"Ansible Runner Return Code: {r.rc}") if r.stdout: print("\n--- Ansible Playbook Output ---") # You can access stdout as a file-like object or iterate through events # For simplicity, let's print the entire stdout content stdout_path = os.path.join(r.artifact_dir, 'stdout') if os.path.exists(stdout_path): with open(stdout_path, 'r') as f: print(f.read()) else: print("Stdout file not found.") if r.stderr: print("\n--- Ansible Playbook Error Output ---") stderr_path = os.path.join(r.artifact_dir, 'stderr') if os.path.exists(stderr_path): with open(stderr_path, 'r') as f: print(f.read()) else: print("Stderr file not found.") # Clean up the environment variable del os.environ['TEST_VAR']