API

ansible_compat.config

Store configuration options as a singleton.

AnsibleConfig

Bases: UserDict[str, object]

Interface to query Ansible configuration.

This should allow user to access everything provided by ansible-config dump without having to parse the data himself.

Attributes:

• _aliases (dict[str, str]) –
• action_warnings (bool) –
• agnostic_become_prompt (bool) –
• allow_world_readable_tmpfiles (bool) –
• ansible_connection_path (str | None) –
• ansible_cow_acceptlist (list[str]) –
• ansible_cow_path (str | None) –
• ansible_cow_selection (str) –
• ansible_force_color (bool) –
• ansible_nocolor (bool) –
• ansible_nocows (bool) –
• ansible_pipelining (bool) –
• any_errors_fatal (bool) –
• become_allow_same_user (bool) –
• become_plugin_path (list[str]) –
• cache_plugin (str) –
• cache_plugin_connection (str | None) –
• cache_plugin_prefix (str) –
• cache_plugin_timeout (int) –
• callable_accept_list (list[str]) –
• callbacks_enabled (list[str]) –
• collections_on_ansible_version_mismatch (Literal['warning', 'ignore']) –
• collections_paths (list[str]) –
• collections_scan_sys_path (bool) –
• color_changed (str) –
• color_console_prompt (str) –
• color_debug (str) –
• color_deprecate (str) –
• color_diff_add (str) –
• color_diff_lines (str) –
• color_diff_remove (str) –
• color_error (str) –
• color_highlight (str) –
• color_ok (str) –
• color_skip (str) –
• color_unreachable (str) –
• color_verbose (str) –
• color_warn (str) –
• command_warnings (bool) –
• conditional_bare_vars (bool) –
• connection_facts_modules (dict[str, str]) –
• controller_python_warning (bool) –
• coverage_remote_output (str | None) –
• coverage_remote_paths (list[str]) –
• default_action_plugin_path (list[str]) –
• default_allow_unsafe_lookups (bool) –
• default_ask_pass (bool) –
• default_ask_vault_pass (bool) –
• default_become (bool) –
• default_become_ask_pass (bool) –
• default_become_exe (str | None) –
• default_become_flags (str) –
• default_become_method (str) –
• default_become_user (str) –
• default_cache_plugin_path (list[str]) –
• default_callback_plugin_path (list[str]) –
• default_cliconf_plugin_path (list[str]) –
• default_collections_path (list[str]) –
• default_connection_plugin_path (list[str]) –
• default_debug (bool) –
• default_executable (str) –
• default_fact_path (str | None) –
• default_filter_plugin_path (list[str]) –
• default_force_handlers (bool) –
• default_forks (int) –
• default_gather_subset (list[str]) –
• default_gather_timeout (int) –
• default_gathering (Literal['smart', 'explicit', 'implicit']) –
• default_handler_includes_static (bool) –
• default_hash_behaviour (str) –
• default_host_list (list[str]) –
• default_httpapi_plugin_path (list[str]) –
• default_internal_poll_interval (float) –
• default_inventory_plugin_path (list[str]) –
• default_jinja2_extensions (list[str]) –
• default_jinja2_native (bool) –
• default_keep_remote_files (bool) –
• default_libvirt_lxc_noseclabel (bool) –
• default_load_callback_plugins (bool) –
• default_local_tmp (str) –
• default_log_filter (list[str]) –
• default_log_path (str | None) –
• default_lookup_plugin_path (list[str]) –
• default_managed_str (str) –
• default_module_args (str) –
• default_module_compression (str) –
• default_module_name (str) –
• default_module_path (list[str]) –
• default_module_utils_path (list[str]) –
• default_netconf_plugin_path (list[str]) –
• default_no_log (bool) –
• default_no_target_syslog (bool) –
• default_null_representation (str | None) –
• default_poll_interval (int) –
• default_private_key_file (str | None) –
• default_private_role_vars (bool) –
• default_remote_port (str | None) –
• default_remote_user (str | None) –
• default_roles_path (list[str]) –
• default_selinux_special_fs (list[str]) –
• default_stdout_callback (str) –
• default_strategy (str) –
• default_strategy_plugin_path (list[str]) –
• default_su (bool) –
• default_syslog_facility (str) –
• default_task_includes_static (bool) –
• default_terminal_plugin_path (list[str]) –
• default_test_plugin_path (list[str]) –
• default_timeout (int) –
• default_transport (str) –
• default_undefined_var_behavior (bool) –
• default_vars_plugin_path (list[str]) –
• default_vault_encrypt_identity (str | None) –
• default_vault_id_match (bool) –
• default_vault_identity (str) –
• default_vault_identity_list (list[str]) –
• default_vault_password_file (str | None) –
• default_verbosity (int) –
• deprecation_warnings (bool) –
• devel_warning (bool) –
• diff_always (bool) –
• diff_context (int) –
• display_args_to_stdout (bool) –
• display_skipped_hosts (bool) –
• doc_fragment_plugin_path (list[str]) –
• docsite_root_url (str) –
• duplicate_yaml_dict_key (Literal['warn', 'error', 'ignore']) –
• enable_task_debugger (bool) –
• error_on_missing_handler (bool) –
• facts_modules (list[str]) –
• galaxy_cache_dir (str) –
• galaxy_display_progress (str | None) –
• galaxy_ignore_certs (bool) –
• galaxy_role_skeleton (str | None) –
• galaxy_role_skeleton_ignore (list[str]) –
• galaxy_server (str) –
• galaxy_server_list (str | None) –
• galaxy_token_path (str) –
• host_key_checking (bool) –
• host_pattern_mismatch (Literal['warning', 'error', 'ignore']) –
• inject_facts_as_vars (bool) –
• interpreter_python (str) –
• interpreter_python_distro_map (dict[str, str]) –
• interpreter_python_fallback (list[str]) –
• invalid_task_attribute_failed (bool) –
• inventory_any_unparsed_is_failed (bool) –
• inventory_cache_enabled (bool) –
• inventory_cache_plugin (str | None) –
• inventory_cache_plugin_connection (str | None) –
• inventory_cache_plugin_prefix (str) –
• inventory_cache_timeout (int) –
• inventory_enabled (list[str]) –
• inventory_export (bool) –
• inventory_ignore_exts (str) –
• inventory_ignore_patterns (list[str]) –
• inventory_unparsed_is_failed (bool) –
• localhost_warning (bool) –
• max_file_size_for_diff (int) –
• module_ignore_exts (str) –
• netconf_ssh_config (str | None) –
• network_group_modules (list[str]) –
• old_plugin_cache_clearing (bool) –
• paramiko_host_key_auto_add (bool) –
• paramiko_look_for_keys (bool) –
• persistent_command_timeout (int) –
• persistent_connect_retry_timeout (int) –
• persistent_connect_timeout (int) –
• persistent_control_path_dir (str) –
• playbook_dir (str | None) –
• playbook_vars_root (Literal['top', 'bottom', 'all']) –
• plugin_filters_cfg (str | None) –
• python_module_rlimit_nofile (int) –
• retry_files_enabled (bool) –
• retry_files_save_path (str | None) –
• run_vars_plugins (str) –
• show_custom_stats (bool) –
• string_conversion_action (Literal['warn', 'error', 'ignore']) –
• string_type_filters (list[str]) –
• system_warnings (bool) –
• tags_run (list[str]) –
• tags_skip (list[str]) –
• task_debugger_ignore_errors (bool) –
• task_timeout (int) –
• transform_invalid_group_chars (Literal['always', 'never', 'ignore', 'silently']) –
• use_persistent_connections (bool) –
• variable_plugins_enabled (list[str]) –
• variable_precedence (list[str]) –
• verbose_to_stderr (bool) –
• win_async_startup_timeout (int) –
• worker_shutdown_poll_count (int) –
• worker_shutdown_poll_delay (float) –
• yaml_filename_extensions (list[str]) –

Source code in src/ansible_compat/config.py

