utils.py

Utilities to work with API objects.

`HttpMethod` ¶

Helper class for common used request methods.

Attributes:

Name	Type	Description
`GET`	`str`	`GET`
`POST`	`str`	`POST`
`PUT`	`str`	`PUT`
`DELETE`	`str`	`DELETE`

Source code in celonis_api/utils.py

class HttpMethod:
    """Helper class for common used request methods.

    Attributes:
        GET (str): `GET`
        POST (str): `POST`
        PUT (str): `PUT`
        DELETE (str): `DELETE`
    """

    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"

`KeyType` ¶

Helper class for Celonis API key types.

Attributes:

Name	Type	Description
`USER_KEY`	`str`	`USER_KEY`
`APP_KEY`	`str`	`APP_KEY`

Source code in celonis_api/utils.py

class KeyType:
    """Helper class for Celonis API key types.

    Attributes:
        USER_KEY (str): `USER_KEY`
        APP_KEY (str): `APP_KEY`
    """

    USER_KEY = "USER_KEY"
    APP_KEY = "APP_KEY"

    @classmethod
    def values(cls) -> str:
        return ', '.join([cls.USER_KEY, cls.APP_KEY])

`check_outdated(version)` ¶

Checks if the current PyCelonis version is outdated.

Parameters:

Name	Type	Description	Default
`version`	`str`	PyCelonis version to be checked.	required

Returns:

Type	Description
`Tuple[bool, str]`	Tuple of bool if version is outdated and string of latest version.

Source code in celonis_api/utils.py

def check_outdated(version: str) -> typing.Tuple[bool, str]:
    """Checks if the current PyCelonis version is outdated.

    Args:
        version: PyCelonis version to be checked.

    Returns:
         Tuple of bool if version is outdated and string of latest version.
    """
    from functools import reduce
    import pkg_resources
    import requests

    try:
        r = requests.get("https://pypi.celonis.cloud/pycelonis/")
        versions = [
            p.replace("pycelonis-", "").replace(".tar.gz", "")
            for p in r.text.split('"')
            if (
                ".tar.gz" in p
                and "href" not in p
                and "body" not in p
                and "a0" not in p
                and "a1" not in p
                and "b0" not in p
                and "b1" not in p
                and "dev" not in p
            )
        ]

        parse_versions = []
        for v in versions:
            try:
                parse_versions.append(pkg_resources.parse_version(v))
            except ValueError:
                pass

        latest_version = reduce(lambda x, y: x if x > y else y, parse_versions)
        is_outdated = latest_version > pkg_resources.parse_version(version)
    except Exception:
        is_outdated = False
        latest_version = version

    return is_outdated, latest_version

`check_uuid(_id)` ¶

Checks whether a string is a valid uuid4.

Parameters:

Name	Type	Description	Default
`_id`	`str`	The ID as string.	required

Returns:

Type	Description
`bool`	True if the string is a valid uuid4, else False.

Source code in celonis_api/utils.py

def check_uuid(_id: str) -> bool:
    """Checks whether a string is a valid uuid4.

    Args:
        _id: The ID as string.

    Returns:
        True if the string is a valid uuid4, else False.
    """
    return True if re.search(uuid_regex, _id) else False

`equal_dicts(d1, d2, ignore_keys=None)` ¶

Verifies whether two dicts are equal.

Parameters:

Name	Type	Description	Default
`d1`	`Dict`	First dictionary to be compared.	required
`d2`	`Dict`	Second dictionary to be compared.	required
`ignore_keys`	`List[str]`	List of keys to be ignored.	`None`

Returns:

Type	Description
`bool`	True if dictionaries are equal given `ignore_keys`, else False.

Source code in celonis_api/utils.py

def equal_dicts(d1: typing.Dict, d2: typing.Dict, ignore_keys: typing.List[str] = None) -> bool:
    """Verifies whether two dicts are equal.

    Args:
        d1: First dictionary to be compared.
        d2: Second dictionary to be compared.
        ignore_keys: List of keys to be ignored.

    Returns:
        True if dictionaries are equal given `ignore_keys`, else False.
    """
    if not (isinstance(d1, dict) and isinstance(d2, dict)):
        return False
    ignore_keys = ignore_keys or []
    d1_filtered = {k: v for k, v in d1.items() if k not in ignore_keys}
    d2_filtered = {k: v for k, v in d2.items() if k not in ignore_keys}
    return d1_filtered == d2_filtered

`find_object_from_list(list_of_dicts, _id)` ¶

Finds an object in a list of dicts by its ID.

Parameters:

Name	Type	Description	Default
`list_of_dicts`	`List[Dict]`	List of dictionaries with each dictionary containing a key called "id".	required
`_id`	`Union[str, int]`	The ID of the item to find in the list.	required

Returns:

