celonis_data_objects.py
CelonisCollection (Generic, UserList)
¶
A UserList
implementation
to work with Celonis data objects.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
init_list |
Iterable[~T] |
An iterable of type |
required |
Source code in celonis_api/base_objects/celonis_data_objects.py
class CelonisCollection(typing.Generic[T], collections.UserList):
"""A [`UserList`](https://docs.python.org/3/library/collections.html#collections.UserList) implementation
to work with Celonis data objects.
Args:
init_list: An iterable of type `T` used to store the contents of the
[CelonisCollection][celonis_api.base_objects.celonis_data_objects.CelonisCollection].
"""
def __init__(self, init_list: typing.Iterable[T]):
super().__init__(init_list)
@property
def ids(self) -> typing.Dict[str, T]:
"""Dictionary of Celonis data objects with their respective ID as key."""
return {i.id: i for i in self}
@property
def names(self) -> typing.Dict[str, T]:
"""Dictionary of Celonis data objects with their respective `Name` as key.
If multiple Celonis data objects have the same `Name`, only the first one is mapped.
"""
d = {}
for i in self:
if hasattr(i, '_name') and i._name not in d:
d[i._name] = i
return d
@property
def keys(self) -> typing.Dict[str, T]:
"""Dictionary of Celonis data objects with their respective `Key` as key.
If multiple Celonis data objects have the same `Key`, only the first one is mapped.
"""
d = {}
for i in self:
if hasattr(i, '_key') and i._key not in d:
d[i._key] = i
return d
def find(
self,
id_or_name: str,
default: typing.Any = "error",
ignore_name_mapping: bool = True,
**kwargs,
) -> T:
"""Performs a search in the [CelonisCollection][celonis_api.base_objects.celonis_data_objects.CelonisCollection]
to find an object by ID or `Name` and returns the first matching result.
Args:
id_or_name: The ID or `Name` of the object to look for.
default: Whatever is set to default is returned if no matching object is found,
if `"error"` will raise [celonis_api.errors.PyCelonisNotFoundError][].
ignore_name_mapping: If True ignores name mappings of `#{<actual name>}`
to only check against `<actual name>`.
Raises:
PyCelonisNotFoundError: if default is set to `"error"`
Examples:
Use find to get the object via name:
```py
# iterate over all spaces and get the space by name "Default"
space = celonis.spaces.find("Default")
```
Returns:
The first item that matches `id_or_name` or the default.
"""
if isinstance(id_or_name, str):
if check_uuid(id_or_name):
search_by = "id"
else:
search_by = "_name"
else:
raise PyCelonisTypeError("Argument 'id_or_name' must be type string or int.")
if search_by == "id":
id_or_name_dict = self.ids
else:
id_or_name_dict = self.names
if ignore_name_mapping:
id_or_name = re.sub(r"#{([^}]+)}", r"\1", str(id_or_name)).replace('"', "")
id_or_name_dict = {
re.sub(r"#{([^}]+)}", r"\1", str(key)).replace('"', ""): value
for key, value in id_or_name_dict.items()
if value is not None and key is not None
}
result: T = id_or_name_dict.get(id_or_name, default)
if result == "error":
id_or_name_dict = self.keys
result = id_or_name_dict.get(id_or_name, default)
if result == "error":
raise PyCelonisNotFoundError(f"No result found for {id_or_name}.")
return result
def filter(self, condition: typing.Union[typing.Callable, str, int], field: str = None) -> 'CelonisCollection[T]':
"""Reduces the collection given the `condition`.
Args:
condition: Passed to builtin [`filter`](https://docs.python.org/3/library/functions.html#filter)
field: Define to check against a specific `data[field]` value with the condition.
Examples:
Use filter to reduce the Collection to objects matching the condition:
```py
# reduce list to all datamodels which have "Celonis" in their name
celonis_dms = c.datamodels.filter(lambda dm: "Celonis" in dm.name)
```
Returns:
A new CelonisCollection which items match the condition.
"""
if field:
return CelonisCollection(filter(lambda x: getattr(x, "data", {}).get(field, None) == condition, self))
elif callable(condition):
return CelonisCollection(filter(condition, self))
else:
return CelonisCollection([])
def __repr__(self):
if len(self) == 0:
return "[]"
else:
s = "".join([f"{str(i)},\n" for i in self])
return f"[\n{s}]"
ids: Dict[str, ~T]
property
readonly
¶
Dictionary of Celonis data objects with their respective ID as key.
keys: Dict[str, ~T]
property
readonly
¶
Dictionary of Celonis data objects with their respective Key
as key.
If multiple Celonis data objects have the same Key
, only the first one is mapped.
names: Dict[str, ~T]
property
readonly
¶
Dictionary of Celonis data objects with their respective Name
as key.
If multiple Celonis data objects have the same Name
, only the first one is mapped.
filter(self, condition, field=None)
¶
Reduces the collection given the condition
.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
condition |
Union[Callable, str, int] |
Passed to builtin |
required |
field |
str |
Define to check against a specific |
None |
Examples:
Use filter to reduce the Collection to objects matching the condition:
# reduce list to all datamodels which have "Celonis" in their name
celonis_dms = c.datamodels.filter(lambda dm: "Celonis" in dm.name)
Returns:
Type | Description |
---|---|
CelonisCollection[T] |
A new CelonisCollection which items match the condition. |
Source code in celonis_api/base_objects/celonis_data_objects.py
def filter(self, condition: typing.Union[typing.Callable, str, int], field: str = None) -> 'CelonisCollection[T]':
"""Reduces the collection given the `condition`.
Args:
condition: Passed to builtin [`filter`](https://docs.python.org/3/library/functions.html#filter)
field: Define to check against a specific `data[field]` value with the condition.
Examples:
Use filter to reduce the Collection to objects matching the condition:
```py
# reduce list to all datamodels which have "Celonis" in their name
celonis_dms = c.datamodels.filter(lambda dm: "Celonis" in dm.name)
```
Returns:
A new CelonisCollection which items match the condition.
"""
if field:
return CelonisCollection(filter(lambda x: getattr(x, "data", {}).get(field, None) == condition, self))
elif callable(condition):
return CelonisCollection(filter(condition, self))
else:
return CelonisCollection([])
find(self, id_or_name, default='error', ignore_name_mapping=True, **kwargs)
¶
Performs a search in the CelonisCollection
to find an object by ID or Name
and returns the first matching result.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id_or_name |
str |
The ID or |
required |
default |
Any |
Whatever is set to default is returned if no matching object is found,
if |
'error' |
ignore_name_mapping |
bool |
If True ignores name mappings of |
True |
Exceptions:
Type | Description |
---|---|
PyCelonisNotFoundError |
if default is set to |
Examples:
Use find to get the object via name:
# iterate over all spaces and get the space by name "Default"
space = celonis.spaces.find("Default")
Returns:
Type | Description |
---|---|
~T |
The first item that matches |
Source code in celonis_api/base_objects/celonis_data_objects.py
def find(
self,
id_or_name: str,
default: typing.Any = "error",
ignore_name_mapping: bool = True,
**kwargs,
) -> T:
"""Performs a search in the [CelonisCollection][celonis_api.base_objects.celonis_data_objects.CelonisCollection]
to find an object by ID or `Name` and returns the first matching result.
Args:
id_or_name: The ID or `Name` of the object to look for.
default: Whatever is set to default is returned if no matching object is found,
if `"error"` will raise [celonis_api.errors.PyCelonisNotFoundError][].
ignore_name_mapping: If True ignores name mappings of `#{<actual name>}`
to only check against `<actual name>`.
Raises:
PyCelonisNotFoundError: if default is set to `"error"`
Examples:
Use find to get the object via name:
```py
# iterate over all spaces and get the space by name "Default"
space = celonis.spaces.find("Default")
```
Returns:
The first item that matches `id_or_name` or the default.
"""
if isinstance(id_or_name, str):
if check_uuid(id_or_name):
search_by = "id"
else:
search_by = "_name"
else:
raise PyCelonisTypeError("Argument 'id_or_name' must be type string or int.")
if search_by == "id":
id_or_name_dict = self.ids
else:
id_or_name_dict = self.names
if ignore_name_mapping:
id_or_name = re.sub(r"#{([^}]+)}", r"\1", str(id_or_name)).replace('"', "")
id_or_name_dict = {
re.sub(r"#{([^}]+)}", r"\1", str(key)).replace('"', ""): value
for key, value in id_or_name_dict.items()
if value is not None and key is not None
}
result: T = id_or_name_dict.get(id_or_name, default)
if result == "error":
id_or_name_dict = self.keys
result = id_or_name_dict.get(id_or_name, default)
if result == "error":
raise PyCelonisNotFoundError(f"No result found for {id_or_name}.")
return result
CelonisDataObject (ABC)
¶
Base Celonis data object.
Each Celonis resource object can be uniquely identified by a key and an ID. In order to map the relation in the hierarchical structure, each data object references its parent object, e.g. a Studio Package's parent is the respective Studio Space. This object class is only intended to model the Celonis data object structure as an Abstract Base Class.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
parent |
CelonisApiObject |
The parent object. |
required |
data |
Dict |
The data of this object provided by the Celonis API. |
required |
Source code in celonis_api/base_objects/celonis_data_objects.py
class CelonisDataObject(abc.ABC):
"""Base Celonis data object.
Each Celonis resource object can be uniquely identified by a key and an ID.
In order to map the relation in the hierarchical structure, each data object
references its parent object, e.g. a Studio Package's parent is the respective
Studio Space.
This object class is only intended to model the Celonis data object structure
as an Abstract Base Class.
Args:
parent: The parent object.
data: The data of this object provided by the Celonis API.
"""
_name_data_key = "name"
_key_data_key = "key"
_id_data_key = "id"
def __init__(self, parent: 'CelonisApiObject', data: typing.Dict):
self.parent = parent
self._process_data(data)
def _process_data(self, data: typing.Dict):
self._data = data
self.id = data.get(self._id_data_key, None)
self._name = data.get(self._name_data_key, f"Unnamed object with ID: {self.id}")
self._key = data.get(self._key_data_key, f"No key in object with ID: {self.id}")
if not self.id:
self._logger.debug(f"Warning: no id found in: {data}")
@property
@abc.abstractmethod
def data(self):
"""A reference to the data of this object."""
@property
def name(self) -> str:
"""Name of the object via `data['name']` if available."""
self._name = self.data.get(self._name_data_key, f"Unnamed object with ID: {self.id}")
return self._name
@name.setter
def name(self, value):
"""Set the name only if `data['name']` exists."""
if self._name_data_key in self.data:
self.data[self._name_data_key] = value
self._name = value
@property
def key(self) -> str:
"""Key of the object via `data['key']` if available."""
self._key = self.data.get(self._key_data_key, f"No key in object with ID: {self.id}")
return self._key
@key.setter
def key(self, value):
"""Set the name only if `data['key']` exists."""
if self._key_data_key in self.data:
self.data[self._key_data_key] = value
self._key = value
def __repr__(self):
return f"<{self.__class__.__name__}, id {self.id}, name {self._name}>"
def __eq__(self, other):
return str(self.id) == str(other.id) if isinstance(other, CelonisDataObject) else False