class AnsibleConfig(
    UserDict[str, object],
):  # pylint: disable=too-many-ancestors # noqa: DOC605
    """Interface to query Ansible configuration.

    This should allow user to access everything provided by `ansible-config dump` without having to parse the data himself.

    Attributes:
        _aliases:
        action_warnings:
        agnostic_become_prompt:
        allow_world_readable_tmpfiles:
        ansible_connection_path:
        ansible_cow_acceptlist:
        ansible_cow_path:
        ansible_cow_selection:
        ansible_force_color:
        ansible_nocolor:
        ansible_nocows:
        ansible_pipelining:
        any_errors_fatal:
        become_allow_same_user:
        become_plugin_path:
        cache_plugin:
        cache_plugin_connection:
        cache_plugin_prefix:
        cache_plugin_timeout:
        callable_accept_list:
        callbacks_enabled:
        collections_on_ansible_version_mismatch:
        collections_paths:
        collections_scan_sys_path:
        color_changed:
        color_console_prompt:
        color_debug:
        color_deprecate:
        color_diff_add:
        color_diff_lines:
        color_diff_remove:
        color_error:
        color_highlight:
        color_ok:
        color_skip:
        color_unreachable:
        color_verbose:
        color_warn:
        command_warnings:
        conditional_bare_vars:
        connection_facts_modules:
        controller_python_warning:
        coverage_remote_output:
        coverage_remote_paths:
        default_action_plugin_path:
        default_allow_unsafe_lookups:
        default_ask_pass:
        default_ask_vault_pass:
        default_become:
        default_become_ask_pass:
        default_become_exe:
        default_become_flags:
        default_become_method:
        default_become_user:
        default_cache_plugin_path:
        default_callback_plugin_path:
        default_cliconf_plugin_path:
        default_collections_path:
        default_connection_plugin_path:
        default_debug:
        default_executable:
        default_fact_path:
        default_filter_plugin_path:
        default_force_handlers:
        default_forks:
        default_gather_subset:
        default_gather_timeout:
        default_gathering:
        default_handler_includes_static:
        default_hash_behaviour:
        default_host_list:
        default_httpapi_plugin_path:
        default_internal_poll_interval:
        default_inventory_plugin_path:
        default_jinja2_extensions:
        default_jinja2_native:
        default_keep_remote_files:
        default_libvirt_lxc_noseclabel:
        default_load_callback_plugins:
        default_local_tmp:
        default_log_filter:
        default_log_path:
        default_lookup_plugin_path:
        default_managed_str:
        default_module_args:
        default_module_compression:
        default_module_name:
        default_module_path:
        default_module_utils_path:
        default_netconf_plugin_path:
        default_no_log:
        default_no_target_syslog:
        default_null_representation:
        default_poll_interval:
        default_private_key_file:
        default_private_role_vars:
        default_remote_port:
        default_remote_user:
        default_roles_path:
        default_selinux_special_fs:
        default_stdout_callback:
        default_strategy:
        default_strategy_plugin_path:
        default_su:
        default_syslog_facility:
        default_task_includes_static:
        default_terminal_plugin_path:
        default_test_plugin_path:
        default_timeout:
        default_transport:
        default_undefined_var_behavior:
        default_vars_plugin_path:
        default_vault_encrypt_identity:
        default_vault_id_match:
        default_vault_identity:
        default_vault_identity_list:
        default_vault_password_file:
        default_verbosity:
        deprecation_warnings:
        devel_warning:
        diff_always:
        diff_context:
        display_args_to_stdout:
        display_skipped_hosts:
        doc_fragment_plugin_path:
        docsite_root_url:
        duplicate_yaml_dict_key:
        enable_task_debugger:
        error_on_missing_handler:
        facts_modules:
        galaxy_cache_dir:
        galaxy_display_progress:
        galaxy_ignore_certs:
        galaxy_role_skeleton:
        galaxy_role_skeleton_ignore:
        galaxy_server:
        galaxy_server_list:
        galaxy_token_path:
        host_key_checking:
        host_pattern_mismatch:
        inject_facts_as_vars:
        interpreter_python:
        interpreter_python_distro_map:
        interpreter_python_fallback:
        invalid_task_attribute_failed:
        inventory_any_unparsed_is_failed:
        inventory_cache_enabled:
        inventory_cache_plugin:
        inventory_cache_plugin_connection:
        inventory_cache_plugin_prefix:
        inventory_cache_timeout:
        inventory_enabled:
        inventory_export:
        inventory_ignore_exts:
        inventory_ignore_patterns:
        inventory_unparsed_is_failed:
        localhost_warning:
        max_file_size_for_diff:
        module_ignore_exts:
        netconf_ssh_config:
        network_group_modules:
        old_plugin_cache_clearing:
        paramiko_host_key_auto_add:
        paramiko_look_for_keys:
        persistent_command_timeout:
        persistent_connect_retry_timeout:
        persistent_connect_timeout:
        persistent_control_path_dir:
        playbook_dir:
        playbook_vars_root:
        plugin_filters_cfg:
        python_module_rlimit_nofile:
        retry_files_enabled:
        retry_files_save_path:
        run_vars_plugins:
        show_custom_stats:
        string_conversion_action:
        string_type_filters:
        system_warnings:
        tags_run:
        tags_skip:
        task_debugger_ignore_errors:
        task_timeout:
        transform_invalid_group_chars:
        use_persistent_connections:
        variable_plugins_enabled:
        variable_precedence:
        verbose_to_stderr:
        win_async_startup_timeout:
        worker_shutdown_poll_count:
        worker_shutdown_poll_delay:
        yaml_filename_extensions:
    """

    _aliases: ClassVar[dict[str, str]] = {
        "COLLECTIONS_PATH": "COLLECTIONS_PATHS",  # 2.9 -> 2.10
    }
    # Expose some attributes to enable auto-complete in editors, based on
    # https://docs.ansible.com/ansible/latest/reference_appendices/config.html
    action_warnings: bool = True
    agnostic_become_prompt: bool = True
    allow_world_readable_tmpfiles: bool = False
    ansible_connection_path: str | None = None
    ansible_cow_acceptlist: list[str]
    ansible_cow_path: str | None = None
    ansible_cow_selection: str = "default"
    ansible_force_color: bool = False
    ansible_nocolor: bool = False
    ansible_nocows: bool = False
    ansible_pipelining: bool = False
    any_errors_fatal: bool = False
    become_allow_same_user: bool = False
    become_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/become",
        "/usr/share/ansible/plugins/become",
    ]
    cache_plugin: str = "memory"
    cache_plugin_connection: str | None = None
    cache_plugin_prefix: str = "ansible_facts"
    cache_plugin_timeout: int = 86400
    callable_accept_list: ClassVar[list[str]] = []
    callbacks_enabled: ClassVar[list[str]] = []
    collections_on_ansible_version_mismatch: Literal["warning", "ignore"] = "warning"
    collections_paths: list[str] = [
        "~/.ansible/collections",
        "/usr/share/ansible/collections",
    ]
    collections_scan_sys_path: bool = True
    color_changed: str = "yellow"
    color_console_prompt: str = "white"
    color_debug: str = "dark gray"
    color_deprecate: str = "purple"
    color_diff_add: str = "green"
    color_diff_lines: str = "cyan"
    color_diff_remove: str = "red"
    color_error: str = "red"
    color_highlight: str = "white"
    color_ok: str = "green"
    color_skip: str = "cyan"
    color_unreachable: str = "bright red"
    color_verbose: str = "blue"
    color_warn: str = "bright purple"
    command_warnings: bool = False
    conditional_bare_vars: bool = False
    connection_facts_modules: dict[str, str]
    controller_python_warning: bool = True
    coverage_remote_output: str | None
    coverage_remote_paths: list[str]
    default_action_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/action",
        "/usr/share/ansible/plugins/action",
    ]
    default_allow_unsafe_lookups: bool = False
    default_ask_pass: bool = False
    default_ask_vault_pass: bool = False
    default_become: bool = False
    default_become_ask_pass: bool = False
    default_become_exe: str | None = None
    default_become_flags: str
    default_become_method: str = "sudo"
    default_become_user: str = "root"
    default_cache_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/cache",
        "/usr/share/ansible/plugins/cache",
    ]
    default_callback_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/callback",
        "/usr/share/ansible/plugins/callback",
    ]
    default_cliconf_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/cliconf",
        "/usr/share/ansible/plugins/cliconf",
    ]
    default_connection_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/connection",
        "/usr/share/ansible/plugins/connection",
    ]
    default_debug: bool = False
    default_executable: str = "/bin/sh"
    default_fact_path: str | None = None
    default_filter_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/filter",
        "/usr/share/ansible/plugins/filter",
    ]
    default_force_handlers: bool = False
    default_forks: int = 5
    default_gather_subset: ClassVar[list[str]] = ["all"]
    default_gather_timeout: int = 10
    default_gathering: Literal["smart", "explicit", "implicit"] = "smart"
    default_handler_includes_static: bool = False
    default_hash_behaviour: str = "replace"
    default_host_list: ClassVar[list[str]] = ["/etc/ansible/hosts"]
    default_httpapi_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/httpapi",
        "/usr/share/ansible/plugins/httpapi",
    ]
    default_internal_poll_interval: float = 0.001
    default_inventory_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/inventory",
        "/usr/share/ansible/plugins/inventory",
    ]
    default_jinja2_extensions: ClassVar[list[str]] = []
    default_jinja2_native: bool = False
    default_keep_remote_files: bool = False
    default_libvirt_lxc_noseclabel: bool = False
    default_load_callback_plugins: bool = False
    default_local_tmp: str = "~/.ansible/tmp"
    default_log_filter: ClassVar[list[str]] = []
    default_log_path: str | None = None
    default_lookup_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/lookup",
        "/usr/share/ansible/plugins/lookup",
    ]
    default_managed_str: str = "Ansible managed"
    default_module_args: str
    default_module_compression: str = "ZIP_DEFLATED"
    default_module_name: str = "command"
    default_module_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/modules",
        "/usr/share/ansible/plugins/modules",
    ]
    default_module_utils_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/module_utils",
        "/usr/share/ansible/plugins/module_utils",
    ]
    default_netconf_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/netconf",
        "/usr/share/ansible/plugins/netconf",
    ]
    default_no_log: bool = False
    default_no_target_syslog: bool = False
    default_null_representation: str | None = None
    default_poll_interval: int = 15
    default_private_key_file: str | None = None
    default_private_role_vars: bool = False
    default_remote_port: str | None = None
    default_remote_user: str | None = None
    # https://docs.ansible.com/ansible/latest/reference_appendices/config.html#collections-paths
    default_collections_path: ClassVar[list[str]] = [
        "~/.ansible/collections",
        "/usr/share/ansible/collections",
    ]
    default_roles_path: ClassVar[list[str]] = [
        "~/.ansible/roles",
        "/usr/share/ansible/roles",
        "/etc/ansible/roles",
    ]
    default_selinux_special_fs: ClassVar[list[str]] = [
        "fuse",
        "nfs",
        "vboxsf",
        "ramfs",
        "9p",
        "vfat",
    ]
    default_stdout_callback: str = "default"
    default_strategy: str = "linear"
    default_strategy_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/strategy",
        "/usr/share/ansible/plugins/strategy",
    ]
    default_su: bool = False
    default_syslog_facility: str = "LOG_USER"
    default_task_includes_static: bool = False
    default_terminal_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/terminal",
        "/usr/share/ansible/plugins/terminal",
    ]
    default_test_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/test",
        "/usr/share/ansible/plugins/test",
    ]
    default_timeout: int = 10
    default_transport: str = "smart"
    default_undefined_var_behavior: bool = True
    default_vars_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/vars",
        "/usr/share/ansible/plugins/vars",
    ]
    default_vault_encrypt_identity: str | None = None
    default_vault_id_match: bool = False
    default_vault_identity: str = "default"
    default_vault_identity_list: ClassVar[list[str]] = []
    default_vault_password_file: str | None = None
    default_verbosity: int = 0
    deprecation_warnings: bool = False
    devel_warning: bool = True
    diff_always: bool = False
    diff_context: int = 3
    display_args_to_stdout: bool = False
    display_skipped_hosts: bool = True
    docsite_root_url: str = "https://docs.ansible.com/ansible/"
    doc_fragment_plugin_path: ClassVar[list[str]] = [
        "~/.ansible/plugins/doc_fragments",
        "/usr/share/ansible/plugins/doc_fragments",
    ]
    duplicate_yaml_dict_key: Literal["warn", "error", "ignore"] = "warn"
    enable_task_debugger: bool = False
    error_on_missing_handler: bool = True
    facts_modules: ClassVar[list[str]] = ["smart"]
    galaxy_cache_dir: str = "~/.ansible/galaxy_cache"
    galaxy_display_progress: str | None = None
    galaxy_ignore_certs: bool = False
    galaxy_role_skeleton: str | None = None
    galaxy_role_skeleton_ignore: ClassVar[list[str]] = ["^.git$", "^.*/.git_keep$"]
    galaxy_server: str = "https://galaxy.ansible.com"
    galaxy_server_list: str | None = None
    galaxy_token_path: str = "~/.ansible/galaxy_token"  # ruff:ignore[hardcoded-password-string]
    host_key_checking: bool = True
    host_pattern_mismatch: Literal["warning", "error", "ignore"] = "warning"
    inject_facts_as_vars: bool = True
    interpreter_python: str = "auto_legacy"
    interpreter_python_distro_map: dict[str, str]
    interpreter_python_fallback: list[str]
    invalid_task_attribute_failed: bool = True
    inventory_any_unparsed_is_failed: bool = False
    inventory_cache_enabled: bool = False
    inventory_cache_plugin: str | None = None
    inventory_cache_plugin_connection: str | None = None
    inventory_cache_plugin_prefix: str = "ansible_facts"
    inventory_cache_timeout: int = 3600
    inventory_enabled: ClassVar[list[str]] = [
        "host_list",
        "script",
        "auto",
        "yaml",
        "ini",
        "toml",
    ]
    inventory_export: bool = False
    inventory_ignore_exts: str
    inventory_ignore_patterns: ClassVar[list[str]] = []
    inventory_unparsed_is_failed: bool = False
    localhost_warning: bool = True
    max_file_size_for_diff: int = 104448
    module_ignore_exts: str
    netconf_ssh_config: str | None = None
    network_group_modules: ClassVar[list[str]] = [
        "eos",
        "nxos",
        "ios",
        "iosxr",
        "junos",
        "enos",
        "ce",
        "vyos",
        "sros",
        "dellos9",
        "dellos10",
        "dellos6",
        "asa",
        "aruba",
        "aireos",
        "bigip",
        "ironware",
        "onyx",
        "netconf",
        "exos",
        "voss",
        "slxos",
    ]
    old_plugin_cache_clearing: bool = False
    paramiko_host_key_auto_add: bool = False
    paramiko_look_for_keys: bool = True
    persistent_command_timeout: int = 30
    persistent_connect_retry_timeout: int = 15
    persistent_connect_timeout: int = 30
    persistent_control_path_dir: str = "~/.ansible/pc"
    playbook_dir: str | None
    playbook_vars_root: Literal["top", "bottom", "all"] = "top"
    plugin_filters_cfg: str | None = None
    python_module_rlimit_nofile: int = 0
    retry_files_enabled: bool = False
    retry_files_save_path: str | None = None
    run_vars_plugins: str = "demand"
    show_custom_stats: bool = False
    string_conversion_action: Literal["warn", "error", "ignore"] = "warn"
    string_type_filters: ClassVar[list[str]] = [
        "string",
        "to_json",
        "to_nice_json",
        "to_yaml",
        "to_nice_yaml",
        "ppretty",
        "json",
    ]
    system_warnings: bool = True
    tags_run: ClassVar[list[str]] = []
    tags_skip: ClassVar[list[str]] = []
    task_debugger_ignore_errors: bool = True
    task_timeout: int = 0
    transform_invalid_group_chars: Literal[
        "always",
        "never",
        "ignore",
        "silently",
    ] = "never"
    use_persistent_connections: bool = False
    variable_plugins_enabled: ClassVar[list[str]] = ["host_group_vars"]
    variable_precedence: ClassVar[list[str]] = [
        "all_inventory",
        "groups_inventory",
        "all_plugins_inventory",
        "all_plugins_play",
        "groups_plugins_inventory",
        "groups_plugins_play",
    ]
    verbose_to_stderr: bool = False
    win_async_startup_timeout: int = 5
    worker_shutdown_poll_count: int = 0
    worker_shutdown_poll_delay: float = 0.1
    yaml_filename_extensions: ClassVar[list[str]] = [".yml", ".yaml", ".json"]

    def __init__(
        self,
        config_dump: str | None = None,
        data: dict[str, object] | None = None,
        cache_dir: Path | None = None,
    ) -> None:
        """Load config dictionary."""
        super().__init__()

        self.cache_dir = cache_dir
        if data:
            self.data = copy.deepcopy(data)
        else:
            if not config_dump:
                env = os.environ.copy()
                # Avoid possible ANSI garbage
                env["ANSIBLE_FORCE_COLOR"] = "0"
                config_dump = subprocess.check_output(
                    ["ansible-config", "dump"],
                    universal_newlines=True,
                    env=env,
                )

            for match in re.finditer(
                r"^(?P<key>\w+)(?:\([^)]*\))? = (?P<value>.*)$",
                config_dump,
                re.MULTILINE,
            ):
                key = match.groupdict()["key"]
                value = match.groupdict()["value"]
                try:
                    self[key] = ast.literal_eval(value)
                except (NameError, SyntaxError, ValueError):
                    self[key] = value
        if data:
            return

    def __getattribute__(self, attr_name: str) -> object:
        """Allow access of config options as attributes."""
        parent_dict = super().__dict__  # pylint: disable=no-member
        if attr_name in parent_dict:
            return parent_dict[attr_name]

        data = super().__getattribute__("data")
        if attr_name == "data":  # pragma: no cover
            return data

        name = attr_name.upper()
        if name in data:
            return data[name]
        if name in AnsibleConfig._aliases:
            return data[AnsibleConfig._aliases[name]]

        return super().__getattribute__(attr_name)

    def __getitem__(self, name: str) -> object:
        """Allow access to config options using indexing."""
        return super().__getitem__(name.upper())

    def __copy__(self) -> AnsibleConfig:
        """Allow users to run copy on Config."""
        return AnsibleConfig(data=self.data)

    def __deepcopy__(self, memo: object) -> AnsibleConfig:
        """Allow users to run deepcopy on Config."""
        return AnsibleConfig(data=self.data)

