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.

Source code in ansible_compat/config.py

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

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

    _aliases = {
        "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: 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: list[str] = []
    callbacks_enabled: 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: 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: list[str] = [
        "~/.ansible/plugins/cache",
        "/usr/share/ansible/plugins/cache",
    ]
    default_callback_plugin_path: list[str] = [
        "~/.ansible/plugins/callback",
        "/usr/share/ansible/plugins/callback",
    ]
    default_cliconf_plugin_path: list[str] = [
        "~/.ansible/plugins/cliconf",
        "/usr/share/ansible/plugins/cliconf",
    ]
    default_connection_plugin_path: 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: list[str] = [
        "~/.ansible/plugins/filter",
        "/usr/share/ansible/plugins/filter",
    ]
    default_force_handlers: bool = False
    default_forks: int = 5
    default_gathering: Literal["smart", "explicit", "implicit"] = "smart"
    default_gather_subset: list[str] = ["all"]
    default_gather_timeout: int = 10
    default_handler_includes_static: bool = False
    default_hash_behaviour: str = "replace"
    default_host_list: list[str] = ["/etc/ansible/hosts"]
    default_httpapi_plugin_path: list[str] = [
        "~/.ansible/plugins/httpapi",
        "/usr/share/ansible/plugins/httpapi",
    ]
    default_internal_poll_interval: float = 0.001
    default_inventory_plugin_path: list[str] = [
        "~/.ansible/plugins/inventory",
        "/usr/share/ansible/plugins/inventory",
    ]
    default_jinja2_extensions: 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: list[str] = []
    default_log_path: str | None = None
    default_lookup_lugin_path: 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: list[str] = [
        "~/.ansible/plugins/modules",
        "/usr/share/ansible/plugins/modules",
    ]
    default_module_utils_path: list[str] = [
        "~/.ansible/plugins/module_utils",
        "/usr/share/ansible/plugins/module_utils",
    ]
    default_netconf_plugin_path: 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: list[str] = [
        "~/.ansible/collections",
        "/usr/share/ansible/collections",
    ]
    default_roles_path: list[str] = [
        "~/.ansible/roles",
        "/usr/share/ansible/roles",
        "/etc/ansible/roles",
    ]
    default_selinux_special_fs: list[str] = [
        "fuse",
        "nfs",
        "vboxsf",
        "ramfs",
        "9p",
        "vfat",
    ]
    default_stdout_callback: str = "default"
    default_strategy: str = "linear"
    default_strategy_plugin_path: 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: list[str] = [
        "~/.ansible/plugins/terminal",
        "/usr/share/ansible/plugins/terminal",
    ]
    default_test_plugin_path: 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: list[str] = [
        "~/.ansible/plugins/vars",
        "/usr/share/ansible/plugins/vars",
    ]
    default_vault_encrypt_identity: str | None = None
    default_vault_identity: str = "default"
    default_vault_identity_list: list[str] = []
    default_vault_id_match: bool = False
    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: 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: 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: 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"
    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: list[str] = [
        "host_list",
        "script",
        "auto",
        "yaml",
        "ini",
        "toml",
    ]
    inventory_export: bool = False
    inventory_ignore_exts: str
    inventory_ignore_patterns: 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: 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: list[str] = [
        "string",
        "to_json",
        "to_nice_json",
        "to_yaml",
        "to_nice_yaml",
        "ppretty",
        "json",
    ]
    system_warnings: bool = True
    tags_run: list[str] = []
    tags_skip: 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: list[str] = ["host_group_vars"]
    variable_precedence: 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: list[str] = [".yml", ".yaml", ".json"]

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

        if data:
            self.data = copy.deepcopy(data)
            return

        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"],  # noqa: S603
                universal_newlines=True,
                env=env,
            )

        for match in re.finditer(
            r"^(?P<key>[A-Za-z0-9_]+).* = (?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

    def __getattribute__(self, attr_name: str) -> object:
        """Allow access of config options as attributes."""
        _dict = super().__dict__  # pylint: disable=no-member
        if attr_name in _dict:
            return _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 deeepcopy on Config."""
        return AnsibleConfig(data=self.data)

__copy__

__copy__() -> AnsibleConfig

Allow users to run copy on Config.

Source code in 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 deeepcopy on Config.

Source code in ansible_compat/config.py

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

__getattribute__

__getattribute__(attr_name: str) -> object

Allow access of config options as attributes.

Source code in ansible_compat/config.py

def __getattribute__(self, attr_name: str) -> object:
    """Allow access of config options as attributes."""
    _dict = super().__dict__  # pylint: disable=no-member
    if attr_name in _dict:
        return _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 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,
) -> None

Load config dictionary.

Source code in ansible_compat/config.py

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

    if data:
        self.data = copy.deepcopy(data)
        return

    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"],  # noqa: S603
            universal_newlines=True,
            env=env,
        )

    for match in re.finditer(
        r"^(?P<key>[A-Za-z0-9_]+).* = (?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

ansible_collections_path

ansible_collections_path() -> str

Return collection path variable for current version of Ansible.

Source code in ansible_compat/config.py

def ansible_collections_path() -> str:
    """Return collection path variable for current version of Ansible."""
    for env_var in [
        "ANSIBLE_COLLECTIONS_PATH",
        "ANSIBLE_COLLECTIONS_PATHS",
    ]:
        if env_var in os.environ:
            return env_var
    return "ANSIBLE_COLLECTIONS_PATH"

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 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"],  # noqa: S603
        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 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 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

Construct an exception given a completed process.

Source code in 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 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

Construct generic library exception.

Source code in 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 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 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 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 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 ansible_compat/loaders.py

def yaml_from_file(path: Path) -> Any:  # noqa: ANN401
    """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.

get_cache_dir

get_cache_dir(project_dir: Path) -> Path

Compute cache directory to be used based on project path.

Source code in ansible_compat/prerun.py

def get_cache_dir(project_dir: Path) -> Path:
    """Compute cache directory to be used based on project path."""
    # we only use the basename instead of the full path in order to ensure that
    # we would use the same key regardless the location of the user home
    # directory or where the project is clones (as long the project folder uses
    # the same name).
    basename = project_dir.resolve().name.encode(encoding="utf-8")
    # 6 chars of entropy should be enough
    cache_key = hashlib.sha256(basename).hexdigest()[:6]
    cache_dir = (
        Path(os.getenv("XDG_CACHE_HOME", "~/.cache")).expanduser()
        / "ansible-compat"
        / cache_key
    )
    return cache_dir

ansible_compat.runtime

Ansible runtime environment manager.

AnsibleWarning

Bases: Warning

Warnings related to Ansible runtime.

Source code in ansible_compat/runtime.py

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

Collection dataclass

Container for Ansible collection information.

Source code in 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 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

Initialize collection version.

Source code in 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 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):  # noqa: ANN204
        """Get attribute."""
        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 ansible_version() < Version("2.14") and attr in {"filter", "test"}:
                    msg = "Ansible version below 2.14 does not support retrieving filter and test plugins."
                    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.

