Skip to content

knowledge_model.py

KPI (KnowledgeModelPQLColumn)

A KPI in a Knowledge Model.

Source code in celonis_api/studio/knowledge_model.py
class KPI(KnowledgeModelPQLColumn):
    """A KPI 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"<KPI, {s}>"

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
    {
        "withVariableReplacement": False,
        "withAutogeneratedDataModelData": False
    }
    

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}!"
    )