__copy__

__copy__() -> AnsibleConfig

Allow users to run copy on Config.

Source code in src/ansible_compat/config.py

def __copy__(self) -> AnsibleConfig:
    """Allow users to run copy on Config."""
    return AnsibleConfig(data=self.data)

__deepcopy__

__deepcopy__(memo: object) -> AnsibleConfig

Allow users to run deepcopy on Config.

Source code in src/ansible_compat/config.py

def __deepcopy__(self, memo: object) -> AnsibleConfig:
    """Allow users to run deepcopy on Config."""
    return AnsibleConfig(data=self.data)

__getattribute__

__getattribute__(attr_name: str) -> object

Allow access of config options as attributes.

Source code in src/ansible_compat/config.py

def __getattribute__(self, attr_name: str) -> object:
    """Allow access of config options as attributes."""
    parent_dict = super().__dict__  # pylint: disable=no-member
    if attr_name in parent_dict:
        return parent_dict[attr_name]

    data = super().__getattribute__("data")
    if attr_name == "data":  # pragma: no cover
        return data

    name = attr_name.upper()
    if name in data:
        return data[name]
    if name in AnsibleConfig._aliases:
        return data[AnsibleConfig._aliases[name]]

    return super().__getattribute__(attr_name)

__getitem__

__getitem__(name: str) -> object

Allow access to config options using indexing.

Source code in src/ansible_compat/config.py

def __getitem__(self, name: str) -> object:
    """Allow access to config options using indexing."""
    return super().__getitem__(name.upper())

__init__

__init__(
    config_dump: str | None = None,
    data: dict[str, object] | None = None,
    cache_dir: Path | None = None,
) -> None

Source code in src/ansible_compat/config.py

def __init__(
    self,
    config_dump: str | None = None,
    data: dict[str, object] | None = None,
    cache_dir: Path | None = None,
) -> None:
    """Load config dictionary."""
    super().__init__()

    self.cache_dir = cache_dir
    if data:
        self.data = copy.deepcopy(data)
    else:
        if not config_dump:
            env = os.environ.copy()
            # Avoid possible ANSI garbage
            env["ANSIBLE_FORCE_COLOR"] = "0"
            config_dump = subprocess.check_output(
                ["ansible-config", "dump"],
                universal_newlines=True,
                env=env,
            )

        for match in re.finditer(
            r"^(?P<key>\w+)(?:\([^)]*\))? = (?P<value>.*)$",
            config_dump,
            re.MULTILINE,
        ):
            key = match.groupdict()["key"]
            value = match.groupdict()["value"]
            try:
                self[key] = ast.literal_eval(value)
            except (NameError, SyntaxError, ValueError):
                self[key] = value
    if data:
        return

ansible_version

ansible_version(version: str = '') -> Version

Return current Version object for Ansible.

If version is not mentioned, it returns current version as detected. When version argument is mentioned, it return converts the version string to Version object in order to make it usable in comparisons.

Source code in src/ansible_compat/config.py

@cache
def ansible_version(version: str = "") -> Version:
    """Return current Version object for Ansible.

    If version is not mentioned, it returns current version as detected.
    When version argument is mentioned, it return converts the version string
    to Version object in order to make it usable in comparisons.
    """
    if version:
        return Version(version)

    proc = subprocess.run(
        ["ansible", "--version"],
        text=True,
        check=False,
        capture_output=True,
    )
    if proc.returncode != 0:
        raise MissingAnsibleError(proc=proc)

    return parse_ansible_version(proc.stdout)

parse_ansible_version

parse_ansible_version(stdout: str) -> Version

Parse output of 'ansible --version'.

Source code in src/ansible_compat/config.py

def parse_ansible_version(stdout: str) -> Version:
    """Parse output of 'ansible --version'."""
    # Ansible can produce extra output before displaying version in debug mode.

    # ansible-core 2.11+: 'ansible [core 2.11.3]'
    match = re.search(
        r"^ansible \[(?:core|base) (?P<version>[^\]]+)\]",
        stdout,
        re.MULTILINE,
    )
    if match:
        return Version(match.group("version"))
    msg = f"Unable to parse ansible cli version: {stdout}\nKeep in mind that only {ANSIBLE_MIN_VERSION} or newer are supported."
    raise InvalidPrerequisiteError(msg)

ansible_compat.errors

Module to deal with errors.

AnsibleCommandError

Bases: RuntimeError

Exception running an Ansible command.

Source code in src/ansible_compat/errors.py

class AnsibleCommandError(RuntimeError):
    """Exception running an Ansible command."""

    def __init__(self, proc: CompletedProcess[Any]) -> None:
        """Construct an exception given a completed process."""
        message = (
            f"Got {proc.returncode} exit code while running: {' '.join(proc.args)}"
        )
        super().__init__(message)
        self.proc = proc

__init__

__init__(proc: CompletedProcess[Any]) -> None

Source code in src/ansible_compat/errors.py

def __init__(self, proc: CompletedProcess[Any]) -> None:
    """Construct an exception given a completed process."""
    message = (
        f"Got {proc.returncode} exit code while running: {' '.join(proc.args)}"
    )
    super().__init__(message)
    self.proc = proc

AnsibleCompatError

Bases: RuntimeError

Generic error originating from ansible_compat library.

Source code in src/ansible_compat/errors.py

class AnsibleCompatError(RuntimeError):
    """Generic error originating from ansible_compat library."""

    code = 1  # generic error

    def __init__(
        self,
        message: str | None = None,
        proc: CompletedProcess[Any] | None = None,
    ) -> None:
        """Construct generic library exception."""
        super().__init__(message)
        self.proc = proc

__init__

__init__(
    message: str | None = None,
    proc: CompletedProcess[Any] | None = None,
) -> None

Source code in src/ansible_compat/errors.py

def __init__(
    self,
    message: str | None = None,
    proc: CompletedProcess[Any] | None = None,
) -> None:
    """Construct generic library exception."""
    super().__init__(message)
    self.proc = proc

InvalidPrerequisiteError

Bases: AnsibleCompatError

Reports a missing requirement.

Source code in src/ansible_compat/errors.py

class InvalidPrerequisiteError(AnsibleCompatError):
    """Reports a missing requirement."""

    code = INVALID_PREREQUISITES_RC

MissingAnsibleError

Bases: AnsibleCompatError

Reports a missing or broken Ansible installation.

Source code in src/ansible_compat/errors.py

class MissingAnsibleError(AnsibleCompatError):
    """Reports a missing or broken Ansible installation."""

    code = ANSIBLE_MISSING_RC

    def __init__(
        self,
        message: str | None = "Unable to find a working copy of ansible executable.",
        proc: CompletedProcess[Any] | None = None,
    ) -> None:
        """."""
        super().__init__(message)
        self.proc = proc

__init__

__init__(
    message: (
        str | None
    ) = "Unable to find a working copy of ansible executable.",
    proc: CompletedProcess[Any] | None = None,
) -> None

Source code in src/ansible_compat/errors.py

def __init__(
    self,
    message: str | None = "Unable to find a working copy of ansible executable.",
    proc: CompletedProcess[Any] | None = None,
) -> None:
    """."""
    super().__init__(message)
    self.proc = proc

ansible_compat.loaders

Utilities for loading various files.

colpath_from_path

colpath_from_path(path: Path) -> str | None

Return a FQCN from a path.

Source code in src/ansible_compat/loaders.py

def colpath_from_path(path: Path) -> str | None:
    """Return a FQCN from a path."""
    galaxy_file = path / "galaxy.yml"
    if galaxy_file.exists():
        galaxy = yaml_from_file(galaxy_file)
        for k in ("namespace", "name"):
            if k not in galaxy:
                msg = f"{galaxy_file} is missing the following mandatory field {k}"
                raise InvalidPrerequisiteError(msg)
        return f"{galaxy['namespace']}/{galaxy['name']}"
    return None

yaml_from_file

yaml_from_file(path: Path) -> Any

Return a loaded YAML file.

Source code in src/ansible_compat/loaders.py

def yaml_from_file(path: Path) -> Any:  # ruff:ignore[any-type]
    """Return a loaded YAML file."""
    with path.open(encoding="utf-8") as content:
        return yaml.load(content, Loader=yaml.SafeLoader)

ansible_compat.prerun

Utilities for configuring ansible runtime environment.

No import of ansible-core must happen within this file!

get_cache_dir

get_cache_dir(
    project_dir: Path, *, isolated: bool = True
) -> Path

Compute cache directory to be used based on project path.

Parameters:

• project_dir (Path) –

  Path to the project directory.

• isolated (bool, default: True ) –

  Whether to use isolated cache directory.

Returns:

• Path –

  A writable cache directory.

Raises:

• RuntimeError –

  if cache directory is not writable.

• OSError –

  if cache directory cannot be created.

Source code in src/ansible_compat/prerun.py

def get_cache_dir(project_dir: Path, *, isolated: bool = True) -> Path:
    """Compute cache directory to be used based on project path.

    Args:
        project_dir: Path to the project directory.
        isolated: Whether to use isolated cache directory.

    Returns:
        A writable cache directory.

    Raises:
        RuntimeError: if cache directory is not writable.
        OSError: if cache directory cannot be created.
    """
    cache_dir: Path | None = None
    if "VIRTUAL_ENV" in os.environ:
        path = Path(os.environ["VIRTUAL_ENV"]).resolve() / ".ansible"
        if is_writable(path):
            cache_dir = path
        else:
            msg = f"Found VIRTUAL_ENV={os.environ['VIRTUAL_ENV']} but we cannot use it for caching as it is not writable."  # pylint: disable=W0621
            warnings.warn(
                message=msg,
                stacklevel=2,
                source={"msg": msg},
            )

    if isolated:
        project_dir = project_dir.resolve() / ".ansible"
        if is_writable(project_dir):
            cache_dir = project_dir
        else:
            msg = f"Project directory {project_dir} cannot be used for caching as it is not writable."
            warnings.warn(msg, stacklevel=2)
    else:
        cache_dir = Path(os.environ.get("ANSIBLE_HOME", "~/.ansible")).expanduser()
        # This code should be never be reached because import from ansible-core
        #  would trigger a fatal error if this location is not writable.
        if not is_writable(cache_dir):  # pragma: no cover
            msg = f"Cache directory {cache_dir} is not writable."
            raise OSError(msg)

    if not cache_dir:
        # As "project_dir" can also be "/" and user might not be able
        # to write to it, we use a temporary directory as fallback.
        checksum = hashlib.sha256(
            project_dir.as_posix().encode("utf-8"),
        ).hexdigest()[:4]

        cache_dir = Path(tempfile.gettempdir()) / f".ansible-{checksum}"
        cache_dir.mkdir(parents=True, exist_ok=True)
        msg = f"Using unique temporary directory {cache_dir} for caching."
        warnings.warn(msg, stacklevel=2)

    # Ensure basic folder structure exists so `ansible-galaxy list` does not
    # fail with: None of the provided paths were usable. Please specify a valid path with
    try:
        for name in ("roles", "collections"):
            (cache_dir / name).mkdir(parents=True, exist_ok=True)
    except OSError as exc:  # pragma: no cover
        msg = "Failed to create cache directory."
        raise RuntimeError(msg) from exc

    # We succeed only if the path is writable.
    return cache_dir

