iripau.functools module

Function utilities with similar philosophy as the functools module.

iripau.functools.wait_for(condition, *args, _timeout=None, _outcome=True, _poll_time=10, _stop_condition=None, **kwargs)

Wait for condition(*args, **kwargs) to be _outcome.

Parameters:

• condition (Callable) – Call this function many times with *args and **kwargs until its return value is _outcome.

• *args – Pass these arguments to condition.

• **kwargs – Pass these keyword arguments to condition.

• _timeout (float) – Max time in seconds to continue on calling condition.

• _outcome (bool) – The desired outcome to gracefully stop waiting.

• _poll_time (float) – Seconds in between condition calls.

• _stop_condition (Callable[[], bool]) – Stop waiting if the result of calling this function is True.

Raises:

• TimeoutError – If timeout reached.

• InterruptedError – If _stop_condition() returns True.

Example

Wait for a network interface to have an IP address:

import netifaces


def iface_is_connected(iface):
    """ ``iface`` has an IP v4 address """
    return netifaces.AF_INET in netifaces.ifaddresses(iface)


wait_for(iface_is_connected, "eth0", _timeout=10)

Example

Wait for a process to finish, but stop waiting if there is an error log:

import psutil


def log_file_contains_error():
    """ The log informs the process has stalled """
    with open("/tmp/events.log") as f:
        return "Error: Process stalled" in f.read()


wait_for(
    psutil.pid_exists, 12345,
    _outcome=False,
    _timeout=3600,
    _poll_time=600,
    _stop_condition=log_file_contains_error
)

Tip

Create condition and _stop_condition functions that read nicely when used in if statements, not in questions. For example, prefer host_is_reachable() instead of is_host_reachable().

iripau.functools.retry(tries, exceptions=<class 'Exception'>, retry_condition=None, backoff_time=0)

Add retry capabilities to the decorated function.

Call the decorated function up to tries times, until it succeeds or it raises an unexpected exception. The decorated function Will only be retried if it raises one of the expected exceptions and retry_condition is fulfilled. Any other exception Will be raised immediately without further tries.

Parameters:

• tries (int) – The decorated function will be called at most this amount of times.

• exceptions (Union[Exception, Tuple[Exception]]) – The expected exceptions to trigger the retry.

• retry_condition (Callable[[Exception], bool]) – Function to call with the caught exception to verify if the retry process should continue. If it returns False, the caught exception will be raised.

• backoff_time (int) – Seconds to wait before the next try.

Return type:

Callable[[Callable[..., Any]], Callable[..., Any]]

Returns:

The decorator.

Example

Try to get the status of a server through a request to one of its REST API endpoints. Retry up to five times if Forbidden 403:

import requests

from urllib.parse import urljoin
from requests.exceptions import HTTPError


def http_code_is_forbidden(e):
    """ The status code in the response is 403 """
    return 403 == e.response.status_code


@retry(tries=5, exceptions=HTTPError, retry_condition=http_code_is_forbidden)
def get_server_status(server, headers):
    response = requests.get(urljoin(server, "api/status"), headers=headers)
    response.raise_for_status()  # Maybe HTTPError
    return response.json()

Example

Find a line with a specific content in a log file. Retry if the file has not been created yet or if the content could not be found:

@retry(tries=3, exceptions=(FileNotFoundError, StopIteration))
def get_log_entry(path, content):
    with open(path) as log:  # Maybe FileNotFoundError
        return next(line for line in log if content in line)  # Maybe StopIteration

Example

Get information of a book through a REST API endpoint. Retry if the book does not exist, but fail immediately if the response JSON does not contain the key "book":

import requests


@retry(tries=3, exceptions=KeyError)
def get_book_data(book):
    response = requests.get("https://localhost:8080/api/books")
    response.raise_for_status()
    books = response.json()["books"]  # Maybe KeyError

    yield  # Only retry if exceptions occur after this line

    return books[name]  # Maybe KeyError

iripau.functools.globalize(function)

Make function globally available in the module it was defined so it can be serialized. Useful when calling local functions with multiprocessing.

Parameters:

function (Callable[..., Any]) – Function to make global.

Return type:

Callable[..., Any]

Returns:

The decorated function.

Example

Use a nested function in some processes with multiprocessing:

import multiprocessing


def increase_multiple_processes(values, increment):

    @globalize
    def increase(x):
        return x + increment

    with multiprocessing.Pool(processes=3) as pool:
        return pool.map(increase, values)