knowledge_model.py
KPI (KnowledgeModelPQLColumn)
¶
KPIEvaluator
¶
Helper class to evaluate kpis in a PQL string.
Source code in celonis_api/studio/knowledge_model.py
class KPIEvaluator:
"""Helper class to evaluate kpis in a PQL string."""
KPI_NAME_REGEX = r"""(KPI[ \s]*\([A-Za-z_0-9, "'{}$\s]+\))"""
@staticmethod
def _extract_kpi_substrings(pql_string: str) -> typing.List[str]:
"""Returns all KPI substrings from PQL query (e.g. KPI("TEST_KPI"))"""
return re.findall(KPIEvaluator.KPI_NAME_REGEX, pql_string)
@staticmethod
def _clean_kpi_substring(kpi_substring):
kpi_substring = kpi_substring.strip()
kpi_substring = kpi_substring[3:].strip() # Remove "KPI"
kpi_substring = kpi_substring[1:-1].strip() # Remove brackets
return kpi_substring
@staticmethod
def _extract_kpi_names_and_parameter_values(
kpi_substrings: typing.List[str],
) -> typing.Tuple[typing.List[str], typing.List[typing.List[str]]]:
"""Returns names and parameters for all KPIs present in a KPI substring
E.g. for KPI("TEST", "PARAM1") it would return ['TEST'], [['PARAM1']]
"""
kpi_substrings_clean = [KPIEvaluator._clean_kpi_substring(k).split(",") for k in kpi_substrings]
names = [m[0].strip().replace('"', "").replace("'", "") for m in kpi_substrings_clean]
parameter_values = [[p.strip() for p in m[1:]] for m in kpi_substrings_clean]
return names, parameter_values
@staticmethod
def _replace_kpis(pql_string: str, kpis_processed: typing.Dict[str, KPI]) -> str:
kpi_substrings = KPIEvaluator._extract_kpi_substrings(pql_string)
kpi_names, kpi_parameter_values = KPIEvaluator._extract_kpi_names_and_parameter_values(kpi_substrings)
for kpi_substring, kpi_name, kpi_parameter_value in zip(kpi_substrings, kpi_names, kpi_parameter_values):
pql_string = KPIEvaluator._replace_single_kpi_substring(
pql_string, kpi_substring, kpi_name, kpi_parameter_value, kpis_processed
)
return pql_string
@staticmethod
def _replace_single_kpi_substring(
pql_string: str,
kpi_substring: str,
kpi_name: str,
kpi_parameter_values: typing.List[str],
kpis_processed: typing.Dict[str, KPI],
) -> str:
if kpi_name not in kpis_processed:
raise PyCelonisValueError(f"KPI {kpi_name} PQL does not exist in knowledge model.")
# Replace KPI string in PQL
kpi_replacement = kpis_processed[kpi_name]
kpi_replacement_str = kpi_replacement.pql
# Replace KPI parameters with values
if kpi_replacement.parameters:
for i, parameter in enumerate(kpi_replacement.parameters):
replacement = parameter.get("defaultValue", None)
# If parameter given in KPI call use this instead of default
if i < len(kpi_parameter_values):
replacement = kpi_parameter_values[i]
if replacement is not None:
kpi_replacement_str = kpi_replacement_str.replace(f"{{p{i+1}}}", str(replacement))
pql_string = pql_string.replace(kpi_substring, kpi_replacement_str)
return pql_string
@staticmethod
def _evaluate_single_kpi(kpi: KPI, kpis_processed: typing.Dict[str, KPI], kpi_dict: typing.Dict[str, KPI]):
kpi_substrings = KPIEvaluator._extract_kpi_substrings(kpi.pql)
kpi_names, _ = KPIEvaluator._extract_kpi_names_and_parameter_values(kpi_substrings)
if kpi.id in kpis_processed:
# KPI already computed. No need to do anything
return kpis_processed[kpi.id]
if not kpi_names:
# KPI does not contain nested KPIs and can just be returned
kpis_processed[kpi.id] = kpi
else:
# KPI contains nested KPIs which need to be evaluated first
unresolved_kpis = [kpi_dict[name] for name in kpi_names]
# Recursively evaluate unresolved kpis needed for the current one
for k in unresolved_kpis:
KPIEvaluator._evaluate_single_kpi(k, kpis_processed, kpi_dict)
# Replace kpis of current kpi with evaluated unresolved kpis
kpi.pql = KPIEvaluator._replace_kpis(kpi.pql, kpis_processed)
kpis_processed[kpi.id] = kpi
@staticmethod
def evaluate_kpis(pql_string: str, kpis: CelonisCollection[KPI]) -> str:
"""Evaluates the KPIs in a PQL string.
Args:
pql_string: PQL to be evaluated.
kpis: Collection of Knowledge Model KPIs to be processed.
Returns:
Evaluated PQL string.
"""
kpis_processed: typing.Dict[str, KPI] = {}
kpi_dict = {k.id: k for k in kpis}
# Get needed KPIs for pql_string
kpi_substrings = KPIEvaluator._extract_kpi_substrings(pql_string)
kpi_names, _ = KPIEvaluator._extract_kpi_names_and_parameter_values(kpi_substrings)
# Evaluate KPIs recursively
for kpi_name in kpi_names:
if kpi_name not in kpi_dict:
raise PyCelonisValueError("KPI does not exist!")
KPIEvaluator._evaluate_single_kpi(kpi_dict[kpi_name], kpis_processed, kpi_dict)
# Replace KPIs in pql_string with evaluated form
evaluated_pql_string = KPIEvaluator._replace_kpis(pql_string, kpis_processed)
return evaluated_pql_string
evaluate_kpis(pql_string, kpis)
staticmethod
¶
Evaluates the KPIs in a PQL string.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
pql_string |
str |
PQL to be evaluated. |
required |
kpis |
pycelonis.celonis_api.base_objects.celonis_data_objects.CelonisCollection[celonis_api.studio.knowledge_model.KPI] |
Collection of Knowledge Model KPIs to be processed. |
required |
Returns:
Type | Description |
---|---|
str |
Evaluated PQL string. |
Source code in celonis_api/studio/knowledge_model.py
@staticmethod
def evaluate_kpis(pql_string: str, kpis: CelonisCollection[KPI]) -> str:
"""Evaluates the KPIs in a PQL string.
Args:
pql_string: PQL to be evaluated.
kpis: Collection of Knowledge Model KPIs to be processed.
Returns:
Evaluated PQL string.
"""
kpis_processed: typing.Dict[str, KPI] = {}
kpi_dict = {k.id: k for k in kpis}
# Get needed KPIs for pql_string
kpi_substrings = KPIEvaluator._extract_kpi_substrings(pql_string)
kpi_names, _ = KPIEvaluator._extract_kpi_names_and_parameter_values(kpi_substrings)
# Evaluate KPIs recursively
for kpi_name in kpi_names:
if kpi_name not in kpi_dict:
raise PyCelonisValueError("KPI does not exist!")
KPIEvaluator._evaluate_single_kpi(kpi_dict[kpi_name], kpis_processed, kpi_dict)
# Replace KPIs in pql_string with evaluated form
evaluated_pql_string = KPIEvaluator._replace_kpis(pql_string, kpis_processed)
return evaluated_pql_string
KnowledgeModel (Node)
¶
Knowledge Model object to interact with Celonis Studio API.
Source code in celonis_api/studio/knowledge_model.py
class KnowledgeModel(Node):
"""Knowledge Model object to interact with Celonis Studio API."""
KIND = ["BASE", "EXTENSION"]
def __init__(self, parent, celonis, id_or_data):
super().__init__(parent, celonis, id_or_data)
self.static = True
self._content = None
if isinstance(id_or_data, str):
r = celonis.api_request(parent.package_nodes_url)
data = [
data
for data in r
if data['nodeType'] == 'ASSET' and data['assetType'] == 'SEMANTIC_MODEL' and data['id'] == id_or_data
]
if len(data) == 0:
raise ValueError(f"ID not found: {id_or_data}.")
data = data[0]
else:
data = id_or_data
self._url_final = f'{self.celonis.url}/semantic-layer/api/layer/{data["rootNodeKey"]}.{data["key"]}/final'
@property
def content(self) -> typing.Dict:
"""Get content of the Knowledge Model as raw dict.
Content is loaded form final endpoint to also obtain content of a possible base model.
Content is loaded from endpoint if initial load or `self.static = False`.
!!! api "API"
- `POST: /semantic-layer/api/layer/{root_node_key}.{key}/final`
```json
{
"withVariableReplacement": False,
"withAutogeneratedDataModelData": False
}
```
Returns:
Content as Dictionary.
"""
if self._content is None or self.static is False: # Load content from final endpoint
data = self.celonis.api_request(
self._url_final, message={"withVariableReplacement": True, "withAutogeneratedDataModelData": True}
)
self._content = yaml.load(data["yamlObject"], Loader=yaml.FullLoader)
return self._content
@property
def draft(self) -> 'KnowledgeModelDraft':
"""Obtain a draft of this knowledge model to edit content.
Draft is designed to behave similarly to the UI YAML edit mode.
Raises:
PyCelonisValueError: Draft is allowed with static=False (eager loading).
Returns:
Editable draft version of this Knowledge Model.
"""
if self.static:
raise PyCelonisValueError("Draft is not allowed for static Knowledge Models, set static=False first.")
data = self.data # Load from package-manager endpoint
return KnowledgeModelDraft(self.parent, self.celonis, data, self)
@property
def custom_objects(self) -> typing.List[typing.Dict]:
"""Returns custom objects of the Knowledge Model and its base model in case of an extension.
Returns:
List of custom objects.
"""
content = self.content
if "customObjects" not in content:
return []
return content["customObjects"]
@property
def kpis(self) -> 'CelonisCollection[KPI]':
"""Get all KPIs of the Knowledge Model and its base model in case of an extension.
Returns:
A Collection of KPIs.
"""
content = self.content
if "kpis" not in content:
return CelonisCollection([])
return CelonisCollection([KPI(self, kpi) for kpi in content["kpis"]])
@property
def filters(self) -> 'CelonisCollection[KnowledgeModelPQLFilter]':
"""Get all Filters of the Knowledge Model and its base model in case of an extension.
Returns:
A Collection of Filters.
"""
content = self.content
if "filters" not in content:
return CelonisCollection([])
return CelonisCollection([KnowledgeModelPQLFilter(self, f) for f in content["filters"]])
@property
def variables(self) -> typing.List[typing.Dict]:
"""Get all Variables of the Knowledge Model and its base model in case of an extension.
Returns:
List of variables
"""
content = self.content
if "variables" not in content:
return []
return content["variables"]
@property
def records(self) -> 'CelonisCollection[Record]':
"""Get all Records of knowledge model and its base model in case of an extension.
Returns:
A Collection of Records.
"""
content = self.content
if "records" not in content:
return CelonisCollection([])
return CelonisCollection([Record(self, k) for k in content["records"]])
content: Dict
property
readonly
¶
Get content of the Knowledge Model as raw dict.
Content is loaded form final endpoint to also obtain content of a possible base model.
Content is loaded from endpoint if initial load or self.static = False
.
API
POST: /semantic-layer/api/layer/{root_node_key}.{key}/final
Returns:
Type | Description |
---|---|
Dict |
Content as Dictionary. |
custom_objects: List[Dict]
property
readonly
¶
Returns custom objects of the Knowledge Model and its base model in case of an extension.
Returns:
Type | Description |
---|---|
List[Dict] |
List of custom objects. |
draft: KnowledgeModelDraft
property
readonly
¶
Obtain a draft of this knowledge model to edit content. Draft is designed to behave similarly to the UI YAML edit mode.
Exceptions:
Type | Description |
---|---|
PyCelonisValueError |
Draft is allowed with static=False (eager loading). |
Returns:
Type | Description |
---|---|
KnowledgeModelDraft |
Editable draft version of this Knowledge Model. |
filters: CelonisCollection[KnowledgeModelPQLFilter]
property
readonly
¶
Get all Filters of the Knowledge Model and its base model in case of an extension.
Returns:
Type | Description |
---|---|
CelonisCollection[KnowledgeModelPQLFilter] |
A Collection of Filters. |
kpis: CelonisCollection[KPI]
property
readonly
¶
Get all KPIs of the Knowledge Model and its base model in case of an extension.
Returns:
Type | Description |
---|---|
CelonisCollection[KPI] |
A Collection of KPIs. |
records: CelonisCollection[Record]
property
readonly
¶
Get all Records of knowledge model and its base model in case of an extension.
Returns:
Type | Description |
---|---|
CelonisCollection[Record] |
A Collection of Records. |
variables: List[Dict]
property
readonly
¶
Get all Variables of the Knowledge Model and its base model in case of an extension.
Returns:
Type | Description |
---|---|
List[Dict] |
List of variables |
KnowledgeModelDraft (Node)
¶
Knowledge Model Draft object to interact with Celonis Studio API.
Source code in celonis_api/studio/knowledge_model.py
class KnowledgeModelDraft(Node):
"""Knowledge Model Draft object to interact with Celonis Studio API."""
def __init__(self, parent, celonis, id_or_data, km: KnowledgeModel):
super().__init__(parent, celonis, id_or_data)
self.km = km
self.celonis._tracker.track(
"Connect to knowledge model",
extra={
"tracking_type": "STUDIO",
"km_id": self.id,
},
)
@property
def content(self) -> typing.Dict:
"""Get/Set the Content of the Knowledge Model, excluding a possible base model.
Returns:
Content of Knowledge Model.
"""
# Content is loaded from package-manager endpoint here.
return yaml.load(self.data["serializedContent"], Loader=yaml.FullLoader)
@content.setter
def content(self, value: typing.Dict):
data = self.data.copy()
data["serializedContent"] = yaml.dump(value, sort_keys=False)
self.data = data
@property
def custom_objects(self) -> typing.List[typing.Dict]:
"""Get/Set Custom Objects of the Knowledge Model, excluding a possible base model.
Returns:
List of custom objects
"""
content = self.content
if "customObjects" not in content:
content["customObjects"] = []
self.content = content
return content["customObjects"]
@custom_objects.setter
def custom_objects(self, value: typing.List[typing.Dict]):
content = self.content
content["customObjects"] = value
self.content = content
def add_custom_object(self, id: str, display_name: str, custom_attributes: typing.Dict = None):
"""Adds a Custom Object to the Knowledge Model.
Args:
id: ID of the Custom Object.
display_name: Display name used to display the Custom Object.
custom_attributes: Attributes of the Custom Object.
Raises:
PyCelonisValueError: If 'id' is not set or already exists.
PyCelonisValueError: If 'display_name' is not set.
"""
if not id:
raise PyCelonisValueError("Argument 'id' needs to be set for each custom object.")
if not display_name:
raise PyCelonisValueError("Argument 'display_name' needs to be set for each custom object.")
custom_objects = self.custom_objects
if id in [co["id"] for co in custom_objects]:
raise PyCelonisValueError(f"Custom Object with id {id} already exists.")
custom_object: typing.Dict = {"id": id, "displayName": display_name}
if custom_attributes:
custom_object["customAttributes"] = custom_attributes
custom_objects.append(custom_object)
self.custom_objects = custom_objects
def remove_custom_object(self, id: str):
"""Removes a Custom Object from the Knowledge Model by ID.
Args:
id: ID of the Custom Object.
"""
custom_objects = self.custom_objects
updated_custom_objects = [co for co in custom_objects if co["id"] != id]
if len(updated_custom_objects) == len(custom_objects):
raise PyCelonisValueError(f"Custom Object with id {id} does not exist.")
self.custom_objects = updated_custom_objects
@property
def kpis(self) -> typing.List[typing.Dict]:
"""Get/Set KPI objects of the Knowledge Model, excluding a possible base model.
Returns:
List of KPIs.
"""
content = self.content
if "kpis" not in content:
content["kpis"] = []
self.content = content
return content["kpis"]
@kpis.setter
def kpis(self, value: typing.List[typing.Dict]):
content = self.content
content["kpis"] = list(value)
self.content = content
@property
def filters(self) -> typing.List[typing.Dict]:
"""Get/Set Filter objects of the Knowledge Model, excluding a possible base model.
Returns:
List of Filters.
"""
content = self.content
if "filters" not in content:
content["filters"] = []
self.content = content
return content["filters"]
@filters.setter
def filters(self, value: typing.List[typing.Dict]):
content = self.content
content["filters"] = list(value)
self.content = content
@property
def variables(self) -> typing.List[typing.Dict]:
"""Get/Set Variable objects of the Knowledge Model, excluding a possible base model.
Returns:
List of variables
"""
content = self.content
if "variables" not in content:
content["variables"] = []
self.content = content
return content["variables"]
@variables.setter
def variables(self, value: typing.List[typing.Dict]):
content = self.content
content["variables"] = list(value)
self.content = content
@property
def records(self) -> typing.List[typing.Dict]:
"""Get/Set Record objects of the Knowledge Model, excluding a possible base model.
Returns:
List of records
"""
content = self.content
if "records" not in content:
content["records"] = []
self.content = content
return content["records"]
@records.setter
def records(self, value):
content = self.content
content["records"] = list(value)
self.content = content
def add_record(
self,
id: str,
display_name: str,
identifier: typing.Dict[str, str] = None,
pql: str = None,
filter_ids: typing.List[str] = None,
attributes: typing.List[typing.Dict[str, str]] = None,
):
"""Adds a Record object to the Knowledge Model.
Args:
id: ID of the Record.
display_name: Display name used to display the Record.
identifier: Identifier of the Record.
pql: PQL of the Record.
filter_ids: Filter IDs of the Record.
attributes: Attributes of the Record.
"""
records = self.content.get("records", [])
if id in [rec["id"] for rec in records]:
raise PyCelonisValueError(f"Record with id {id} already exists.")
record = {
"id": id,
"displayName": display_name,
"identifier": identifier if identifier else {},
"pql": pql if pql else "",
"filterIds": filter_ids if filter_ids else [],
"attributes": attributes if attributes else [],
}
records.append(record)
self.records = records
def remove_record(self, id: str):
"""Removes the Record from the Knowledge Model by ID.
Args:
id: ID of the Record.
"""
records = self.content.get("records", [])
if not records:
raise PyCelonisNotFoundError("No records found in knowledge model.")
updated_records = [rec for rec in records if rec["id"] != id]
if len(records) == len(updated_records):
raise PyCelonisValueError(f"Record with id {id} does not exist.")
self.records = updated_records
content: Dict
property
writable
¶
Get/Set the Content of the Knowledge Model, excluding a possible base model.
Returns:
Type | Description |
---|---|
Dict |
Content of Knowledge Model. |
custom_objects: List[Dict]
property
writable
¶
Get/Set Custom Objects of the Knowledge Model, excluding a possible base model.
Returns:
Type | Description |
---|---|
List[Dict] |
List of custom objects |
filters: List[Dict]
property
writable
¶
Get/Set Filter objects of the Knowledge Model, excluding a possible base model.
Returns:
Type | Description |
---|---|
List[Dict] |
List of Filters. |
kpis: List[Dict]
property
writable
¶
Get/Set KPI objects of the Knowledge Model, excluding a possible base model.
Returns:
Type | Description |
---|---|
List[Dict] |
List of KPIs. |
records: List[Dict]
property
writable
¶
Get/Set Record objects of the Knowledge Model, excluding a possible base model.
Returns:
Type | Description |
---|---|
List[Dict] |
List of records |
variables: List[Dict]
property
writable
¶
Get/Set Variable objects of the Knowledge Model, excluding a possible base model.
Returns:
Type | Description |
---|---|
List[Dict] |
List of variables |
add_custom_object(self, id, display_name, custom_attributes=None)
¶
Adds a Custom Object to the Knowledge Model.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id |
str |
ID of the Custom Object. |
required |
display_name |
str |
Display name used to display the Custom Object. |
required |
custom_attributes |
Dict |
Attributes of the Custom Object. |
None |
Exceptions:
Type | Description |
---|---|
PyCelonisValueError |
If 'id' is not set or already exists. |
PyCelonisValueError |
If 'display_name' is not set. |
Source code in celonis_api/studio/knowledge_model.py
def add_custom_object(self, id: str, display_name: str, custom_attributes: typing.Dict = None):
"""Adds a Custom Object to the Knowledge Model.
Args:
id: ID of the Custom Object.
display_name: Display name used to display the Custom Object.
custom_attributes: Attributes of the Custom Object.
Raises:
PyCelonisValueError: If 'id' is not set or already exists.
PyCelonisValueError: If 'display_name' is not set.
"""
if not id:
raise PyCelonisValueError("Argument 'id' needs to be set for each custom object.")
if not display_name:
raise PyCelonisValueError("Argument 'display_name' needs to be set for each custom object.")
custom_objects = self.custom_objects
if id in [co["id"] for co in custom_objects]:
raise PyCelonisValueError(f"Custom Object with id {id} already exists.")
custom_object: typing.Dict = {"id": id, "displayName": display_name}
if custom_attributes:
custom_object["customAttributes"] = custom_attributes
custom_objects.append(custom_object)
self.custom_objects = custom_objects
add_record(self, id, display_name, identifier=None, pql=None, filter_ids=None, attributes=None)
¶
Adds a Record object to the Knowledge Model.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id |
str |
ID of the Record. |
required |
display_name |
str |
Display name used to display the Record. |
required |
identifier |
Dict[str, str] |
Identifier of the Record. |
None |
pql |
str |
PQL of the Record. |
None |
filter_ids |
List[str] |
Filter IDs of the Record. |
None |
attributes |
List[Dict[str, str]] |
Attributes of the Record. |
None |
Source code in celonis_api/studio/knowledge_model.py
def add_record(
self,
id: str,
display_name: str,
identifier: typing.Dict[str, str] = None,
pql: str = None,
filter_ids: typing.List[str] = None,
attributes: typing.List[typing.Dict[str, str]] = None,
):
"""Adds a Record object to the Knowledge Model.
Args:
id: ID of the Record.
display_name: Display name used to display the Record.
identifier: Identifier of the Record.
pql: PQL of the Record.
filter_ids: Filter IDs of the Record.
attributes: Attributes of the Record.
"""
records = self.content.get("records", [])
if id in [rec["id"] for rec in records]:
raise PyCelonisValueError(f"Record with id {id} already exists.")
record = {
"id": id,
"displayName": display_name,
"identifier": identifier if identifier else {},
"pql": pql if pql else "",
"filterIds": filter_ids if filter_ids else [],
"attributes": attributes if attributes else [],
}
records.append(record)
self.records = records
remove_custom_object(self, id)
¶
Removes a Custom Object from the Knowledge Model by ID.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id |
str |
ID of the Custom Object. |
required |
Source code in celonis_api/studio/knowledge_model.py
def remove_custom_object(self, id: str):
"""Removes a Custom Object from the Knowledge Model by ID.
Args:
id: ID of the Custom Object.
"""
custom_objects = self.custom_objects
updated_custom_objects = [co for co in custom_objects if co["id"] != id]
if len(updated_custom_objects) == len(custom_objects):
raise PyCelonisValueError(f"Custom Object with id {id} does not exist.")
self.custom_objects = updated_custom_objects
remove_record(self, id)
¶
Removes the Record from the Knowledge Model by ID.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id |
str |
ID of the Record. |
required |
Source code in celonis_api/studio/knowledge_model.py
def remove_record(self, id: str):
"""Removes the Record from the Knowledge Model by ID.
Args:
id: ID of the Record.
"""
records = self.content.get("records", [])
if not records:
raise PyCelonisNotFoundError("No records found in knowledge model.")
updated_records = [rec for rec in records if rec["id"] != id]
if len(records) == len(updated_records):
raise PyCelonisValueError(f"Record with id {id} does not exist.")
self.records = updated_records
KnowledgeModelPQL
¶
A PQL in a Knowledge Model.
Source code in celonis_api/studio/knowledge_model.py
class KnowledgeModelPQL:
"""A PQL in a Knowledge Model."""
def __init__(self, knowledge_model: KnowledgeModel, data: typing.Dict):
self.knowledge_model = knowledge_model
self.data = data
self._name = data["id"]
self.id = data["id"]
self.pql = data["pql"]
self.description = data["description"] if "description" in data else None
self.parameters = data["parameters"] if "parameters" in data else None
self._supported_parameters = ["id", "displayName", "pql", "parameters", "type", "description", "attributes"]
@property
def query(self) -> str: # type: ignore
"""Get the PQL query of the Knowledge Model object."""
self._check_supported_parameters()
return KPIEvaluator.evaluate_kpis(self.pql, self.knowledge_model.kpis)
@query.setter
def query(self, value):
pass
def _check_supported_parameters(self):
not_supported_keys = [key for key in self.data.keys() if key not in self._supported_parameters]
if not_supported_keys:
self._logger.warning(
f"Parameters {', '.join(not_supported_keys)} currently not taken into account by pycelonis for "
f"PQL generation"
)
def __repr__(self) -> str:
s = ", ".join(f"{v}: {getattr(self, v)}" for v in ["id", "pql"] if getattr(self, v))
return f"<KnowledgeModelPQL, {s}>"
query: str
property
writable
¶
Get the PQL query of the Knowledge Model object.
KnowledgeModelPQLColumn (KnowledgeModelPQL, PQLColumn)
¶
A PQL Column in a Knowledge Model.
Source code in celonis_api/studio/knowledge_model.py
class KnowledgeModelPQLColumn(KnowledgeModelPQL, PQLColumn): # type: ignore
"""A PQL Column in a Knowledge Model."""
def __init__(self, knowledge_model: KnowledgeModel, data: typing.Dict):
KnowledgeModelPQL.__init__(self, knowledge_model=knowledge_model, data=data)
PQLColumn.__init__(
self, name=data["id"], format=None, sorting_index=None, sorting_direction=None, not_included=False
)
def __repr__(self) -> str:
s = ", ".join(f"{v}: {getattr(self, v)}" for v in ["id", "pql"] if getattr(self, v))
return f"<KnowledgeModelPQLColumn, {s}>"
KnowledgeModelPQLFilter (KnowledgeModelPQL, PQLFilter)
¶
A PQL Filter in a Knowledge Model.
Source code in celonis_api/studio/knowledge_model.py
class KnowledgeModelPQLFilter(KnowledgeModelPQL, PQLFilter): # type: ignore
"""A PQL Filter in a Knowledge Model."""
def __repr__(self) -> str:
s = ", ".join(f"{v}: {getattr(self, v)}" for v in ["id", "pql"] if getattr(self, v))
return f"<KnowledgeModelPQLFilter, {s}>"
def __hash__(self):
return hash(self.query)
def __str__(self):
return str(self.query)
Record (PQL)
¶
A Record in a Knowledge Model.
Source code in celonis_api/studio/knowledge_model.py
class Record(PQL):
"""A Record in a Knowledge Model."""
def __init__(self, knowledge_model: KnowledgeModel, data: typing.Dict):
super().__init__()
self.knowledge_model = knowledge_model
self._name = data["id"]
self.id = data["id"]
self.pql = data.get("pql", "")
self.display_name = data.get("displayName", "")
self.attributes = CelonisCollection(
[KnowledgeModelPQLColumn(self.knowledge_model, k) for k in data["attributes"]]
)
if "identifier" in data:
self.identifier = KnowledgeModelPQLColumn(self.knowledge_model, data["identifier"])
self.columns = self.attributes + [self.identifier] # type: ignore
else:
self.columns = self.attributes # type: ignore
def __repr__(self) -> str:
s = ", ".join(f"{v}: {getattr(self, v)}" for v in ["id", "attributes", "identifier"] if getattr(self, v, None))
return f"<Record, {s}>"
VariableEvaluator
¶
Helper class to evaluate variables in a PQL string.
Source code in celonis_api/studio/knowledge_model.py
class VariableEvaluator:
"""Helper class to evaluate variables in a PQL string."""
MAX_RECURSION = 20
@staticmethod
@deprecated("Variables are now automatically replaced in knowledge model.")
def evaluate_variables(pql_string: str, variables: typing.List[typing.Dict]) -> str:
"""Evaluates the variables in a PQL string.
Args:
pql_string: PQL to be evaluated.
variables: List of Knowledge Model variables to be processed.
Returns:
Evaluated PQL string.
"""
previous_pql_string = None
for i in range(VariableEvaluator.MAX_RECURSION):
if previous_pql_string == pql_string:
return pql_string # No changes done in last iteration -> All variables replaced
# Replace Variables
previous_pql_string = pql_string
for v in variables:
pql_string = pql_string.replace(f"${{{v['id']}}}", str(v['value']))
raise ValueError(
f"Could not evaluate variables in PQL {pql_string}. Maximum depth for nested variables is "
f"{VariableEvaluator.MAX_RECURSION}!"
)
evaluate_variables(pql_string, variables)
staticmethod
¶
Evaluates the variables in a PQL string.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
pql_string |
str |
PQL to be evaluated. |
required |
variables |
List[Dict] |
List of Knowledge Model variables to be processed. |
required |
Returns:
Type | Description |
---|---|
str |
Evaluated PQL string. |
Source code in celonis_api/studio/knowledge_model.py
@staticmethod
@deprecated("Variables are now automatically replaced in knowledge model.")
def evaluate_variables(pql_string: str, variables: typing.List[typing.Dict]) -> str:
"""Evaluates the variables in a PQL string.
Args:
pql_string: PQL to be evaluated.
variables: List of Knowledge Model variables to be processed.
Returns:
Evaluated PQL string.
"""
previous_pql_string = None
for i in range(VariableEvaluator.MAX_RECURSION):
if previous_pql_string == pql_string:
return pql_string # No changes done in last iteration -> All variables replaced
# Replace Variables
previous_pql_string = pql_string
for v in variables:
pql_string = pql_string.replace(f"${{{v['id']}}}", str(v['value']))
raise ValueError(
f"Could not evaluate variables in PQL {pql_string}. Maximum depth for nested variables is "
f"{VariableEvaluator.MAX_RECURSION}!"
)