is_writable

is_writable(path: Path) -> bool

Check if path is writable, creating if necessary.

Parameters:

• path (Path) –

  Path to check.

Returns:

• bool –

  True if path is writable, False otherwise.

Source code in src/ansible_compat/prerun.py

def is_writable(path: Path) -> bool:
    """Check if path is writable, creating if necessary.

    Args:
        path: Path to check.

    Returns:
        True if path is writable, False otherwise.
    """
    try:
        path.mkdir(parents=True, exist_ok=True)
    except OSError:
        return False
    return path.exists() and os.access(path, os.W_OK)

ansible_compat.runtime

Ansible runtime environment manager.

AnsibleWarning

Bases: Warning

Warnings related to Ansible runtime.

Source code in src/ansible_compat/runtime.py

class AnsibleWarning(Warning):
    """Warnings related to Ansible runtime."""

Collection dataclass

Container for Ansible collection information.

Source code in src/ansible_compat/runtime.py

@dataclass
class Collection:
    """Container for Ansible collection information."""

    name: str
    version: str
    path: Path

CollectionVersion

Bases: Version

Collection version.

Source code in src/ansible_compat/runtime.py

class CollectionVersion(Version):
    """Collection version."""

    def __init__(self, version: str) -> None:
        """Initialize collection version."""
        # As packaging Version class does not support wildcard, we convert it
        # to "0", as this being the smallest version possible.
        if version == "*":
            version = "0"
        super().__init__(version)

__init__

__init__(version: str) -> None

Source code in src/ansible_compat/runtime.py

def __init__(self, version: str) -> None:
    """Initialize collection version."""
    # As packaging Version class does not support wildcard, we convert it
    # to "0", as this being the smallest version possible.
    if version == "*":
        version = "0"
    super().__init__(version)

Plugins dataclass

Dataclass to access installed Ansible plugins, uses ansible-doc to retrieve them.

Source code in src/ansible_compat/runtime.py

@dataclass
class Plugins:  # pylint: disable=too-many-instance-attributes
    """Dataclass to access installed Ansible plugins, uses ansible-doc to retrieve them."""

    runtime: Runtime
    become: dict[str, str] = field(init=False)
    cache: dict[str, str] = field(init=False)
    callback: dict[str, str] = field(init=False)
    cliconf: dict[str, str] = field(init=False)
    connection: dict[str, str] = field(init=False)
    httpapi: dict[str, str] = field(init=False)
    inventory: dict[str, str] = field(init=False)
    lookup: dict[str, str] = field(init=False)
    netconf: dict[str, str] = field(init=False)
    shell: dict[str, str] = field(init=False)
    vars: dict[str, str] = field(init=False)
    module: dict[str, str] = field(init=False)
    strategy: dict[str, str] = field(init=False)
    test: dict[str, str] = field(init=False)
    filter: dict[str, str] = field(init=False)
    role: dict[str, str] = field(init=False)
    keyword: dict[str, str] = field(init=False)

    @no_type_check
    def __getattribute__(self, attr: str):  # ruff:ignore[missing-return-type-special-method]
        """Get attribute.

        Raises:
            RuntimeError: If plugin loader has not been enabled yet.
            AnsibleCompatError: If unexpected output from ansible-doc.
        """
        if attr in {
            "become",
            "cache",
            "callback",
            "cliconf",
            "connection",
            "httpapi",
            "inventory",
            "lookup",
            "netconf",
            "shell",
            "vars",
            "module",
            "strategy",
            "test",
            "filter",
            "role",
            "keyword",
        }:
            try:
                result = super().__getattribute__(attr)
            except AttributeError as exc:
                # If plugin loader is not enabled yet, raise RuntimeError
                if not Runtime.plugin_loader_enabled:
                    msg = (
                        f"Plugin loader has not been enabled yet. "
                        f"Cannot access runtime.plugins.{attr} before calling "
                        f"runtime.enable_plugin_loader()."
                    )
                    raise RuntimeError(msg) from exc
                proc = self.runtime.run(
                    ["ansible-doc", "--json", "-l", "-t", attr],
                )
                data = json.loads(proc.stdout)
                if not isinstance(data, dict):  # pragma: no cover
                    msg = "Unexpected output from ansible-doc"
                    raise AnsibleCompatError(msg) from exc
                result = data
        else:
            result = super().__getattribute__(attr)

        return result

__getattribute__

__getattribute__(attr: str)

Get attribute.

Raises:

• RuntimeError –

  If plugin loader has not been enabled yet.

• AnsibleCompatError –

  If unexpected output from ansible-doc.

Source code in src/ansible_compat/runtime.py

@no_type_check
def __getattribute__(self, attr: str):  # ruff:ignore[missing-return-type-special-method]
    """Get attribute.

    Raises:
        RuntimeError: If plugin loader has not been enabled yet.
        AnsibleCompatError: If unexpected output from ansible-doc.
    """
    if attr in {
        "become",
        "cache",
        "callback",
        "cliconf",
        "connection",
        "httpapi",
        "inventory",
        "lookup",
        "netconf",
        "shell",
        "vars",
        "module",
        "strategy",
        "test",
        "filter",
        "role",
        "keyword",
    }:
        try:
            result = super().__getattribute__(attr)
        except AttributeError as exc:
            # If plugin loader is not enabled yet, raise RuntimeError
            if not Runtime.plugin_loader_enabled:
                msg = (
                    f"Plugin loader has not been enabled yet. "
                    f"Cannot access runtime.plugins.{attr} before calling "
                    f"runtime.enable_plugin_loader()."
                )
                raise RuntimeError(msg) from exc
            proc = self.runtime.run(
                ["ansible-doc", "--json", "-l", "-t", attr],
            )
            data = json.loads(proc.stdout)
            if not isinstance(data, dict):  # pragma: no cover
                msg = "Unexpected output from ansible-doc"
                raise AnsibleCompatError(msg) from exc
            result = data
    else:
        result = super().__getattribute__(attr)

    return result

Runtime

Ansible Runtime manager.

Attributes:

• _version (Version | None) –

  Cached version of Ansible.

• collections (OrderedDict[str, Collection]) –

  Ordered dictionary of collections.

• cache_dir (Path) –

  Path to cache directory.

• initialized (bool) –

  Flag tracking if Ansible runtime has been initialized.

• plugin_loader_enabled (bool) –

  Flag controlling when plugin loader initialization is allowed.

• plugins (Plugins) –

  Plugins instance for accessing Ansible plugins.

• _has_playbook_cache (dict[tuple[str, Path | None], bool]) –

  Cache for playbook existence checks.

• require_module (bool) –

  Flag indicating if module is required.

Source code in src/ansible_compat/runtime.py

