ansible-later/ansiblelater/utils/__init__.py

137 lines
3.2 KiB
Python
Raw Normal View History

2019-04-10 12:50:48 +00:00
"""Global utils collection."""
2019-03-28 00:20:43 +00:00
import contextlib
import re
import sys
from contextlib import suppress
2024-01-29 07:34:08 +00:00
from functools import lru_cache
2018-12-19 10:19:07 +00:00
2019-03-28 00:20:43 +00:00
import yaml
2024-01-29 07:34:08 +00:00
from ansible.plugins.loader import module_loader
2019-04-02 14:34:03 +00:00
from ansiblelater import logger
2018-12-19 10:19:07 +00:00
try:
2020-04-05 12:33:43 +00:00
import ConfigParser as configparser # noqa
2018-12-19 10:19:07 +00:00
except ImportError:
2020-04-05 12:33:43 +00:00
import configparser # noqa
2019-04-02 14:34:03 +00:00
LOG = logger.get_logger(__name__)
2019-01-25 11:32:28 +00:00
def count_spaces(c_string):
leading_spaces = 0
trailing_spaces = 0
for _i, e in enumerate(c_string):
2019-01-25 11:32:28 +00:00
if not e.isspace():
break
leading_spaces += 1
for _i, e in reversed(list(enumerate(c_string))):
2019-01-25 11:32:28 +00:00
if not e.isspace():
break
trailing_spaces += 1
return (leading_spaces, trailing_spaces)
2019-01-25 11:32:28 +00:00
2018-12-19 10:19:07 +00:00
def lines_ranges(lines_spec):
if not lines_spec:
return None
result = []
for interval in lines_spec.split(","):
(start, end) = interval.split("-")
result.append(range(int(start), int(end) + 1))
return result
def is_line_in_ranges(line, ranges):
return not ranges or any(line in r for r in ranges)
2018-12-19 10:19:07 +00:00
2019-03-28 00:20:43 +00:00
def safe_load(string):
"""
Parse the provided string returns a dict.
2019-03-28 00:20:43 +00:00
:param string: A string to be parsed.
:returns: dict
2019-03-28 00:20:43 +00:00
"""
with suppress(yaml.scanner.ScannerError):
2019-03-28 00:20:43 +00:00
return yaml.safe_load(string) or {}
@contextlib.contextmanager
def open_file(filename, mode="r"):
2019-03-28 00:20:43 +00:00
"""
Open the provide file safely and returns a file type.
2019-03-28 00:20:43 +00:00
:param filename: A string containing an absolute path to the file to open.
:param mode: A string describing the way in which the file will be used.
:returns: file type
2019-03-28 00:20:43 +00:00
"""
with open(filename, mode) as stream:
yield stream
def add_dict_branch(tree, vector, value):
key = vector[0]
tree[key] = (
value
if len(vector) == 1
else add_dict_branch(tree[key] if key in tree else {}, vector[1:], value)
)
return tree
2019-04-02 14:34:03 +00:00
def has_jinja(value):
"""Return true if a string seems to contain jinja templating."""
re_has_jinja = re.compile(r"{[{%#].*[%#}]}", re.DOTALL)
return bool(isinstance(value, str) and re_has_jinja.search(value))
def has_glob(value):
"""Return true if a string looks like having a glob pattern."""
re_has_glob = re.compile("[][*?]")
return bool(isinstance(value, str) and re_has_glob.search(value))
2019-04-02 14:34:03 +00:00
def sysexit(code=1):
sys.exit(code)
def sysexit_with_message(msg, code=1):
LOG.critical(msg)
sysexit(code)
class Singleton(type):
"""Meta singleton class."""
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
2024-01-29 07:34:08 +00:00
@lru_cache
def load_plugin(name):
"""Return loaded ansible plugin/module."""
loaded_module = module_loader.find_plugin_with_context(
name,
ignore_deprecated=True,
check_aliases=True,
)
if not loaded_module.resolved and name.startswith("ansible.builtin."):
# fallback to core behavior of using legacy
loaded_module = module_loader.find_plugin_with_context(
name.replace("ansible.builtin.", "ansible.legacy."),
ignore_deprecated=True,
check_aliases=True,
)
return loaded_module