Type	Description
`Dict`	The item with the matching ID.

Source code in celonis_api/utils.py

def find_object_from_list(list_of_dicts: typing.List[typing.Dict], _id: typing.Union[str, int]) -> typing.Dict:
    """Finds an object in a list of dicts by its ID.

    Args:
        list_of_dicts: List of dictionaries with each dictionary containing a key called "id".
        _id: The ID of the item to find in the list.

    Returns:
        The item with the matching ID.
    """
    i = None

    for dic in list_of_dicts:
        if dic["id"] == _id:
            i = dic
            break
    assert i, f"Object with {_id} not found!"
    return i

`pathify(weird_string, symbols_to_keep=' .')` ¶

Deletes anything that is not an alphanumeric character or something in symbols_to_keep from the string.

Parameters:

Name	Type	Description	Default
`weird_string`	`str`	String to be checked.	required
`symbols_to_keep`	`str`	String containing symbols to keep.	`' .'`

Returns:

Type	Description
`str`	String without non alphanumeric characters except for `symbols_to_keep`.

Source code in celonis_api/utils.py

def pathify(weird_string: str, symbols_to_keep: str = " .") -> str:
    """Deletes anything that is not an alphanumeric character or something in symbols_to_keep from the string.

    Args:
        weird_string: String to be checked.
        symbols_to_keep: String containing symbols to keep.

    Returns:
         String without non alphanumeric characters except for `symbols_to_keep`.
    """
    return "".join(c for c in weird_string if c.isalnum() or c in symbols_to_keep).rstrip()

`threaded(iterable, function, *args, **kwargs)` ¶

Can be used to write parallel for loops using multiprocessing using joblib.Parallel.

Parameters:

Name	Type	Description	Default
`iterable`	`Iterable`	An iterable of which each item will passed as argument to the function.	required
`function`	`Callable`	A function that should be able to handle each item passed by the iterable.	required
`*args`		Additional arguments passed to the function.	`()`
`**kwargs`		Additional keyword named arguments passed to function.	`{}`

Yields:

Type	Description
`Generator`	Result of the function for each item in the iterable.

Source code in celonis_api/utils.py

def threaded(iterable: typing.Iterable, function: typing.Callable, *args, **kwargs) -> typing.Generator:
    """Can be used to write parallel for loops using multiprocessing
        using [joblib.Parallel](https://joblib.readthedocs.io/en/latest/generated/joblib.Parallel.html#joblib.Parallel).

    Args:
        iterable: An iterable of which each item will passed as argument to the function.
        function: A function that should be able to handle each item passed by the iterable.
        *args: Additional arguments passed to the function.
        **kwargs: Additional keyword named arguments passed to function.

    Yields:
        Result of the function for each item in the iterable.
    """
    n_jobs = int(os.environ.get("PYCELONIS_NUM_THREADS") or -1)
    if n_jobs == 1:
        for i in iterable:
            yield function(i, *args, **kwargs)
    else:
        for r in Parallel(n_jobs=n_jobs, backend="threading")(delayed(function)(i, *args, **kwargs) for i in iterable):
            yield r

`update_dict(d1, d2)` ¶

Updates a dictionary with values from another dictionary recursively.

Parameters:

Name	Type	Description	Default
`d1`	`Dict`	First dictionary to be extended.	required
`d2`	`Mapping`	Second dictionary to be copied.	required

Returns:

Type	Description
`Dict`	First dictionary updated with valued form second dictionary.

Source code in celonis_api/utils.py

def update_dict(d1: typing.Dict, d2: collections.abc.Mapping) -> typing.Dict:
    """Updates a dictionary with values from another dictionary recursively.

    Args:
        d1: First dictionary to be extended.
        d2: Second dictionary to be copied.

    Returns:
        First dictionary updated with valued form second dictionary.
    """
    for k, v in d2.items():
        if isinstance(v, collections.abc.Mapping):
            d1[k] = update_dict(d1.get(k, {}), v)
        else:
            d1[k] = v
    return d1

utils.py

HttpMethod ¶

KeyType ¶

check_outdated(version) ¶

check_uuid(_id) ¶

equal_dicts(d1, d2, ignore_keys=None) ¶

find_object_from_list(list_of_dicts, _id) ¶

pathify(weird_string, symbols_to_keep=' .') ¶

threaded(iterable, function, *args, **kwargs) ¶

update_dict(d1, d2) ¶

`HttpMethod` ¶

`KeyType` ¶

`check_outdated(version)` ¶

`check_uuid(_id)` ¶

`equal_dicts(d1, d2, ignore_keys=None)` ¶

`find_object_from_list(list_of_dicts, _id)` ¶

`pathify(weird_string, symbols_to_keep=' .')` ¶

`threaded(iterable, function, *args, **kwargs)` ¶

`update_dict(d1, d2)` ¶