class Runtime:
    """Ansible Runtime manager.

    Attributes:
        _version: Cached version of Ansible.
        collections: Ordered dictionary of collections.
        cache_dir: Path to cache directory.
        initialized: Flag tracking if Ansible runtime has been initialized.
        plugin_loader_enabled: Flag controlling when plugin loader initialization is allowed.
        plugins: Plugins instance for accessing Ansible plugins.
        _has_playbook_cache: Cache for playbook existence checks.
        require_module: Flag indicating if module is required.
    """

    _version: Version | None = None
    collections: OrderedDict[str, Collection]
    cache_dir: Path
    # Used to track if we have already initialized the Ansible runtime as attempts
    # to do it multiple times will cause runtime warnings from within ansible-core
    initialized: bool = False
    # Flag to control when plugin loader initialization is allowed
    plugin_loader_enabled: bool = False
    plugins: Plugins
    _has_playbook_cache: ClassVar[dict[tuple[str, Path | None], bool]] = {}
    require_module: bool = False

    def __init__(
        self,
        project_dir: Path | None = None,
        *,
        isolated: bool = False,
        min_required_version: str | None = None,
        require_module: bool = False,
        max_retries: int = 0,
        environ: dict[str, str] | None = None,
        verbosity: int = 0,
    ) -> None:
        """Initialize Ansible runtime environment.

        :param project_dir: The directory containing the Ansible project. If
                            not mentioned it will be guessed from the current
                            working directory.
        :param isolated: Assure that installation of collections or roles
                         does not affect Ansible installation, an unique cache
                         directory being used instead.
        :param min_required_version: Minimal version of Ansible required. If
                                     not found, a :class:`RuntimeError`
                                     exception is raised.
        :param require_module: If set, instantiation will fail if Ansible
                               Python module is missing or is not matching
                               the same version as the Ansible command line.
                               That is useful for consumers that expect to
                               also perform Python imports from Ansible.
        :param max_retries: Number of times it should retry network operations.
                            Default is 0, no retries.
        :param environ: Environment dictionary to use, if undefined
                        ``os.environ`` will be copied and used.
        :param verbosity: Verbosity level to use.
        """
        self.project_dir = project_dir or Path.cwd()
        self.isolated = isolated
        self.max_retries = max_retries
        self.environ = environ or os.environ.copy()
        if "ANSIBLE_COLLECTIONS_PATHS" in self.environ:
            msg = "ANSIBLE_COLLECTIONS_PATHS was detected, replace it with ANSIBLE_COLLECTIONS_PATH to continue."
            raise RuntimeError(msg)
        self.plugins = Plugins(runtime=self)
        self.verbosity = verbosity

        self.initialize_logger(level=self.verbosity)

        # Reduce noise from paramiko, unless user already defined PYTHONWARNINGS
        # paramiko/transport.py:236: CryptographyDeprecationWarning: Blowfish has been deprecated
        # https://github.com/paramiko/paramiko/issues/2038
        # As CryptographyDeprecationWarning is not a builtin, we cannot use
        # PYTHONWARNINGS to ignore it using category but we can use message.
        # https://stackoverflow.com/q/68251969/99834
        if "PYTHONWARNINGS" not in self.environ:  # pragma: no cover
            self.environ["PYTHONWARNINGS"] = "ignore:Blowfish has been deprecated"

        self.cache_dir = get_cache_dir(self.project_dir, isolated=self.isolated)

        self.config = AnsibleConfig(cache_dir=self.cache_dir)

        # Add the sys.path to the collection paths if not isolated
        self._patch_collection_paths()

        if not self.version_in_range(lower=min_required_version):
            msg = f"Found incompatible version of ansible runtime {self.version}, instead of {min_required_version} or newer."
            raise RuntimeError(msg)
        if require_module:
            self.require_module = True
            self._ensure_module_available()

        # For backwards compatibility with ansible-lint < 25.8.1
        if should_auto_enable_plugin_loader():
            self.enable_plugin_loader()

        # pylint: disable=import-outside-toplevel
        from ansible.utils.display import Display

        # pylint: disable=unused-argument
        def warning(  # noqa: DOC103
            self: Display,  # ruff:ignore[unused-function-argument]
            msg: str,
            formatted: bool = False,  # ruff:ignore[unused-function-argument, boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
            *,
            help_text: str | None = None,  # ruff:ignore[unused-function-argument]
            obj: Any = None,  # ruff:ignore[unused-function-argument, any-type]
        ) -> None:  # pragma: no cover
            """Override ansible.utils.display.Display.warning to avoid printing warnings."""
            warnings.warn(
                message=msg,
                category=AnsibleWarning,
                stacklevel=2,
                source={"msg": msg},
            )

        # Monkey patch ansible warning in order to use warnings module.
        Display.warning = warning

    def initialize_logger(self, level: int = 0) -> None:  # ruff:ignore[no-self-use]
        """Set up the global logging level based on the verbosity number."""
        verbosity_map = {
            -2: logging.CRITICAL,
            -1: logging.ERROR,
            0: logging.WARNING,
            1: logging.INFO,
            2: logging.DEBUG,
        }
        # Unknown logging level is treated as DEBUG
        logging_level = verbosity_map.get(level, logging.DEBUG)
        _logger.setLevel(logging_level)
        # Use module-level _logger instance to validate it
        _logger.debug("Logging initialized to level %s", logging_level)

    def _patch_collection_paths(self) -> None:
        """Modify Ansible collection path for testing purposes.

        - Add the sys.path to the end of collection paths.
        - Add the site-packages to the beginning of collection paths to match
          ansible-core and ade behavior and trick ansible-galaxy to install
          default to the venv site-packages location (isolation).
        """
        # ansible-core normal precedence is: adjacent, local paths, configured paths, site paths
        collections_paths: list[str] = self.config.collections_paths.copy()
        if self.config.collections_scan_sys_path:
            for path in sys.path:
                if (
                    path not in collections_paths
                    and (Path(path) / "ansible_collections").is_dir()
                ):
                    collections_paths.append(  # pylint: disable=E1101
                        path,
                    )
            # When inside a venv, we also add the site-packages to the end of the
            # collections path because this is the last place where ansible-core
            # will look for them. This also ensures that when calling ansible-galaxy
            # to install content.
            for path in reversed(site.getsitepackages()):
                if path not in collections_paths:
                    collections_paths.append(path)

            if collections_paths != self.config.collections_paths:
                _logger.info(
                    "Collection paths was patched to include extra directories %s",
                    ",".join(collections_paths),
                )
        else:
            msg = "ANSIBLE_COLLECTIONS_SCAN_SYS_PATH is disabled, not patching collection paths. This may lead to unexpected behavior when using dev tools and prevent full isolation from user environment."
            _logger.warning(msg)
        self.config.collections_paths = collections_paths

    def load_collections(self) -> None:
        """Load collection data."""
        self.collections = OrderedDict()
        no_collections_msg = "None of the provided paths were usable"

        # do not use --path because it does not allow multiple values
        proc = self.run(
            [
                "ansible-galaxy",
                "collection",
                "list",
                "--format=json",
            ],
        )
        if proc.returncode == RC_ANSIBLE_OPTIONS_ERROR and (
            no_collections_msg in proc.stdout or no_collections_msg in proc.stderr
        ):  # pragma: no cover
            _logger.debug("Ansible reported no installed collections at all.")
            return
        if proc.returncode != 0:
            _logger.error(proc)
            msg = f"Unable to list collections: {proc}"
            raise RuntimeError(msg)
        try:
            data = json.loads(proc.stdout)
        except json.decoder.JSONDecodeError as exc:
            msg = f"Unable to parse galaxy output as JSON: {proc.stdout}"
            raise RuntimeError(msg) from exc
        if not isinstance(data, dict):
            msg = f"Unexpected collection data, {data}"
            raise TypeError(msg)
        self._parse_galaxy_collection_list(data)

    def _parse_galaxy_collection_list(self, data: dict[str, Any]) -> None:
        """Parse ansible-galaxy collection list JSON output into self.collections.

        :param data: Parsed JSON output from ``ansible-galaxy collection list --format=json``.

        Raises:
            TypeError: If the collection data has an unexpected type.
        """
        for path, path_data in data.items():
            if not isinstance(path_data, dict):
                msg = f"Unexpected collection data, {path_data}"
                raise TypeError(msg)
            for collection, collection_info in path_data.items():
                if not isinstance(collection_info, dict):
                    msg = f"Unexpected collection data, {collection_info}"
                    raise TypeError(msg)

                if collection in self.collections:
                    msg = f"Another version of '{collection}' {collection_info['version']} was found installed in {path}, only the first one will be used, {self.collections[collection].version} ({self.collections[collection].path})."
                    _logger.warning(msg)
                else:
                    self.collections[collection] = Collection(
                        name=collection,
                        version=collection_info["version"],
                        path=Path(path),
                    )

    def _ensure_module_available(self) -> None:
        """Assure that Ansible Python module is installed and matching CLI version."""
        ansible_release_module = None
        with contextlib.suppress(ModuleNotFoundError, ImportError):
            ansible_release_module = importlib.import_module("ansible.release")

        if ansible_release_module is None:
            msg = "Unable to find Ansible python module."
            raise RuntimeError(msg)

        ansible_module_version = Version(
            ansible_release_module.__version__,
        )
        if ansible_module_version != self.version:
            msg = f"Ansible CLI ({self.version}) and python module ({ansible_module_version}) versions do not match. This indicates a broken execution environment."
            raise RuntimeError(msg)

    def enable_plugin_loader(self) -> None:
        """Enable plugin loader initialization with current collections paths."""
        Runtime.plugin_loader_enabled = True

        # Initialize the plugin loader now with the full collections paths
        if not Runtime.initialized:
            col_paths = self.config.collections_paths
            if self.isolated:
                col_paths = [str(self.cache_dir / "collections"), *col_paths]
            # ruff: disable[import-private-name]
            # noinspection PyProtectedMember
            # pylint: disable=import-outside-toplevel,no-name-in-module
            from ansible.plugins.loader import init_plugin_loader
            from ansible.utils.collection_loader._collection_finder import (  # pylint: disable=import-outside-toplevel
                _AnsibleCollectionFinder,
            )
            # ruff: enable[import-private-name]

            with contextlib.suppress(Exception):
                _AnsibleCollectionFinder()._remove()  # pylint: disable=protected-access  # ruff:ignore[private-member-access]

            init_plugin_loader(col_paths)
            Runtime.initialized = True

    def clean(self) -> None:
        """Remove content of cache_dir and reset global Runtime state."""
        shutil.rmtree(self.cache_dir, ignore_errors=True)

        # Reset global Runtime class state to allow fresh initialization
        Runtime.initialized = False
        Runtime.plugin_loader_enabled = False

        # Unload ansible modules that cache plugin/collection state
        # This ensures fresh plugin loader and collection finder state
        modules_to_remove = [
            mod
            for mod in sys.modules
            if mod.startswith(
                (
                    "ansible.plugins.",  # Plugin loaders and cached plugins
                    "ansible.utils.collection_loader",  # Collection finder system
                    "ansible.collections.",  # Loaded collection modules
                ),
            )
        ]
        for mod in modules_to_remove:
            del sys.modules[mod]

    def run(
        self,
        args: str | list[str],
        *,
        retry: bool = False,
        tee: bool = False,
        env: dict[str, str] | None = None,
        cwd: Path | None = None,
        set_acp: bool = True,
    ) -> CompletedProcess:
        """Execute a command inside an Ansible environment.

        :param retry: Retry network operations on failures.
        :param tee: Also pass captured stdout/stderr to system while running.
        :param set_acp: Set the ANSIBLE_COLLECTIONS_PATH
        """
        if tee:
            run_func: Callable[..., CompletedProcess] = subprocess_tee.run
        else:
            run_func = subprocess.run
        env = self.environ if env is None else env.copy()
        # Presence of ansible debug variable or config option will prevent us
        # from parsing its JSON output due to extra debug messages on stdout.
        env["ANSIBLE_DEBUG"] = "0"

        # https://github.com/ansible/ansible-lint/issues/3522
        env["ANSIBLE_VERBOSE_TO_STDERR"] = "True"

        if set_acp:
            env["ANSIBLE_COLLECTIONS_PATH"] = ":".join(
                list(dict.fromkeys(self.config.collections_paths)),
            )

        for _ in range(self.max_retries + 1 if retry else 1):
            result = run_func(
                args,
                universal_newlines=True,
                check=False,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=env,
                cwd=str(cwd) if cwd else None,
            )
            if result.returncode == 0:
                break
            _logger.debug("Environment: %s", env)
            if retry:
                _logger.warning(
                    "Retrying execution failure %s of: %s",
                    result.returncode,
                    " ".join(args),
                )
        return result

    @property
    def version(self) -> Version:
        """Current Version object for Ansible.

        If version is not mentioned, it returns current version as detected.
        When version argument is mentioned, it return converts the version string
        to Version object in order to make it usable in comparisons.
        """
        if self._version:
            return self._version

        proc = self.run(["ansible", "--version"])
        if proc.returncode == 0:
            self._version = parse_ansible_version(proc.stdout)
            return self._version

        msg = "Unable to find a working copy of ansible executable."
        raise MissingAnsibleError(msg, proc=proc)

    def version_in_range(
        self,
        lower: str | None = None,
        upper: str | None = None,
    ) -> bool:
        """Check if Ansible version is inside a required range.

        The lower limit is inclusive and the upper one exclusive.
        """
        if lower and self.version < Version(lower):
            return False
        return not (upper and self.version >= Version(upper))

    def has_playbook(self, playbook: str, *, basedir: Path | None = None) -> bool:
        """Return true if ansible can load a given playbook.

        This is also used for checking if playbooks from within collections
        are present and if they pass syntax check.
        """
        if (playbook, basedir) in self._has_playbook_cache:
            return self._has_playbook_cache[playbook, basedir]

        proc = self.run(["ansible-playbook", "--syntax-check", playbook], cwd=basedir)
        result = proc.returncode == 0
        if not result:
            if not basedir:
                basedir = Path()
            msg = f"has_playbook returned false for '{basedir / playbook}' due to syntax check returning {proc.returncode}"
            _logger.debug(msg)

        # cache the result
        self._has_playbook_cache[playbook, basedir] = result

        return result

    def install_collection(
        self,
        collection: str | Path,
        *,
        destination: Path | None = None,
        force: bool = False,
    ) -> None:
        """Install an Ansible collection.

        Can accept arguments like:
            'foo.bar:>=1.2.3'
            'git+https://github.com/ansible-collections/ansible.posix.git,main'
        """
        cmd = [
            "ansible-galaxy",
            "collection",
            "install",
            "-vvv",  # this is needed to make ansible display important info in case of failures
        ]
        if force:
            cmd.append("--force")

        if isinstance(collection, Path):
            collection = str(collection)
        # As ansible-galaxy install is not able to automatically determine
        # if the range requires a pre-release, we need to manually add the --pre
        # flag when needed.
        matches = version_re.search(collection)

        if (
            not is_url(collection)
            and matches
            and CollectionVersion(matches[1]).is_prerelease
        ):
            cmd.append("--pre")

        cpaths: list[str] = self.config.collections_paths
        if destination and str(destination) not in cpaths:
            # we cannot use '-p' because it breaks galaxy ability to ignore already installed collections, so
            # we hack ANSIBLE_COLLECTIONS_PATH instead and inject our own path there.
            # pylint: disable=no-member
            cpaths.insert(0, str(destination))
        cmd.append(f"{collection}")

        _logger.info("Running from %s : %s", Path.cwd(), " ".join(cmd))
        process = self.run(
            cmd,
            retry=True,
            env={**self.environ, "ANSIBLE_COLLECTIONS_PATH": ":".join(cpaths)},
        )
        if process.returncode != 0:
            msg = f"Command {' '.join(cmd)}, returned {process.returncode} code:\n{process.stdout}\n{process.stderr}"
            _logger.error(msg)
            raise InvalidPrerequisiteError(msg)

    def install_collection_from_disk(
        self,
        path: Path,
        destination: Path | None = None,
    ) -> None:
        """Build and install collection from a given disk path."""
        self.install_collection(path, destination=destination, force=True)

    def install_requirements(
        self,
        requirement: Path,
        *,
        retry: bool = False,
        offline: bool = False,
    ) -> None:
        """Install dependencies from a requirements.yml.

        :param requirement: path to requirements.yml file
        :param retry: retry network operations on failures
        :param offline: bypass installation, may fail if requirements are not met.
        """
        if not Path(requirement).exists():
            return
        reqs_yaml = yaml_from_file(Path(requirement))
        _validate_requirements_yaml(reqs_yaml, requirement)

        if isinstance(reqs_yaml, list) or "roles" in reqs_yaml:
            self._install_role_requirements(requirement, retry=retry, offline=offline)

        if (
            isinstance(reqs_yaml, dict)
            and "collections" in reqs_yaml
            and reqs_yaml["collections"] is not None
        ):
            self._install_collection_requirements(
                requirement,
                reqs_yaml,
                retry=retry,
                offline=offline,
            )

        if self.require_module:
            Runtime.initialized = False
            self._ensure_module_available()

    def _install_role_requirements(
        self,
        requirement: Path,
        *,
        retry: bool,
        offline: bool,
    ) -> None:
        """Install role dependencies from a requirements file.

        :param requirement: path to requirements.yml file
        :param retry: retry network operations on failures
        :param offline: bypass installation, may fail if requirements are not met.

        Raises:
            AnsibleCommandError: If the ansible-galaxy command fails.
        """
        cmd = [
            "ansible-galaxy",
            "role",
            "install",
            "-r",
            f"{requirement}",
        ]
        if self.verbosity > 0:
            cmd.extend(["-" + ("v" * self.verbosity)])
        cmd.extend(["--roles-path", f"{self.cache_dir}/roles"])

        if offline:
            _logger.warning(
                "Skipped installing old role dependencies due to running in offline mode.",
            )
        else:
            _logger.info("Running %s", " ".join(cmd))

            result = self.run(cmd, retry=retry)
            _logger.debug(result.stdout)
            if result.returncode != 0:
                _logger.error(result.stderr)
                raise AnsibleCommandError(result)

    def _install_collection_requirements(
        self,
        requirement: Path,
        reqs_yaml: dict[str, Any],
        *,
        retry: bool,
        offline: bool,
    ) -> None:
        """Install collection dependencies from a v2 requirements file.

        :param requirement: path to requirements.yml file
        :param reqs_yaml: parsed requirements YAML (must contain a ``collections`` key)
        :param retry: retry network operations on failures
        :param offline: bypass installation, may fail if requirements are not met.

        Raises:
            AnsibleCommandError: If the ansible-galaxy command fails.
        """
        cmd = [
            "ansible-galaxy",
            "collection",
            "install",
        ]
        if self.verbosity > 0:
            cmd.extend(["-" + ("v" * self.verbosity)])

        for collection in reqs_yaml["collections"]:
            if isinstance(collection, dict) and collection.get("type", "") == "git":
                _logger.info(
                    "Adding '--pre' to ansible-galaxy collection install because we detected one collection being sourced from git.",
                )
                cmd.append("--pre")
                break
        if offline:
            _logger.warning(
                "Skipped installing collection dependencies due to running in offline mode.",
            )
        else:
            cmd.extend(["-r", str(requirement)])
            _logger.info("Running %s", " ".join(cmd))
            result = self.run(
                cmd,
                retry=retry,
            )
            _logger.debug(result.stdout)
            if result.returncode != 0:
                _logger.error(result.stderr)
                raise AnsibleCommandError(result)

    def prepare_environment(
        self,
        required_collections: dict[str, str] | None = None,
        *,
        retry: bool = False,
        install_local: bool = False,
        offline: bool = False,
        role_name_check: int = 0,
    ) -> None:
        """Make dependencies available if needed."""
        destination: Path = self.cache_dir / "collections"
        if required_collections is None:
            required_collections = {}

        self._prepare_ansible_paths()

        # first one is standard for collection layout repos and the last two
        # are part of Tower specification
        # https://docs.ansible.com/ansible-tower/latest/html/userguide/projects.html#ansible-galaxy-support
        # https://docs.ansible.com/ansible-tower/latest/html/userguide/projects.html#collections-support
        for req_file in REQUIREMENT_LOCATIONS:
            file_path = Path(req_file)
            if self.project_dir:
                file_path = self.project_dir / req_file
            self.install_requirements(file_path, retry=retry, offline=offline)

        if not install_local:
            return

        for item in search_galaxy_paths(self.project_dir):
            if item.exists():
                self._install_deps_from_galaxy(item, destination)

        for name, min_version in required_collections.items():
            self.install_collection(
                f"{name}:>={min_version}",
                destination=destination,
            )

        self._handle_local_install(destination, role_name_check)

    def _install_deps_from_galaxy(
        self,
        galaxy_path: Path,
        destination: Path,
    ) -> None:
        """Install collection dependencies declared in a galaxy.yml file.

        :param galaxy_path: Path to a ``galaxy.yml`` file.
        :param destination: Target collections directory.
        """
        data = yaml_from_file(galaxy_path)
        if not isinstance(data, dict) or "dependencies" not in data:
            return
        for name, required_version in data["dependencies"].items():
            _logger.info(
                "Provisioning collection %s:%s from galaxy.yml",
                name,
                required_version,
            )
            self.install_collection(
                f"{name}{',' if is_url(name) else ':'}{required_version}",
                destination=destination,
            )

    def _handle_local_install(
        self,
        destination: Path,
        role_name_check: int,
    ) -> None:
        """Install the local collection or role and reload collections.

        :param destination: Target collections directory.
        :param role_name_check: Role name validation level (0/1/2).
        """
        galaxy_path = self.project_dir / GALAXY_YML
        if galaxy_path.exists():
            colpath = Path(
                f"{destination}/ansible_collections/{colpath_from_path(self.project_dir)}",
            )
            if colpath.is_symlink():
                if os.path.realpath(colpath) == str(Path.cwd()):
                    _logger.warning(
                        "Found symlinked collection, skipping its installation.",
                    )
                    return
                _logger.warning(
                    "Collection is symlinked, but not pointing to %s directory, so we will remove it.",
                    Path.cwd(),
                )
                colpath.unlink()

            self.install_collection_from_disk(
                galaxy_path.parent,
                destination=destination,
            )
        elif Path.cwd().parent.name == "roles" and Path("../../galaxy.yml").exists():
            self.install_collection_from_disk(
                Path("../.."),
                destination=destination,
            )
        else:
            self._install_galaxy_role(
                self.project_dir,
                role_name_check=role_name_check,
                ignore_errors=True,
            )
        self.load_collections()

    def require_collection(
        self,
        name: str,
        version: str | None = None,
        *,
        install: bool = True,
    ) -> tuple[CollectionVersion, Path]:
        """Check if a minimal collection version is present or exits.

        In the future this method may attempt to install a missing or outdated
        collection before failing.

        Args:
            name: collection name
            version: minimal version required
            install: if True, attempt to install a missing collection

        Returns:
            tuple of (found_version, collection_path)
        """
        try:
            ns, coll = name.split(".", 1)
        except ValueError as exc:
            msg = f"Invalid collection name supplied: {name}%s"
            raise InvalidPrerequisiteError(
                msg,
            ) from exc

        paths: list[str] = self.config.collections_paths
        if not paths or not isinstance(paths, list):
            msg = f"Unable to determine ansible collection paths. ({paths})"
            raise InvalidPrerequisiteError(
                msg,
            )

        for path in paths:
            collpath = Path(path) / "ansible_collections" / ns / coll
            result = self._check_collection_at_path(
                collpath,
                name=name,
                version=version,
                install=install,
            )
            if result is not None:
                return result
        if install:
            self.install_collection(f"{name}:>={version}" if version else name)
            return self.require_collection(
                name=name,
                version=version,
                install=False,
            )
        msg = f"Collection '{name}' not found in '{paths}'"
        _logger.fatal(msg)
        raise InvalidPrerequisiteError(msg)

    def _check_collection_at_path(
        self,
        collpath: Path,
        *,
        name: str,
        version: str | None,
        install: bool,
    ) -> tuple[CollectionVersion, Path] | None:
        """Return (version, path) if the collection exists at *collpath*, else None.

        When the installed version is too old and *install* is True the
        collection is upgraded in-place before returning.

        Raises:
            InvalidPrerequisiteError: If MANIFEST.json is missing or version is too old.
        """
        if not collpath.exists():
            return None
        mpath = collpath / "MANIFEST.json"
        if not mpath.exists():
            msg = f"Found collection at '{collpath}' but missing MANIFEST.json, cannot get info."
            _logger.fatal(msg)
            raise InvalidPrerequisiteError(msg)

        with mpath.open(encoding="utf-8") as f:
            manifest = json.loads(f.read())
            found_version = CollectionVersion(
                manifest["collection_info"]["version"],
            )
            if version and found_version < CollectionVersion(version):
                if install:
                    self.install_collection(f"{name}:>={version}")
                    self.require_collection(name, version, install=False)
                else:
                    msg = f"Found {name} collection {found_version} but {version} or newer is required."
                    _logger.fatal(msg)
                    raise InvalidPrerequisiteError(msg)
            return found_version, collpath.resolve()

    def _prepare_ansible_paths(self) -> None:
        """Configure Ansible environment variables."""
        try:
            library_paths: list[str] = self.config.default_module_path.copy()
            roles_path: list[str] = self.config.default_roles_path.copy()
            collections_path: list[str] = self.config.collections_paths.copy()
        except AttributeError as exc:
            msg = "Unexpected ansible configuration"
            raise RuntimeError(msg) from exc

        alterations_list: list[tuple[list[str], str, bool]] = [
            (
                library_paths,
                str((self.project_dir / "plugins" / "modules").absolute()),
                True,
            ),
            (roles_path, str((self.project_dir / "roles").absolute()), True),
        ]

        alterations_list.extend(
            (
                [
                    (roles_path, f"{self.cache_dir}/roles", False),
                    (library_paths, f"{self.cache_dir}/modules", False),
                    (collections_path, f"{self.cache_dir}/collections", False),
                ]
                if self.isolated
                else []
            ),
        )

        for path_list, path_, must_be_present in alterations_list:
            path = Path(path_)
            if not path.exists():
                if must_be_present:
                    continue
                path.mkdir(parents=True, exist_ok=True)
            if str(path) not in path_list:
                path_list.insert(0, str(path))

        if library_paths != self.config.DEFAULT_MODULE_PATH:
            self._update_env("ANSIBLE_LIBRARY", library_paths)
        if collections_path != self.config.default_collections_path:
            self._update_env("ANSIBLE_COLLECTIONS_PATH", collections_path)
        if roles_path != self.config.default_roles_path:
            self._update_env("ANSIBLE_ROLES_PATH", roles_path)

    def _get_roles_path(self) -> Path:
        """Return roles installation path.

        If `self.isolated` is set to `True`, `self.cache_dir` would be
        created, then it returns the `self.cache_dir/roles`. When `self.isolated` is
        not mentioned or set to `False`, it returns the first path in
        `default_roles_path`.
        """
        path = Path(f"{self.cache_dir}/roles")
        return path

    def _install_galaxy_role(
        self,
        project_dir: Path,
        role_name_check: int = 0,
        *,
        ignore_errors: bool = False,
    ) -> None:
        """Detect standalone galaxy role and installs it.

        Args:
            project_dir: path to the role
            role_name_check: logic to used to check role name
                0: exit with error if name is not compliant (default)
                1: warn if name is not compliant
                2: bypass any name checking

            ignore_errors: if True, bypass installing invalid roles.

        Our implementation aims to match ansible-galaxy's behavior for installing
        roles from a tarball or scm. For example ansible-galaxy will install a role
        that has both galaxy.yml and meta/main.yml present but empty. Also missing
        galaxy.yml is accepted but missing meta/main.yml is not.
        """
        yaml = None
        galaxy_info = {}

        for meta_main in META_MAIN:
            meta_filename = Path(project_dir) / meta_main

            if meta_filename.exists():
                break
        else:
            if ignore_errors:
                return

        yaml = yaml_from_file(meta_filename)

        if yaml and "galaxy_info" in yaml:
            galaxy_info = yaml["galaxy_info"]

        fqrn = _resolve_role_fqrn_with_check(galaxy_info, project_dir, role_name_check)

        path = self._get_roles_path()
        path.mkdir(parents=True, exist_ok=True)
        link_path = path / fqrn
        # despite documentation stating that is_file() reports true for symlinks,
        # it appears that is_dir() reports true instead, so we rely on exists().
        target = Path(project_dir).absolute()
        if not link_path.exists() or (
            link_path.is_symlink() and link_path.readlink() != target
        ):
            # must call unlink before checking exists because a broken
            # link reports as not existing and we want to repair it
            link_path.unlink(missing_ok=True)
            # https://github.com/python/cpython/issues/73843
            link_path.symlink_to(str(target), target_is_directory=True)
        _logger.info(
            "Using %s symlink to current repository in order to enable Ansible to find the role using its expected full name.",
            link_path,
        )

    def _update_env(self, varname: str, value: list[str], default: str = "") -> None:
        """Update colon based environment variable if needed.

        New values are prepended to make sure they take precedence.
        """
        if not value:
            return
        orig_value = self.environ.get(varname, default)
        if orig_value:
            # we just want to avoid repeating the same entries, but order is important
            value = list(dict.fromkeys([*value, *orig_value.split(":")]))
        value_str = ":".join(value)
        if value_str != self.environ.get(varname, ""):
            self.environ[varname] = value_str
            _logger.info("Set %s=%s", varname, value_str)