Source code in ansible_compat/runtime.py

@no_type_check
def __getattribute__(self, attr: str):  # noqa: ANN204
    """Get attribute."""
    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 ansible_version() < Version("2.14") and attr in {"filter", "test"}:
                msg = "Ansible version below 2.14 does not support retrieving filter and test plugins."
                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.

Source code in ansible_compat/runtime.py

class Runtime:
    """Ansible Runtime manager."""

    _version: Version | None = None
    collections: OrderedDict[str, Collection] = OrderedDict()
    cache_dir: Path | None = None
    # Used to track if we have already initialized the Ansible runtime as attempts
    # to do it multiple tilmes will cause runtime warnings from within ansible-core
    initialized: bool = False
    plugins: Plugins

    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()
        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"

        if isolated:
            self.cache_dir = get_cache_dir(self.project_dir)
        self.config = AnsibleConfig()

        # Add the sys.path to the collection paths if not isolated
        self._add_sys_path_to_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._ensure_module_available()

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

        # pylint: disable=unused-argument
        def warning(
            self: Display,  # noqa: ARG001
            msg: str,
            *,
            formatted: bool = False,  # noqa: ARG001
        ) -> None:
            """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:
        """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 _add_sys_path_to_collection_paths(self) -> None:
        """Add the sys.path to the collection paths."""
        if self.config.collections_scan_sys_path:
            for path in sys.path:
                if (
                    path not in self.config.collections_paths
                    and (Path(path) / "ansible_collections").is_dir()
                ):
                    self.config.collections_paths.append(  # pylint: disable=E1101
                        path,
                    )

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

        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
        ):
            _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)
        data = json.loads(proc.stdout)
        if not isinstance(data, dict):
            msg = f"Unexpected collection data, {data}"
            raise TypeError(msg)
        for path in data:
            for collection, collection_info in data[path].items():
                if not isinstance(collection, str):
                    msg = f"Unexpected collection data, {collection}"
                    raise TypeError(msg)
                if not isinstance(collection_info, dict):
                    msg = f"Unexpected collection data, {collection_info}"
                    raise TypeError(msg)

                self.collections[collection] = Collection(
                    name=collection,
                    version=collection_info["version"],
                    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)

        # For ansible 2.15+ we need to initialize the plugin loader
        # https://github.com/ansible/ansible-lint/issues/2945
        if not Runtime.initialized:
            col_path = [f"{self.cache_dir}/collections"]
            if self.version >= Version("2.15.0.dev0"):
                # pylint: disable=import-outside-toplevel,no-name-in-module
                from ansible.plugins.loader import init_plugin_loader

                init_plugin_loader(col_path)
            else:
                # noinspection PyProtectedMember
                from ansible.utils.collection_loader._collection_finder import (  # pylint: disable=import-outside-toplevel
                    _AnsibleCollectionFinder,
                )

                # noinspection PyProtectedMember
                # pylint: disable=protected-access
                col_path += self.config.collections_paths
                col_path += os.path.dirname(  # noqa: PTH120
                    os.environ.get(ansible_collections_path(), "."),
                ).split(":")
                _AnsibleCollectionFinder(  # noqa: SLF001
                    paths=col_path,
                )._install()  # pylint: disable=protected-access
            Runtime.initialized = True

    def clean(self) -> None:
        """Remove content of cache_dir."""
        if self.cache_dir:
            shutil.rmtree(self.cache_dir, ignore_errors=True)

    def run(  # ruff: disable=PLR0913
        self,
        args: str | list[str],
        *,
        retry: bool = False,
        tee: bool = False,
        env: dict[str, str] | None = None,
        cwd: Path | None = None,
    ) -> 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.
        """
        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"

        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:
        """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 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
        if upper and self.version >= Version(upper):
            return False
        return True

    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 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)

    # pylint: disable=too-many-branches
    def install_requirements(  # noqa: C901
        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))
        if not isinstance(reqs_yaml, (dict, list)):
            msg = f"{requirement} file is not a valid Ansible requirements file."
            raise InvalidPrerequisiteError(msg)

        if isinstance(reqs_yaml, dict):
            for key in reqs_yaml:
                if key not in ("roles", "collections"):
                    msg = f"{requirement} file is not a valid Ansible requirements file. Only 'roles' and 'collections' keys are allowed at root level. Recognized valid locations are: {', '.join(REQUIREMENT_LOCATIONS)}"
                    raise InvalidPrerequisiteError(msg)

        if isinstance(reqs_yaml, list) or "roles" in reqs_yaml:
            cmd = [
                "ansible-galaxy",
                "role",
                "install",
                "-r",
                f"{requirement}",
            ]
            if self.verbosity > 0:
                cmd.extend(["-" + ("v" * self.verbosity)])
            if self.cache_dir:
                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)

        # Run galaxy collection install works on v2 requirements.yml
        if "collections" in reqs_yaml and reqs_yaml["collections"] is not None:
            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)])
                cpaths = self.config.collections_paths
                if self.cache_dir:
                    # 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.
                    dest_path = f"{self.cache_dir}/collections"
                    if dest_path not in cpaths:
                        # pylint: disable=no-member
                        cpaths.insert(0, dest_path)
                _logger.info("Running %s", " ".join(cmd))
                result = self.run(
                    cmd,
                    retry=retry,
                    env={**os.environ, "ANSIBLE_COLLECTIONS_PATH": ":".join(cpaths)},
                )
                _logger.debug(result.stdout)
                if result.returncode != 0:
                    _logger.error(result.stderr)
                    raise AnsibleCommandError(result)

    # pylint: disable=too-many-locals
    def prepare_environment(  # noqa: C901
        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 | None = None
        if required_collections is None:
            required_collections = {}

        # 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)

        self._prepare_ansible_paths()

        if not install_local:
            return

        for gpath in search_galaxy_paths(self.project_dir):
            # processing all found galaxy.yml files
            galaxy_path = Path(gpath)
            if galaxy_path.exists():
                data = yaml_from_file(galaxy_path)
                if isinstance(data, dict) and "dependencies" in data:
                    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,
                        )

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

        if (self.project_dir / "galaxy.yml").exists():
            if destination:
                # while function can return None, that would not break the logic
                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()

            # molecule scenario within a collection
            self.install_collection_from_disk(
                galaxy_path.parent,
                destination=destination,
            )
        elif (
            Path().resolve().parent.name == "roles"
            and Path("../../galaxy.yml").exists()
        ):
            # molecule scenario located within roles/<role-name>/molecule inside
            # a collection
            self.install_collection_from_disk(
                Path("../.."),
                destination=destination,
            )
        else:
            # no collection, try to recognize and install a standalone role
            self._install_galaxy_role(
                self.project_dir,
                role_name_check=role_name_check,
                ignore_errors=True,
            )
        # reload collections
        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.

        :param name: collection name
        :param version: minimal version required
        :param 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,
            )

        if self.cache_dir:
            # if we have a cache dir, we want to be use that would be preferred
            # destination when installing a missing collection
            # https://github.com/PyCQA/pylint/issues/4667
            paths.insert(0, f"{self.cache_dir}/collections")  # pylint: disable=E1101

        for path in paths:
            collpath = Path(path) / "ansible_collections" / ns / coll
            if collpath.exists():
                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()
                break
        else:
            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 _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, "plugins/modules", True),
            (roles_path, "roles", 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`.
        """
        if self.cache_dir:
            path = Path(f"{self.cache_dir}/roles")
        else:
            path = Path(self.config.default_roles_path[0]).expanduser()
        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.

        :param: 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

        :param: ignore_errors: if True, bypass installing invalid roles.

        Our implementation aims to match ansible-galaxy's behaviour 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 = _get_role_fqrn(galaxy_info, project_dir)

        if role_name_check in [0, 1]:
            if not re.match(r"[a-z0-9][a-z0-9_-]+\.[a-z][a-z0-9_]+$", fqrn):
                msg = MSG_INVALID_FQRL.format(fqrn)
                if role_name_check == 1:
                    _logger.warning(msg)
                else:
                    _logger.error(msg)
                    raise InvalidPrerequisiteError(msg)
        elif "role_name" in galaxy_info:
            # when 'role-name' is in skip_list, we stick to plain role names
            role_namespace = _get_galaxy_role_ns(galaxy_info)
            role_name = _get_galaxy_role_name(galaxy_info)
            fqrn = f"{role_namespace}{role_name}"
        else:
            fqrn = Path(project_dir).absolute().name
        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:
            value = [*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

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.

__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

Initialize Ansible runtime environment.

Parameters:

• project_dir (Path | None, default: None ) –

  The directory containing the Ansible project. If not mentioned it will be guessed from the current working directory.

• isolated (bool, default: False ) –

  Assure that installation of collections or roles does not affect Ansible installation, an unique cache directory being used instead.

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

  Minimal version of Ansible required. If not found, a :class:RuntimeError exception is raised.

• require_module (bool, default: False ) –

  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.

• max_retries (int, default: 0 ) –

  Number of times it should retry network operations. Default is 0, no retries.

• environ (dict[str, str] | None, default: None ) –

  Environment dictionary to use, if undefined os.environ will be copied and used.

• verbosity (int, default: 0 ) –

  Verbosity level to use.

Source code in 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()
    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"

    if isolated:
        self.cache_dir = get_cache_dir(self.project_dir)
    self.config = AnsibleConfig()

    # Add the sys.path to the collection paths if not isolated
    self._add_sys_path_to_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._ensure_module_available()

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

    # pylint: disable=unused-argument
    def warning(
        self: Display,  # noqa: ARG001
        msg: str,
        *,
        formatted: bool = False,  # noqa: ARG001
    ) -> None:
        """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.

Source code in ansible_compat/runtime.py

def clean(self) -> None:
    """Remove content of cache_dir."""
    if self.cache_dir:
        shutil.rmtree(self.cache_dir, ignore_errors=True)

initialize_logger

initialize_logger(level: int = 0) -> None

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

Source code in ansible_compat/runtime.py

def initialize_logger(self, level: int = 0) -> None:
    """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 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 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 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.

Parameters:

• requirement (Path) –

  path to requirements.yml file

• retry (bool, default: False ) –

  retry network operations on failures

• offline (bool, default: False ) –

  bypass installation, may fail if requirements are not met.

Source code in ansible_compat/runtime.py

def install_requirements(  # noqa: C901
    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))
    if not isinstance(reqs_yaml, (dict, list)):
        msg = f"{requirement} file is not a valid Ansible requirements file."
        raise InvalidPrerequisiteError(msg)

    if isinstance(reqs_yaml, dict):
        for key in reqs_yaml:
            if key not in ("roles", "collections"):
                msg = f"{requirement} file is not a valid Ansible requirements file. Only 'roles' and 'collections' keys are allowed at root level. Recognized valid locations are: {', '.join(REQUIREMENT_LOCATIONS)}"
                raise InvalidPrerequisiteError(msg)

    if isinstance(reqs_yaml, list) or "roles" in reqs_yaml:
        cmd = [
            "ansible-galaxy",
            "role",
            "install",
            "-r",
            f"{requirement}",
        ]
        if self.verbosity > 0:
            cmd.extend(["-" + ("v" * self.verbosity)])
        if self.cache_dir:
            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)

    # Run galaxy collection install works on v2 requirements.yml
    if "collections" in reqs_yaml and reqs_yaml["collections"] is not None:
        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)])
            cpaths = self.config.collections_paths
            if self.cache_dir:
                # 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.
                dest_path = f"{self.cache_dir}/collections"
                if dest_path not in cpaths:
                    # pylint: disable=no-member
                    cpaths.insert(0, dest_path)
            _logger.info("Running %s", " ".join(cmd))
            result = self.run(
                cmd,
                retry=retry,
                env={**os.environ, "ANSIBLE_COLLECTIONS_PATH": ":".join(cpaths)},
            )
            _logger.debug(result.stdout)
            if result.returncode != 0:
                _logger.error(result.stderr)
                raise AnsibleCommandError(result)

load_collections

load_collections() -> None

Load collection data.

Source code in 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"

    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
    ):
        _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)
    data = json.loads(proc.stdout)
    if not isinstance(data, dict):
        msg = f"Unexpected collection data, {data}"
        raise TypeError(msg)
    for path in data:
        for collection, collection_info in data[path].items():
            if not isinstance(collection, str):
                msg = f"Unexpected collection data, {collection}"
                raise TypeError(msg)
            if not isinstance(collection_info, dict):
                msg = f"Unexpected collection data, {collection_info}"
                raise TypeError(msg)

            self.collections[collection] = Collection(
                name=collection,
                version=collection_info["version"],
                path=path,
            )

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 ansible_compat/runtime.py

def prepare_environment(  # noqa: C901
    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 | None = None
    if required_collections is None:
        required_collections = {}

    # 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)

    self._prepare_ansible_paths()

    if not install_local:
        return

    for gpath in search_galaxy_paths(self.project_dir):
        # processing all found galaxy.yml files
        galaxy_path = Path(gpath)
        if galaxy_path.exists():
            data = yaml_from_file(galaxy_path)
            if isinstance(data, dict) and "dependencies" in data:
                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,
                    )

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

    if (self.project_dir / "galaxy.yml").exists():
        if destination:
            # while function can return None, that would not break the logic
            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()

        # molecule scenario within a collection
        self.install_collection_from_disk(
            galaxy_path.parent,
            destination=destination,
        )
    elif (
        Path().resolve().parent.name == "roles"
        and Path("../../galaxy.yml").exists()
    ):
        # molecule scenario located within roles/<role-name>/molecule inside
        # a collection
        self.install_collection_from_disk(
            Path("../.."),
            destination=destination,
        )
    else:
        # no collection, try to recognize and install a standalone role
        self._install_galaxy_role(
            self.project_dir,
            role_name_check=role_name_check,
            ignore_errors=True,
        )
    # reload collections
    self.load_collections()

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 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.

    :param name: collection name
    :param version: minimal version required
    :param 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,
        )

    if self.cache_dir:
        # if we have a cache dir, we want to be use that would be preferred
        # destination when installing a missing collection
        # https://github.com/PyCQA/pylint/issues/4667
        paths.insert(0, f"{self.cache_dir}/collections")  # pylint: disable=E1101

    for path in paths:
        collpath = Path(path) / "ansible_collections" / ns / coll
        if collpath.exists():
            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()
            break
    else:
        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
) -> CompletedProcess

Execute a command inside an Ansible environment.

Parameters:

• retry (bool, default: False ) –

  Retry network operations on failures.

• tee (bool, default: False ) –

  Also pass captured stdout/stderr to system while running.

Source code in ansible_compat/runtime.py

def run(  # ruff: disable=PLR0913
    self,
    args: str | list[str],
    *,
    retry: bool = False,
    tee: bool = False,
    env: dict[str, str] | None = None,
    cwd: Path | None = None,
) -> 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.
    """
    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"

    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 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
    if upper and self.version >= Version(upper):
        return False
    return True

is_url

is_url(name: str) -> bool

Return True if a dependency name looks like an URL.

Source code in ansible_compat/runtime.py

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

search_galaxy_paths

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

Search for galaxy paths (only one level deep).

Source code in ansible_compat/runtime.py

def search_galaxy_paths(search_dir: Path) -> list[str]:
    """Search for galaxy paths (only one level deep)."""
    galaxy_paths: list[str] = []
    for file in [".", *os.listdir(search_dir)]:
        # We ignore any folders that are not valid namespaces, just like
        # ansible galaxy does at this moment.
        if file != "." and not namespace_re.match(file):
            continue
        file_path = search_dir / file / "galaxy.yml"
        if file_path.is_file():
            galaxy_paths.append(str(file_path))
    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 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.

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

to_friendly

to_friendly() -> str

Provide a friendly explanation of the error.

Returns:

• str –

  The error message

Source code in ansible_compat/schema.py

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

    :returns: 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 ansible_compat/schema.py

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

    :param 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 ansible_compat/schema.py

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

    :param 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

Source code in ansible_compat/schema.py

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

    :param schema: the JSON schema to use for validation
    :param data: The data to validate
    :returns: Any errors encountered
    """
    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)  # noqa: TRY301
        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):
            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=validation_error.schema,
                expected=validation_error.validator_value,
                validator=str(validation_error.validator),
                found=str(validation_error.instance),
            )
            errors.append(error)
    return sorted(errors)