utils.py

Common utilities.

__all__ = ['handle_teardown_signals', 'datetime_to_seconds', 'is_true', 'is_false', 'retry', 'extract_http_scheme_host_port', 'remove_http_proxy_env_vars'] module-attribute

datetime_to_seconds(dt)

Convert UTC datetime to seconds since epoch.

Parameters:

Name	Type	Description	Default
dt	datetime.datetime	Date time.	required

Returns:

Type	Description
float	Seconds since epoch.

Source code in solnlib/utils.py

def datetime_to_seconds(dt: datetime.datetime) -> float:
    """Convert UTC datetime to seconds since epoch.

    Arguments:
        dt: Date time.

    Returns:
        Seconds since epoch.
    """

    epoch_time = datetime.datetime.utcfromtimestamp(0)
    return (dt - epoch_time).total_seconds()

extract_http_scheme_host_port(http_url)

Extract scheme, host and port from a HTTP URL.

Parameters:

Name	Type	Description	Default
http_url	str	HTTP URL to extract.	required

Returns:

Type	Description
Tuple	A tuple of scheme, host and port

Raises:

Type	Description
ValueError	If http_url is not in http(s)://hostname:port format.

Source code in solnlib/utils.py

def extract_http_scheme_host_port(http_url: str) -> Tuple:
    """Extract scheme, host and port from a HTTP URL.

    Arguments:
        http_url: HTTP URL to extract.

    Returns:
        A tuple of scheme, host and port

    Raises:
        ValueError: If `http_url` is not in http(s)://hostname:port format.
    """

    http_info = urlparse.urlparse(http_url)
    if not http_info.scheme or not http_info.hostname or not http_info.port:
        raise ValueError(http_url + " is not in http(s)://hostname:port format")
    return http_info.scheme, http_info.hostname, http_info.port

handle_teardown_signals(callback)

Register handler for SIGTERM/SIGINT/SIGBREAK signal.

Catch SIGTERM/SIGINT/SIGBREAK signals, and invoke callback Note: this should be called in main thread since Python only catches signals in main thread.

Parameters:

Name	Type	Description	Default
callback	Callable	Callback for tear down signals.	required

Source code in solnlib/utils.py

def handle_teardown_signals(callback: Callable):
    """Register handler for SIGTERM/SIGINT/SIGBREAK signal.

    Catch SIGTERM/SIGINT/SIGBREAK signals, and invoke callback
    Note: this should be called in main thread since Python only catches
    signals in main thread.

    Arguments:
        callback: Callback for tear down signals.
    """

    signal.signal(signal.SIGTERM, callback)
    signal.signal(signal.SIGINT, callback)

    if os.name == "nt":
        signal.signal(signal.SIGBREAK, callback)

is_false(val)

Decide if val is false.

Parameters:

Name	Type	Description	Default
val	Union[str, int]	Value to check.	required

Returns:

Type	Description
bool	True or False.

Source code in solnlib/utils.py

def is_false(val: Union[str, int]) -> bool:
    """Decide if `val` is false.

    Arguments:
        val: Value to check.

    Returns:
        True or False.
    """

    value = str(val).strip().upper()
    if value in ("0", "FALSE", "F", "N", "NO", "NONE", ""):
        return True
    return False

is_true(val)

Decide if val is true.

Parameters:

Name	Type	Description	Default
val	Union[str, int]	Value to check.	required

Returns:

Type	Description
bool	True or False.

Source code in solnlib/utils.py

def is_true(val: Union[str, int]) -> bool:
    """Decide if `val` is true.

    Arguments:
        val: Value to check.

    Returns:
        True or False.
    """

    value = str(val).strip().upper()
    if value in ("1", "TRUE", "T", "Y", "YES"):
        return True
    return False

remove_http_proxy_env_vars()

Removes HTTP(s) proxies from environment variables.

Removes the following environment variables

• http_proxy
• https_proxy
• HTTP_PROXY
• HTTPS_PROXY

This function can be used in Splunk modular inputs code before starting the ingestion to ensure that no proxy is going to be used when doing requests. In case of proxy is needed, it can be defined in the modular inputs code.

Source code in solnlib/utils.py

def remove_http_proxy_env_vars() -> None:
    """Removes HTTP(s) proxies from environment variables.

    Removes the following environment variables:
        * http_proxy
        * https_proxy
        * HTTP_PROXY
        * HTTPS_PROXY

    This function can be used in Splunk modular inputs code before starting the
    ingestion to ensure that no proxy is going to be used when doing requests.
    In case of proxy is needed, it can be defined in the modular inputs code.
    """
    env_vars_to_remove = (
        "http_proxy",
        "https_proxy",
        "HTTP_PROXY",
        "HTTPS_PROXY",
    )
    for env_var in env_vars_to_remove:
        if env_var in os.environ:
            del os.environ[env_var]

retry(retries=3, reraise=True, default_return=None, exceptions=None)

A decorator to run function with max retries times if there is exception.

Parameters:

Name	Type	Description	Default
retries	int	(optional) Max retries times, default is 3.	3
reraise	bool	Whether exception should be reraised, default is True.	True
default_return	Any	(optional) Default return value for function run after max retries and reraise is False.	None
exceptions	List	(optional) List of exceptions that should retry.	None

Source code in solnlib/utils.py

def retry(
    retries: int = 3,
    reraise: bool = True,
    default_return: Any = None,
    exceptions: List = None,
):
    """A decorator to run function with max `retries` times if there is
    exception.

    Arguments:
        retries: (optional) Max retries times, default is 3.
        reraise: Whether exception should be reraised, default is True.
        default_return: (optional) Default return value for function
            run after max retries and reraise is False.
        exceptions: (optional) List of exceptions that should retry.
    """

    max_tries = max(retries, 0) + 1

    def do_retry(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_ex = None
            for i in range(max_tries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    logging.warning(
                        "Run function: %s failed: %s.",
                        func.__name__,
                        traceback.format_exc(),
                    )
                    if not exceptions or any(
                        isinstance(e, exception) for exception in exceptions
                    ):
                        last_ex = e
                        if i < max_tries - 1:
                            time.sleep(2**i)
                    else:
                        raise

            if reraise:
                raise last_ex
            else:
                return default_return

        return wrapper

    return do_retry