version property

version: Version

Current Version object for Ansible.

If version is not mentioned, it returns current version as detected. When version argument is mentioned, it return converts the version string to Version object in order to make it usable in comparisons.

__init__

__init__(
    project_dir: Path | None = None,
    *,
    isolated: bool = False,
    min_required_version: str | None = None,
    require_module: bool = False,
    max_retries: int = 0,
    environ: dict[str, str] | None = None,
    verbosity: int = 0
) -> None

:param project_dir: The directory containing the Ansible project. If not mentioned it will be guessed from the current working directory. :param isolated: Assure that installation of collections or roles does not affect Ansible installation, an unique cache directory being used instead. :param min_required_version: Minimal version of Ansible required. If not found, a :class:RuntimeError exception is raised. :param require_module: If set, instantiation will fail if Ansible Python module is missing or is not matching the same version as the Ansible command line. That is useful for consumers that expect to also perform Python imports from Ansible. :param max_retries: Number of times it should retry network operations. Default is 0, no retries. :param environ: Environment dictionary to use, if undefined os.environ will be copied and used. :param verbosity: Verbosity level to use.

Source code in src/ansible_compat/runtime.py

def __init__(
    self,
    project_dir: Path | None = None,
    *,
    isolated: bool = False,
    min_required_version: str | None = None,
    require_module: bool = False,
    max_retries: int = 0,
    environ: dict[str, str] | None = None,
    verbosity: int = 0,
) -> None:
    """Initialize Ansible runtime environment.

    :param project_dir: The directory containing the Ansible project. If
                        not mentioned it will be guessed from the current
                        working directory.
    :param isolated: Assure that installation of collections or roles
                     does not affect Ansible installation, an unique cache
                     directory being used instead.
    :param min_required_version: Minimal version of Ansible required. If
                                 not found, a :class:`RuntimeError`
                                 exception is raised.
    :param require_module: If set, instantiation will fail if Ansible
                           Python module is missing or is not matching
                           the same version as the Ansible command line.
                           That is useful for consumers that expect to
                           also perform Python imports from Ansible.
    :param max_retries: Number of times it should retry network operations.
                        Default is 0, no retries.
    :param environ: Environment dictionary to use, if undefined
                    ``os.environ`` will be copied and used.
    :param verbosity: Verbosity level to use.
    """
    self.project_dir = project_dir or Path.cwd()
    self.isolated = isolated
    self.max_retries = max_retries
    self.environ = environ or os.environ.copy()
    if "ANSIBLE_COLLECTIONS_PATHS" in self.environ:
        msg = "ANSIBLE_COLLECTIONS_PATHS was detected, replace it with ANSIBLE_COLLECTIONS_PATH to continue."
        raise RuntimeError(msg)
    self.plugins = Plugins(runtime=self)
    self.verbosity = verbosity

    self.initialize_logger(level=self.verbosity)

    # Reduce noise from paramiko, unless user already defined PYTHONWARNINGS
    # paramiko/transport.py:236: CryptographyDeprecationWarning: Blowfish has been deprecated
    # https://github.com/paramiko/paramiko/issues/2038
    # As CryptographyDeprecationWarning is not a builtin, we cannot use
    # PYTHONWARNINGS to ignore it using category but we can use message.
    # https://stackoverflow.com/q/68251969/99834
    if "PYTHONWARNINGS" not in self.environ:  # pragma: no cover
        self.environ["PYTHONWARNINGS"] = "ignore:Blowfish has been deprecated"

    self.cache_dir = get_cache_dir(self.project_dir, isolated=self.isolated)

    self.config = AnsibleConfig(cache_dir=self.cache_dir)

    # Add the sys.path to the collection paths if not isolated
    self._patch_collection_paths()

    if not self.version_in_range(lower=min_required_version):
        msg = f"Found incompatible version of ansible runtime {self.version}, instead of {min_required_version} or newer."
        raise RuntimeError(msg)
    if require_module:
        self.require_module = True
        self._ensure_module_available()

    # For backwards compatibility with ansible-lint < 25.8.1
    if should_auto_enable_plugin_loader():
        self.enable_plugin_loader()

    # pylint: disable=import-outside-toplevel
    from ansible.utils.display import Display

    # pylint: disable=unused-argument
    def warning(  # noqa: DOC103
        self: Display,  # ruff:ignore[unused-function-argument]
        msg: str,
        formatted: bool = False,  # ruff:ignore[unused-function-argument, boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
        *,
        help_text: str | None = None,  # ruff:ignore[unused-function-argument]
        obj: Any = None,  # ruff:ignore[unused-function-argument, any-type]
    ) -> None:  # pragma: no cover
        """Override ansible.utils.display.Display.warning to avoid printing warnings."""
        warnings.warn(
            message=msg,
            category=AnsibleWarning,
            stacklevel=2,
            source={"msg": msg},
        )

    # Monkey patch ansible warning in order to use warnings module.
    Display.warning = warning

clean

clean() -> None

Remove content of cache_dir and reset global Runtime state.

Source code in src/ansible_compat/runtime.py

def clean(self) -> None:
    """Remove content of cache_dir and reset global Runtime state."""
    shutil.rmtree(self.cache_dir, ignore_errors=True)

    # Reset global Runtime class state to allow fresh initialization
    Runtime.initialized = False
    Runtime.plugin_loader_enabled = False

    # Unload ansible modules that cache plugin/collection state
    # This ensures fresh plugin loader and collection finder state
    modules_to_remove = [
        mod
        for mod in sys.modules
        if mod.startswith(
            (
                "ansible.plugins.",  # Plugin loaders and cached plugins
                "ansible.utils.collection_loader",  # Collection finder system
                "ansible.collections.",  # Loaded collection modules
            ),
        )
    ]
    for mod in modules_to_remove:
        del sys.modules[mod]

enable_plugin_loader

enable_plugin_loader() -> None

Enable plugin loader initialization with current collections paths.

Source code in src/ansible_compat/runtime.py

def enable_plugin_loader(self) -> None:
    """Enable plugin loader initialization with current collections paths."""
    Runtime.plugin_loader_enabled = True

    # Initialize the plugin loader now with the full collections paths
    if not Runtime.initialized:
        col_paths = self.config.collections_paths
        if self.isolated:
            col_paths = [str(self.cache_dir / "collections"), *col_paths]
        # ruff: disable[import-private-name]
        # noinspection PyProtectedMember
        # pylint: disable=import-outside-toplevel,no-name-in-module
        from ansible.plugins.loader import init_plugin_loader
        from ansible.utils.collection_loader._collection_finder import (  # pylint: disable=import-outside-toplevel
            _AnsibleCollectionFinder,
        )
        # ruff: enable[import-private-name]

        with contextlib.suppress(Exception):
            _AnsibleCollectionFinder()._remove()  # pylint: disable=protected-access  # ruff:ignore[private-member-access]

        init_plugin_loader(col_paths)
        Runtime.initialized = True

has_playbook

has_playbook(
    playbook: str, *, basedir: Path | None = None
) -> bool

Return true if ansible can load a given playbook.

This is also used for checking if playbooks from within collections are present and if they pass syntax check.

Source code in src/ansible_compat/runtime.py

def has_playbook(self, playbook: str, *, basedir: Path | None = None) -> bool:
    """Return true if ansible can load a given playbook.

    This is also used for checking if playbooks from within collections
    are present and if they pass syntax check.
    """
    if (playbook, basedir) in self._has_playbook_cache:
        return self._has_playbook_cache[playbook, basedir]

    proc = self.run(["ansible-playbook", "--syntax-check", playbook], cwd=basedir)
    result = proc.returncode == 0
    if not result:
        if not basedir:
            basedir = Path()
        msg = f"has_playbook returned false for '{basedir / playbook}' due to syntax check returning {proc.returncode}"
        _logger.debug(msg)

    # cache the result
    self._has_playbook_cache[playbook, basedir] = result

    return result

initialize_logger

initialize_logger(level: int = 0) -> None

Set up the global logging level based on the verbosity number.

Source code in src/ansible_compat/runtime.py

def initialize_logger(self, level: int = 0) -> None:  # ruff:ignore[no-self-use]
    """Set up the global logging level based on the verbosity number."""
    verbosity_map = {
        -2: logging.CRITICAL,
        -1: logging.ERROR,
        0: logging.WARNING,
        1: logging.INFO,
        2: logging.DEBUG,
    }
    # Unknown logging level is treated as DEBUG
    logging_level = verbosity_map.get(level, logging.DEBUG)
    _logger.setLevel(logging_level)
    # Use module-level _logger instance to validate it
    _logger.debug("Logging initialized to level %s", logging_level)

install_collection

install_collection(
    collection: str | Path,
    *,
    destination: Path | None = None,
    force: bool = False
) -> None

Install an Ansible collection.

Can accept arguments like

'foo.bar:>=1.2.3' 'git+ansible-collections/ansible.posix.git,main'

Source code in src/ansible_compat/runtime.py

def install_collection(
    self,
    collection: str | Path,
    *,
    destination: Path | None = None,
    force: bool = False,
) -> None:
    """Install an Ansible collection.

    Can accept arguments like:
        'foo.bar:>=1.2.3'
        'git+https://github.com/ansible-collections/ansible.posix.git,main'
    """
    cmd = [
        "ansible-galaxy",
        "collection",
        "install",
        "-vvv",  # this is needed to make ansible display important info in case of failures
    ]
    if force:
        cmd.append("--force")

    if isinstance(collection, Path):
        collection = str(collection)
    # As ansible-galaxy install is not able to automatically determine
    # if the range requires a pre-release, we need to manually add the --pre
    # flag when needed.
    matches = version_re.search(collection)

    if (
        not is_url(collection)
        and matches
        and CollectionVersion(matches[1]).is_prerelease
    ):
        cmd.append("--pre")

    cpaths: list[str] = self.config.collections_paths
    if destination and str(destination) not in cpaths:
        # we cannot use '-p' because it breaks galaxy ability to ignore already installed collections, so
        # we hack ANSIBLE_COLLECTIONS_PATH instead and inject our own path there.
        # pylint: disable=no-member
        cpaths.insert(0, str(destination))
    cmd.append(f"{collection}")

    _logger.info("Running from %s : %s", Path.cwd(), " ".join(cmd))
    process = self.run(
        cmd,
        retry=True,
        env={**self.environ, "ANSIBLE_COLLECTIONS_PATH": ":".join(cpaths)},
    )
    if process.returncode != 0:
        msg = f"Command {' '.join(cmd)}, returned {process.returncode} code:\n{process.stdout}\n{process.stderr}"
        _logger.error(msg)
        raise InvalidPrerequisiteError(msg)

install_collection_from_disk

install_collection_from_disk(
    path: Path, destination: Path | None = None
) -> None

Build and install collection from a given disk path.

Source code in src/ansible_compat/runtime.py

def install_collection_from_disk(
    self,
    path: Path,
    destination: Path | None = None,
) -> None:
    """Build and install collection from a given disk path."""
    self.install_collection(path, destination=destination, force=True)

install_requirements

install_requirements(
    requirement: Path,
    *,
    retry: bool = False,
    offline: bool = False
) -> None

Install dependencies from a requirements.yml.

:param requirement: path to requirements.yml file :param retry: retry network operations on failures :param offline: bypass installation, may fail if requirements are not met.

Source code in src/ansible_compat/runtime.py

def install_requirements(
    self,
    requirement: Path,
    *,
    retry: bool = False,
    offline: bool = False,
) -> None:
    """Install dependencies from a requirements.yml.

    :param requirement: path to requirements.yml file
    :param retry: retry network operations on failures
    :param offline: bypass installation, may fail if requirements are not met.
    """
    if not Path(requirement).exists():
        return
    reqs_yaml = yaml_from_file(Path(requirement))
    _validate_requirements_yaml(reqs_yaml, requirement)

    if isinstance(reqs_yaml, list) or "roles" in reqs_yaml:
        self._install_role_requirements(requirement, retry=retry, offline=offline)

    if (
        isinstance(reqs_yaml, dict)
        and "collections" in reqs_yaml
        and reqs_yaml["collections"] is not None
    ):
        self._install_collection_requirements(
            requirement,
            reqs_yaml,
            retry=retry,
            offline=offline,
        )

    if self.require_module:
        Runtime.initialized = False
        self._ensure_module_available()

load_collections

load_collections() -> None

Load collection data.

Source code in src/ansible_compat/runtime.py

def load_collections(self) -> None:
    """Load collection data."""
    self.collections = OrderedDict()
    no_collections_msg = "None of the provided paths were usable"

    # do not use --path because it does not allow multiple values
    proc = self.run(
        [
            "ansible-galaxy",
            "collection",
            "list",
            "--format=json",
        ],
    )
    if proc.returncode == RC_ANSIBLE_OPTIONS_ERROR and (
        no_collections_msg in proc.stdout or no_collections_msg in proc.stderr
    ):  # pragma: no cover
        _logger.debug("Ansible reported no installed collections at all.")
        return
    if proc.returncode != 0:
        _logger.error(proc)
        msg = f"Unable to list collections: {proc}"
        raise RuntimeError(msg)
    try:
        data = json.loads(proc.stdout)
    except json.decoder.JSONDecodeError as exc:
        msg = f"Unable to parse galaxy output as JSON: {proc.stdout}"
        raise RuntimeError(msg) from exc
    if not isinstance(data, dict):
        msg = f"Unexpected collection data, {data}"
        raise TypeError(msg)
    self._parse_galaxy_collection_list(data)

prepare_environment

prepare_environment(
    required_collections: dict[str, str] | None = None,
    *,
    retry: bool = False,
    install_local: bool = False,
    offline: bool = False,
    role_name_check: int = 0
) -> None

Make dependencies available if needed.

Source code in src/ansible_compat/runtime.py

def prepare_environment(
    self,
    required_collections: dict[str, str] | None = None,
    *,
    retry: bool = False,
    install_local: bool = False,
    offline: bool = False,
    role_name_check: int = 0,
) -> None:
    """Make dependencies available if needed."""
    destination: Path = self.cache_dir / "collections"
    if required_collections is None:
        required_collections = {}

    self._prepare_ansible_paths()

    # first one is standard for collection layout repos and the last two
    # are part of Tower specification
    # https://docs.ansible.com/ansible-tower/latest/html/userguide/projects.html#ansible-galaxy-support
    # https://docs.ansible.com/ansible-tower/latest/html/userguide/projects.html#collections-support
    for req_file in REQUIREMENT_LOCATIONS:
        file_path = Path(req_file)
        if self.project_dir:
            file_path = self.project_dir / req_file
        self.install_requirements(file_path, retry=retry, offline=offline)

    if not install_local:
        return

    for item in search_galaxy_paths(self.project_dir):
        if item.exists():
            self._install_deps_from_galaxy(item, destination)

    for name, min_version in required_collections.items():
        self.install_collection(
            f"{name}:>={min_version}",
            destination=destination,
        )

    self._handle_local_install(destination, role_name_check)

require_collection

require_collection(
    name: str,
    version: str | None = None,
    *,
    install: bool = True
) -> tuple[CollectionVersion, Path]

Check if a minimal collection version is present or exits.

In the future this method may attempt to install a missing or outdated collection before failing.

Parameters:

• name (str) –

  collection name

• version (str | None, default: None ) –

  minimal version required

• install (bool, default: True ) –

  if True, attempt to install a missing collection

Returns:

• tuple[CollectionVersion, Path] –

  tuple of (found_version, collection_path)

Source code in src/ansible_compat/runtime.py

def require_collection(
    self,
    name: str,
    version: str | None = None,
    *,
    install: bool = True,
) -> tuple[CollectionVersion, Path]:
    """Check if a minimal collection version is present or exits.

    In the future this method may attempt to install a missing or outdated
    collection before failing.

    Args:
        name: collection name
        version: minimal version required
        install: if True, attempt to install a missing collection

    Returns:
        tuple of (found_version, collection_path)
    """
    try:
        ns, coll = name.split(".", 1)
    except ValueError as exc:
        msg = f"Invalid collection name supplied: {name}%s"
        raise InvalidPrerequisiteError(
            msg,
        ) from exc

    paths: list[str] = self.config.collections_paths
    if not paths or not isinstance(paths, list):
        msg = f"Unable to determine ansible collection paths. ({paths})"
        raise InvalidPrerequisiteError(
            msg,
        )

    for path in paths:
        collpath = Path(path) / "ansible_collections" / ns / coll
        result = self._check_collection_at_path(
            collpath,
            name=name,
            version=version,
            install=install,
        )
        if result is not None:
            return result
    if install:
        self.install_collection(f"{name}:>={version}" if version else name)
        return self.require_collection(
            name=name,
            version=version,
            install=False,
        )
    msg = f"Collection '{name}' not found in '{paths}'"
    _logger.fatal(msg)
    raise InvalidPrerequisiteError(msg)

run

run(
    args: str | list[str],
    *,
    retry: bool = False,
    tee: bool = False,
    env: dict[str, str] | None = None,
    cwd: Path | None = None,
    set_acp: bool = True
) -> CompletedProcess

Execute a command inside an Ansible environment.

:param retry: Retry network operations on failures. :param tee: Also pass captured stdout/stderr to system while running. :param set_acp: Set the ANSIBLE_COLLECTIONS_PATH

Source code in src/ansible_compat/runtime.py

def run(
    self,
    args: str | list[str],
    *,
    retry: bool = False,
    tee: bool = False,
    env: dict[str, str] | None = None,
    cwd: Path | None = None,
    set_acp: bool = True,
) -> CompletedProcess:
    """Execute a command inside an Ansible environment.

    :param retry: Retry network operations on failures.
    :param tee: Also pass captured stdout/stderr to system while running.
    :param set_acp: Set the ANSIBLE_COLLECTIONS_PATH
    """
    if tee:
        run_func: Callable[..., CompletedProcess] = subprocess_tee.run
    else:
        run_func = subprocess.run
    env = self.environ if env is None else env.copy()
    # Presence of ansible debug variable or config option will prevent us
    # from parsing its JSON output due to extra debug messages on stdout.
    env["ANSIBLE_DEBUG"] = "0"

    # https://github.com/ansible/ansible-lint/issues/3522
    env["ANSIBLE_VERBOSE_TO_STDERR"] = "True"

    if set_acp:
        env["ANSIBLE_COLLECTIONS_PATH"] = ":".join(
            list(dict.fromkeys(self.config.collections_paths)),
        )

    for _ in range(self.max_retries + 1 if retry else 1):
        result = run_func(
            args,
            universal_newlines=True,
            check=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=env,
            cwd=str(cwd) if cwd else None,
        )
        if result.returncode == 0:
            break
        _logger.debug("Environment: %s", env)
        if retry:
            _logger.warning(
                "Retrying execution failure %s of: %s",
                result.returncode,
                " ".join(args),
            )
    return result

version_in_range

version_in_range(
    lower: str | None = None, upper: str | None = None
) -> bool

Check if Ansible version is inside a required range.

The lower limit is inclusive and the upper one exclusive.

Source code in src/ansible_compat/runtime.py

def version_in_range(
    self,
    lower: str | None = None,
    upper: str | None = None,
) -> bool:
    """Check if Ansible version is inside a required range.

    The lower limit is inclusive and the upper one exclusive.
    """
    if lower and self.version < Version(lower):
        return False
    return not (upper and self.version >= Version(upper))

is_url

is_url(name: str) -> bool

Return True if a dependency name looks like an URL.

Source code in src/ansible_compat/runtime.py

def is_url(name: str) -> bool:
    """Return True if a dependency name looks like an URL."""
    return bool(re.match(r"^git[+@]", name))

search_galaxy_paths

search_galaxy_paths(search_dir: Path) -> list[Path]

Search for galaxy paths (only one level deep).

Returns:

• list[Path] –

  list[Path]: List of galaxy.yml found.

Source code in src/ansible_compat/runtime.py

def search_galaxy_paths(search_dir: Path) -> list[Path]:
    """Search for galaxy paths (only one level deep).

    Returns:
        list[Path]: List of galaxy.yml found.
    """
    galaxy_paths: list[Path] = []
    for item in [Path(), *search_dir.iterdir()]:
        # We ignore any folders that are not valid namespaces, just like
        # ansible galaxy does at this moment.
        file_path = item.resolve()
        if file_path.is_file() and file_path.name == GALAXY_YML:
            galaxy_paths.append(file_path)
            continue
        if file_path.is_dir() and namespace_re.match(file_path.name):
            file_path /= GALAXY_YML
            try:
                if file_path.exists():
                    galaxy_paths.append(file_path)
            except PermissionError:  # pragma: no cover
                # we silently ignore permissions errors, can happen with use of /tmp
                pass
    return galaxy_paths

ansible_compat.schema

Utils for JSON Schema validation.

JsonSchemaError dataclass

Data structure to hold a json schema validation error.

Source code in src/ansible_compat/schema.py

@dataclass(order=True)
class JsonSchemaError:
    # pylint: disable=too-many-instance-attributes
    """Data structure to hold a json schema validation error."""

    # order of attributes below is important for sorting
    schema_path: str
    data_path: str
    json_path: str
    message: str
    expected: bool | int | str
    relative_schema: str
    validator: str
    found: str

    def to_friendly(self) -> str:
        """Provide a friendly explanation of the error.

        Return: The error message
        """
        return f"In '{self.data_path}': {self.message}."

to_friendly

to_friendly() -> str

Provide a friendly explanation of the error.

Return: The error message

Source code in src/ansible_compat/schema.py

def to_friendly(self) -> str:
    """Provide a friendly explanation of the error.

    Return: The error message
    """
    return f"In '{self.data_path}': {self.message}."

json_path

json_path(absolute_path: Sequence[str | int]) -> str

Flatten a data path to a dot delimited string.

Parameters:

• absolute_path (Sequence[str | int]) –

  The path

Returns:

• str –

  The dot delimited string

Source code in src/ansible_compat/schema.py

def json_path(absolute_path: Sequence[str | int]) -> str:
    """Flatten a data path to a dot delimited string.

    Args:
        absolute_path: The path

    Returns:
        The dot delimited string
    """
    path = "$"
    for elem in absolute_path:
        if isinstance(elem, int):
            path += "[" + str(elem) + "]"
        else:
            path += "." + elem
    return path

to_path

to_path(schema_path: Sequence[str | int]) -> str

Flatten a path to a dot delimited string.

Parameters:

• schema_path (Sequence[str | int]) –

  The schema path

Returns:

• str –

  The dot delimited path

Source code in src/ansible_compat/schema.py

def to_path(schema_path: Sequence[str | int]) -> str:
    """Flatten a path to a dot delimited string.

    Args:
        schema_path: The schema path

    Returns:
        The dot delimited path
    """
    return ".".join(str(index) for index in schema_path)

validate

validate(schema: JSON, data: JSON) -> list[JsonSchemaError]

Validate some data against a JSON schema.

Parameters:

• schema (JSON) –

  the JSON schema to use for validation

• data (JSON) –

  The data to validate

Returns:

• list[JsonSchemaError] –

  Any errors encountered

Raises:

• SchemaError –

  if the schema is invalid

Source code in src/ansible_compat/schema.py

def validate(
    schema: JSON,
    data: JSON,
) -> list[JsonSchemaError]:
    """Validate some data against a JSON schema.

    Args:
        schema: the JSON schema to use for validation
        data: The data to validate

    Returns:
        Any errors encountered

    Raises:
        jsonschema.SchemaError: if the schema is invalid
    """
    errors: list[JsonSchemaError] = []

    if isinstance(schema, str):
        schema = json.loads(schema)
    try:
        if not isinstance(schema, Mapping):
            msg = "Invalid schema, must be a mapping"
            raise jsonschema.SchemaError(msg)  # ruff:ignore[raise-within-try]
        validator = validator_for(schema)
        validator.check_schema(schema)
    except jsonschema.SchemaError as exc:
        error = JsonSchemaError(
            message=str(exc),
            data_path="schema sanity check",
            json_path="",
            schema_path="",
            relative_schema="",
            expected="",
            validator="",
            found="",
        )
        errors.append(error)
        return errors

    for validation_error in validator(schema).iter_errors(data):
        if isinstance(
            validation_error, jsonschema.ValidationError
        ):  # pragma: no branch
            error = JsonSchemaError(
                message=validation_error.message,
                data_path=to_path(validation_error.absolute_path),
                json_path=json_path(validation_error.absolute_path),
                schema_path=to_path(validation_error.schema_path),
                relative_schema=str(validation_error.schema),
                expected=str(validation_error.validator_value),
                validator=str(validation_error.validator),
                found=str(validation_error.instance),
            )
            errors.append(error)
    return sorted(errors)