package.py

Package (Node)

Package object to interact with Celonis Studio API.

Source code in celonis_api/studio/package.py

class Package(Node):
    """Package object to interact with Celonis Studio API."""

    TYPES = ["APP", "LIBRARY", "INSTRUMENT"]
    VARIABLE_TYPES = ["DATA_MODEL", "CONNECTION"]

    @property
    def _parent_class(self):
        from pycelonis.celonis_api.studio.space import Space

        return Space

    @property
    def package_nodes_url(self) -> str:
        """
        !!! api "API"
            - `/package-manager/api/nodes/by-root-id/{package_id}`
        """
        return f"{self.celonis.url}/package-manager/api/nodes/by-root-id/{self.id}"

    @property
    def publish_url(self) -> str:
        """
        !!! api "API"
            - `/package-manager/api/packages/{package_key}/activate`
        """
        return f"{self.celonis.url}/package-manager/api/packages/{self.data['key']}/activate"

    @property
    def variable_assignments_url(self) -> str:
        """
        !!! api "API"
            - `/package-manager/api/nodes/by-package-key/{package_key}/variables`
        """
        return f"{self.celonis.url}/package-manager/api/nodes/by-package-key/{self.key}/variables"

    @property
    def draft_id(self) -> str:
        """Get the Package Draft ID."""
        return self.data["draftId"]

    @property
    def active_version(self):
        """Get the current active version of the Package."""
        current_active_version = [
            h["version"] for h in self.history if "version" in h and "active" in h and h["active"]
        ]
        if current_active_version:
            return current_active_version[0]
        else:  # No version activated so far
            return None

    @property
    def content(self) -> typing.Dict:
        """Get/Set the Package Content.

        Returns:
            Serialized Content of the Package.
        """
        return yaml.load(self.data["serializedContent"], Loader=yaml.FullLoader)

    @content.setter
    def content(self, value: typing.Dict):
        self.data["serializedContent"] = yaml.dump(value, sort_keys=False)

    @property
    def variables(self) -> typing.List[typing.Dict]:
        """Get/Set all Variables of the Package.

        Returns:
            All Variables of the Package.
        """
        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"] = value
        self.content = content

    @property
    def variable_assignments(self) -> typing.List[typing.Dict]:
        """Get all Variable Assignments of the Package.

        !!! api "API"
            - `GET: /package-manager/api/nodes/by-package-key/{package_key}/variables/values`

        Returns:
            List of Variable Assignments.
        """
        return self.celonis.api_request(self.variable_assignments_url + "/values")

    @property
    def history(self) -> typing.List[typing.Dict]:
        """Get the Package History.

        !!! api "API"
            - `GET: /package-manager/api/packages/{package_id}/history`

        Returns:
            List of dictionaries containing history of published versions and active versions.
        """
        return self.celonis.api_request(f"{self.celonis.url}/package-manager/api/packages/{self.id}/history")

    def get_node_by_key(self, node_key: str) -> BaseNode:
        """Get a Node of the Package by Key.

        !!! api "API"
            - `GET: /package-manager/api/nodes/by-root-id/{package_id}`

        Args:
            node_key: Key of the node.

        Raises:
            PyCelonisNotFoundError: If Node was not found.

        Returns:
            The specific Node by Key.
        """
        response = self.celonis.api_request(self.package_nodes_url)
        result = CelonisCollection(
            [
                NodeFactory.create_node_from_data(self, self.celonis, data)
                for data in response
                if data["key"] == node_key
            ]
        )
        if len(result) == 0:
            raise PyCelonisNotFoundError(f"No node with 'id': {node_key} found in package.")

        return result[0]

    @property
    def nodes(self) -> 'CelonisCollection[BaseNode]':
        """Get all Nodes of the Package.

        Returns:
            A Collection of Nodes of the Package.
        """
        response = self.celonis.api_request(self.package_nodes_url)
        return CelonisCollection(
            [
                NodeFactory.create_node_from_data(self, self.celonis, data)
                for data in response
                if data["nodeType"] != 'PACKAGE'
            ]
        )

    @property
    def knowledge_models(self) -> 'CelonisCollection[KnowledgeModel]':
        """Get all Knowledge Models of the Package.

        Returns:
            A Collection of Knowledge Models of the Package.
        """
        return CelonisCollection([n for n in self.nodes if isinstance(n, KnowledgeModel)])

    @property
    def analyses(self) -> 'CelonisCollection[Analysis]':
        """Get all Analysis of the Package.

        Returns:
            A Collection of Analysis of the Package.
        """
        return CelonisCollection([n for n in self.nodes if isinstance(n, Analysis)])

    @property
    def views(self) -> 'CelonisCollection[View]':
        """Get all Views of the Package.

        Returns:
            A Collection of Views of the Package.
        """
        return CelonisCollection([n for n in self.nodes if isinstance(n, View)])

    @property
    def skills(self) -> 'CelonisCollection[Skill]':
        """Get all Skills of the Package.

        Returns:
            A Collection of Skills of the Package.
        """
        return CelonisCollection([n for n in self.nodes if isinstance(n, Skill)])

    @property
    def folders(self) -> 'CelonisCollection[Folder]':
        """Get all Folders of the Package.

        Returns:
            A Collection of Folders of the Package.
        """
        return CelonisCollection([n for n in self.nodes if isinstance(n, Folder)])

    def publish(self, version: str = None, node_ids_to_exclude: typing.List[str] = None):
        """Publishes current package using new version

        Parameters
        ----------
        version : str, optional
            New version number following semantic versioning format (X.X.X). If left blank the currently active version
            number will be used and increased by one (until a nonexistent version number is found).
        node_ids_to_exclude : list of str
            List of ids of nodes that will be excluded from publishing
        """
        if not version:
            version = self._get_next_version()

        node_ids_to_exclude = node_ids_to_exclude or []
        message = {"version": version, "nodeIdsToExclude": node_ids_to_exclude, "packageKey": self.data["key"]}
        self.celonis.api_request(self.publish_url, message=message, method=HttpMethod.POST)
        self._logger.info(f"Successfully published version {version} of package.")

    def create_variable(
        self, name: str, value: str, variable_type: str = "DATA_MODEL", description: str = ""
    ) -> typing.Dict:
        """Creates a new Variable in the Package and assigns the value to it.

        !!! api "API"
            - `POST: /package-manager/api/nodes/by-package-key/{package_key}/variables`
                ```json
                {
                    "key": name,
                    "value": value,
                    "type": variable_type,
                    "description": description
                }
                ```

        Args:
            name: Name of the Variable.
            value: Value to be set for the Variable.
            variable_type: Type of the Variable. One of [`DATA_MODEL`, `CONNECTION`].
            description: Description for the Variable.

        Raises:
            PyCelonisValueError: If `variable_type` is invalid.

        Returns:
            The newly created Variable.
        """
        if variable_type.upper() not in self.VARIABLE_TYPES:
            raise PyCelonisValueError(f"Argument 'variable_type' must be one of {', '.join(self.VARIABLE_TYPES)} .")

        # Update variable assignments
        payload = {'key': name, 'value': value, 'type': variable_type, 'description': description}
        return self.celonis.api_request(self.variable_assignments_url, payload, method=HttpMethod.POST)

    def create_knowledge_model(
        self,
        name: str,
        key: str = None,
        data_model_id: str = "",
        data_model_variable: str = "",
        kind: str = "BASE",
        base_key: str = None,
    ) -> KnowledgeModel:
        """Creates a new Knowledge Model in the Package.

        !!! api "API"
            - `POST: /semantic-layer/api/semantic-models`
                ```json
                {
                    "parentNodeId": self.id,
                    "finalModelOptions": {
                        "withVariableReplacement": True,
                        "withAutogeneratedDataModelData": True
                    },
                    # kind=EXTENSION
                    "base": {
                        "key": base_key,
                        "version": base_km.parent.active_version
                    }
                    "content": {
                        "kind": kind,
                        "metadata": {
                            "key": key or name,
                            "displayName": name,
                        },
                        "dataModelId": data_model_id or "{{data_model_variable}}",
                    }
                }
                ```

        Args:
            name: Name of Knowledge Model.
            key: Key of Knowledge Model.
            data_model_id: ID of associated Datamodel (can't be set at the same time as `data_model_variable`).
            data_model_variable: Key associated with desired Datamodel.
            kind: Kind of Knowledge Model. One of [`BASE`, `EXTENSION`].
            base_key: Key of Base Knowledge Model. Only if `kind=EXTENSION`.

        Raises:
            PyCelonisValueError: If both `data_model_id` and `data_model_variable` are set or none of them.
            PyCelonisValueError: If `kind` is invalid.
            PyCelonisValueError: If 'base_key' not set when `kind=EXTENSION`.

        Returns:
            The newly created Knowledge Model.
        """
        if data_model_id and data_model_variable:
            raise PyCelonisValueError("Can't set both 'data_model_id' and 'data_model_variable'.")
        if not data_model_id and not data_model_variable:
            raise PyCelonisValueError("Either 'data_model_id' or 'data_model_variable' needs to be set.")
        if kind not in KnowledgeModel.KIND:
            raise PyCelonisValueError(f"Argument 'kind' must be one of {', '.join(KnowledgeModel.KIND)}.")
        if kind == "EXTENSION" and base_key is None:
            raise PyCelonisValueError("Argument 'base_key' needs to be set to key of base knowledge model.")

        data_model = data_model_id or "${{" + data_model_variable + "}}"

        content = {
            "kind": kind,
            "metadata": {
                "key": key or name,
                "displayName": name,
            },
            "dataModelId": data_model,
        }

        if kind == "EXTENSION":
            if not base_key:
                raise PyCelonisValueError("base_key needs to be defined for EXTENSION knowledge models")

            if "." not in base_key:
                self.get_node_by_key(base_key)  # Check that Base KM exists
                content.update({"base": {"key": base_key}})
            else:
                base_km = self.celonis.get_knowledge_model_by_full_key(base_key)
                if base_km.parent.active_version is None:
                    raise PyCelonisValueError(
                        f"Base knowledge model package {base_key.split('.')[0]} is not active. "
                        f"Please publish it to make it possible to be used as base."
                    )
                content.update({"base": {"key": base_key, "version": base_km.parent.active_version}})

        payload = {
            "parentNodeId": self.id,
            "finalModelOptions": {"withVariableReplacement": True, "withAutogeneratedDataModelData": True},
            "content": yaml.dump(content, sort_keys=False),
        }

        url = f"{self.celonis.url}/semantic-layer/api/semantic-models"
        response = self.celonis.api_request(url, payload, method=HttpMethod.POST)
        return KnowledgeModel(self, self.celonis, response["id"])

    def create_analysis(
        self, name: str, data_model_variable: str = "", data_model_id: str = "", key: str = None
    ) -> Analysis:
        """Creates a new Analysis in the Package.

        !!! api "API"
            - `POST: /process-analytics/analysis/v2/api/analysis`
                ```json
                {
                    "name": name,
                    "key": key or name,
                    "dataModelId": data_model_id,
                    "knowledgeModelId": "",
                    "rootNodeKey": self.name,
                    "rootNodeId": self.id,
                    "parentNodeKey": self.name,
                    "parentNodeId": self.id,
                    "rootWithKey": self.name,
                }
                ```

        Args:
            name: Name of the Analysis.
            data_model_variable: Key associated with the desired Datamodel.
            data_model_id: Uuid of the Datamodel.
            key: Key of the Analysis.

        Raises:
            PyCelonisValueError: If both `data_model_id` and `data_model_variable` are set or none of them.

        Returns:
            The newly created Analysis.
        """
        if data_model_id and data_model_variable:
            raise PyCelonisValueError("Can't set both 'data_model_id' and 'data_model_variable'.")
        if not data_model_id and not data_model_variable:
            raise PyCelonisValueError("Either 'data_model_id' or 'data_model_variable' needs to be set.")

        if data_model_variable:
            data_model_id = "${{" + data_model_variable + "}}"

        payload = {
            "name": name,
            "key": key or name,
            "dataModelId": data_model_id,
            "knowledgeModelId": "",
            "rootNodeKey": self.name,
            "rootNodeId": self.id,
            "parentNodeKey": self.name,
            "parentNodeId": self.id,
            "rootWithKey": self.name,
        }

        url = f"{self.celonis.url}/process-analytics/analysis/v2/api/analysis"
        response = self.celonis.api_request(url, payload, method=HttpMethod.POST)
        return Analysis(self, self.celonis, response["analysis"]["id"])

    def delete(self):
        """Deletes the Package.

        !!! api "API"
            - `DELETE: /package-manager/api/packages/{self.id}`
        """
        payload = {
            "id": self.id,
            "name": self.name,
        }
        url = f"{self.celonis.url}/package-manager/api/packages/delete/{self.id}"
        self.celonis.api_request(url, payload, method=HttpMethod.POST)

    def _get_next_version(self) -> str:
        if self.active_version is None:  # No version activated so far
            return "0.0.1"

        existing_versions = [h["version"] for h in self.history if "version" in h]

        next_active_version = [int(n) for n in self.active_version.split(".")]

        # Increase version number until nonexistent one is found
        while ".".join([str(n) for n in next_active_version]) in existing_versions:
            next_active_version[-1] += 1
        return ".".join([str(n) for n in next_active_version])

    def _add_dependency(self, dependency_package):
        url = self.celonis.url + f"/package-manager/api/package-dependencies/{self.id}"
        payload = [
            {
                "updateAvailable": False,
                "id": dependency_package.id,
                "deleted": False,
                "draftId": dependency_package.draft_id,
                "external": False,
                "key": dependency_package.key,
                "name": dependency_package.name,
                "rootNodeId": self.data["rootNodeId"],
                "version": dependency_package.active_version,
            }
        ]
        return self.celonis.api_request(url, payload, method=HttpMethod.POST)

active_version property readonly

Get the current active version of the Package.

analyses: CelonisCollection[Analysis] property readonly

Get all Analysis of the Package.

Returns:

Type	Description
CelonisCollection[Analysis]	A Collection of Analysis of the Package.

content: Dict property writable

Get/Set the Package Content.

Returns:

Type	Description
Dict	Serialized Content of the Package.

draft_id: str property readonly

Get the Package Draft ID.

folders: CelonisCollection[Folder] property readonly

Get all Folders of the Package.

Returns:

Type	Description
CelonisCollection[Folder]	A Collection of Folders of the Package.

history: List[Dict] property readonly

Get the Package History.

API

• GET: /package-manager/api/packages/{package_id}/history

Returns:

Type	Description
List[Dict]	List of dictionaries containing history of published versions and active versions.

knowledge_models: CelonisCollection[KnowledgeModel] property readonly

Get all Knowledge Models of the Package.

Returns:

Type	Description
CelonisCollection[KnowledgeModel]	A Collection of Knowledge Models of the Package.

nodes: CelonisCollection[BaseNode] property readonly

Get all Nodes of the Package.

Returns:

Type	Description
CelonisCollection[BaseNode]	A Collection of Nodes of the Package.

package_nodes_url: str property readonly

API

• /package-manager/api/nodes/by-root-id/{package_id}

publish_url: str property readonly

API

• /package-manager/api/packages/{package_key}/activate

skills: CelonisCollection[Skill] property readonly

Get all Skills of the Package.

Returns:

Type	Description
CelonisCollection[Skill]	A Collection of Skills of the Package.

variable_assignments: List[Dict] property readonly

Get all Variable Assignments of the Package.

API

• GET: /package-manager/api/nodes/by-package-key/{package_key}/variables/values

Returns:

Type	Description
List[Dict]	List of Variable Assignments.

variable_assignments_url: str property readonly

API

• /package-manager/api/nodes/by-package-key/{package_key}/variables

variables: List[Dict] property writable

Get/Set all Variables of the Package.

Returns:

Type	Description
List[Dict]	All Variables of the Package.

views: CelonisCollection[View] property readonly

Get all Views of the Package.

Returns:

Type	Description
CelonisCollection[View]	A Collection of Views of the Package.

create_analysis(self, name, data_model_variable='', data_model_id='', key=None)

Creates a new Analysis in the Package.

API

• POST: /process-analytics/analysis/v2/api/analysis

  {
      "name": name,
      "key": key or name,
      "dataModelId": data_model_id,
      "knowledgeModelId": "",
      "rootNodeKey": self.name,
      "rootNodeId": self.id,
      "parentNodeKey": self.name,
      "parentNodeId": self.id,
      "rootWithKey": self.name,
  }
  

Parameters:

Name	Type	Description	Default
name	str	Name of the Analysis.	required
data_model_variable	str	Key associated with the desired Datamodel.	''
data_model_id	str	Uuid of the Datamodel.	''
key	str	Key of the Analysis.	None

Exceptions:

Type	Description
PyCelonisValueError	If both data_model_id and data_model_variable are set or none of them.

Returns:

Type	Description
Analysis	The newly created Analysis.

Source code in celonis_api/studio/package.py

def create_analysis(
    self, name: str, data_model_variable: str = "", data_model_id: str = "", key: str = None
) -> Analysis:
    """Creates a new Analysis in the Package.

    !!! api "API"
        - `POST: /process-analytics/analysis/v2/api/analysis`
            ```json
            {
                "name": name,
                "key": key or name,
                "dataModelId": data_model_id,
                "knowledgeModelId": "",
                "rootNodeKey": self.name,
                "rootNodeId": self.id,
                "parentNodeKey": self.name,
                "parentNodeId": self.id,
                "rootWithKey": self.name,
            }
            ```

    Args:
        name: Name of the Analysis.
        data_model_variable: Key associated with the desired Datamodel.
        data_model_id: Uuid of the Datamodel.
        key: Key of the Analysis.

    Raises:
        PyCelonisValueError: If both `data_model_id` and `data_model_variable` are set or none of them.

    Returns:
        The newly created Analysis.
    """
    if data_model_id and data_model_variable:
        raise PyCelonisValueError("Can't set both 'data_model_id' and 'data_model_variable'.")
    if not data_model_id and not data_model_variable:
        raise PyCelonisValueError("Either 'data_model_id' or 'data_model_variable' needs to be set.")

    if data_model_variable:
        data_model_id = "${{" + data_model_variable + "}}"

    payload = {
        "name": name,
        "key": key or name,
        "dataModelId": data_model_id,
        "knowledgeModelId": "",
        "rootNodeKey": self.name,
        "rootNodeId": self.id,
        "parentNodeKey": self.name,
        "parentNodeId": self.id,
        "rootWithKey": self.name,
    }

    url = f"{self.celonis.url}/process-analytics/analysis/v2/api/analysis"
    response = self.celonis.api_request(url, payload, method=HttpMethod.POST)
    return Analysis(self, self.celonis, response["analysis"]["id"])

create_knowledge_model(self, name, key=None, data_model_id='', data_model_variable='', kind='BASE', base_key=None)

Creates a new Knowledge Model in the Package.

API

• POST: /semantic-layer/api/semantic-models

  {
      "parentNodeId": self.id,
      "finalModelOptions": {
          "withVariableReplacement": True,
          "withAutogeneratedDataModelData": True
      },
      # kind=EXTENSION
      "base": {
          "key": base_key,
          "version": base_km.parent.active_version
      }
      "content": {
          "kind": kind,
          "metadata": {
              "key": key or name,
              "displayName": name,
          },
          "dataModelId": data_model_id or "{{data_model_variable}}",
      }
  }
  

Parameters:

Name	Type	Description	Default
name	str	Name of Knowledge Model.	required
key	str	Key of Knowledge Model.	None
data_model_id	str	ID of associated Datamodel (can't be set at the same time as data_model_variable).	''
data_model_variable	str	Key associated with desired Datamodel.	''
kind	str	Kind of Knowledge Model. One of [BASE, EXTENSION].	'BASE'
base_key	str	Key of Base Knowledge Model. Only if kind=EXTENSION.	None

Exceptions:

Type	Description
PyCelonisValueError	If both data_model_id and data_model_variable are set or none of them.
PyCelonisValueError	If kind is invalid.
PyCelonisValueError	If 'base_key' not set when kind=EXTENSION.

Returns:

Type	Description
KnowledgeModel	The newly created Knowledge Model.

Source code in celonis_api/studio/package.py

def create_knowledge_model(
    self,
    name: str,
    key: str = None,
    data_model_id: str = "",
    data_model_variable: str = "",
    kind: str = "BASE",
    base_key: str = None,
) -> KnowledgeModel:
    """Creates a new Knowledge Model in the Package.

    !!! api "API"
        - `POST: /semantic-layer/api/semantic-models`
            ```json
            {
                "parentNodeId": self.id,
                "finalModelOptions": {
                    "withVariableReplacement": True,
                    "withAutogeneratedDataModelData": True
                },
                # kind=EXTENSION
                "base": {
                    "key": base_key,
                    "version": base_km.parent.active_version
                }
                "content": {
                    "kind": kind,
                    "metadata": {
                        "key": key or name,
                        "displayName": name,
                    },
                    "dataModelId": data_model_id or "{{data_model_variable}}",
                }
            }
            ```

    Args:
        name: Name of Knowledge Model.
        key: Key of Knowledge Model.
        data_model_id: ID of associated Datamodel (can't be set at the same time as `data_model_variable`).
        data_model_variable: Key associated with desired Datamodel.
        kind: Kind of Knowledge Model. One of [`BASE`, `EXTENSION`].
        base_key: Key of Base Knowledge Model. Only if `kind=EXTENSION`.

    Raises:
        PyCelonisValueError: If both `data_model_id` and `data_model_variable` are set or none of them.
        PyCelonisValueError: If `kind` is invalid.
        PyCelonisValueError: If 'base_key' not set when `kind=EXTENSION`.

    Returns:
        The newly created Knowledge Model.
    """
    if data_model_id and data_model_variable:
        raise PyCelonisValueError("Can't set both 'data_model_id' and 'data_model_variable'.")
    if not data_model_id and not data_model_variable:
        raise PyCelonisValueError("Either 'data_model_id' or 'data_model_variable' needs to be set.")
    if kind not in KnowledgeModel.KIND:
        raise PyCelonisValueError(f"Argument 'kind' must be one of {', '.join(KnowledgeModel.KIND)}.")
    if kind == "EXTENSION" and base_key is None:
        raise PyCelonisValueError("Argument 'base_key' needs to be set to key of base knowledge model.")

    data_model = data_model_id or "${{" + data_model_variable + "}}"

    content = {
        "kind": kind,
        "metadata": {
            "key": key or name,
            "displayName": name,
        },
        "dataModelId": data_model,
    }

    if kind == "EXTENSION":
        if not base_key:
            raise PyCelonisValueError("base_key needs to be defined for EXTENSION knowledge models")

        if "." not in base_key:
            self.get_node_by_key(base_key)  # Check that Base KM exists
            content.update({"base": {"key": base_key}})
        else:
            base_km = self.celonis.get_knowledge_model_by_full_key(base_key)
            if base_km.parent.active_version is None:
                raise PyCelonisValueError(
                    f"Base knowledge model package {base_key.split('.')[0]} is not active. "
                    f"Please publish it to make it possible to be used as base."
                )
            content.update({"base": {"key": base_key, "version": base_km.parent.active_version}})

    payload = {
        "parentNodeId": self.id,
        "finalModelOptions": {"withVariableReplacement": True, "withAutogeneratedDataModelData": True},
        "content": yaml.dump(content, sort_keys=False),
    }

    url = f"{self.celonis.url}/semantic-layer/api/semantic-models"
    response = self.celonis.api_request(url, payload, method=HttpMethod.POST)
    return KnowledgeModel(self, self.celonis, response["id"])

create_variable(self, name, value, variable_type='DATA_MODEL', description='')

Creates a new Variable in the Package and assigns the value to it.

API

• POST: /package-manager/api/nodes/by-package-key/{package_key}/variables

  {
      "key": name,
      "value": value,
      "type": variable_type,
      "description": description
  }
  

Parameters:

Name	Type	Description	Default
name	str	Name of the Variable.	required
value	str	Value to be set for the Variable.	required
variable_type	str	Type of the Variable. One of [DATA_MODEL, CONNECTION].	'DATA_MODEL'
description	str	Description for the Variable.	''

Exceptions:

Type	Description
PyCelonisValueError	If variable_type is invalid.

Returns:

Type	Description
Dict	The newly created Variable.

Source code in celonis_api/studio/package.py

def create_variable(
    self, name: str, value: str, variable_type: str = "DATA_MODEL", description: str = ""
) -> typing.Dict:
    """Creates a new Variable in the Package and assigns the value to it.

    !!! api "API"
        - `POST: /package-manager/api/nodes/by-package-key/{package_key}/variables`
            ```json
            {
                "key": name,
                "value": value,
                "type": variable_type,
                "description": description
            }
            ```

    Args:
        name: Name of the Variable.
        value: Value to be set for the Variable.
        variable_type: Type of the Variable. One of [`DATA_MODEL`, `CONNECTION`].
        description: Description for the Variable.

    Raises:
        PyCelonisValueError: If `variable_type` is invalid.

    Returns:
        The newly created Variable.
    """
    if variable_type.upper() not in self.VARIABLE_TYPES:
        raise PyCelonisValueError(f"Argument 'variable_type' must be one of {', '.join(self.VARIABLE_TYPES)} .")

    # Update variable assignments
    payload = {'key': name, 'value': value, 'type': variable_type, 'description': description}
    return self.celonis.api_request(self.variable_assignments_url, payload, method=HttpMethod.POST)

delete(self)

Deletes the Package.

API

• DELETE: /package-manager/api/packages/{self.id}

Source code in celonis_api/studio/package.py

def delete(self):
    """Deletes the Package.

    !!! api "API"
        - `DELETE: /package-manager/api/packages/{self.id}`
    """
    payload = {
        "id": self.id,
        "name": self.name,
    }
    url = f"{self.celonis.url}/package-manager/api/packages/delete/{self.id}"
    self.celonis.api_request(url, payload, method=HttpMethod.POST)

get_node_by_key(self, node_key)

Get a Node of the Package by Key.

API

• GET: /package-manager/api/nodes/by-root-id/{package_id}

Parameters:

Name	Type	Description	Default
node_key	str	Key of the node.	required

Exceptions:

Type	Description
PyCelonisNotFoundError	If Node was not found.

Returns:

Type	Description
BaseNode	The specific Node by Key.

Source code in celonis_api/studio/package.py

def get_node_by_key(self, node_key: str) -> BaseNode:
    """Get a Node of the Package by Key.

    !!! api "API"
        - `GET: /package-manager/api/nodes/by-root-id/{package_id}`

    Args:
        node_key: Key of the node.

    Raises:
        PyCelonisNotFoundError: If Node was not found.

    Returns:
        The specific Node by Key.
    """
    response = self.celonis.api_request(self.package_nodes_url)
    result = CelonisCollection(
        [
            NodeFactory.create_node_from_data(self, self.celonis, data)
            for data in response
            if data["key"] == node_key
        ]
    )
    if len(result) == 0:
        raise PyCelonisNotFoundError(f"No node with 'id': {node_key} found in package.")

    return result[0]

publish(self, version=None, node_ids_to_exclude=None)

Publishes current package using new version

Parameters

version : str, optional New version number following semantic versioning format (X.X.X). If left blank the currently active version number will be used and increased by one (until a nonexistent version number is found). node_ids_to_exclude : list of str List of ids of nodes that will be excluded from publishing

Source code in celonis_api/studio/package.py

def publish(self, version: str = None, node_ids_to_exclude: typing.List[str] = None):
    """Publishes current package using new version

    Parameters
    ----------
    version : str, optional
        New version number following semantic versioning format (X.X.X). If left blank the currently active version
        number will be used and increased by one (until a nonexistent version number is found).
    node_ids_to_exclude : list of str
        List of ids of nodes that will be excluded from publishing
    """
    if not version:
        version = self._get_next_version()

    node_ids_to_exclude = node_ids_to_exclude or []
    message = {"version": version, "nodeIdsToExclude": node_ids_to_exclude, "packageKey": self.data["key"]}
    self.celonis.api_request(self.publish_url, message=message, method=HttpMethod.POST)
    self._logger.info(f"Successfully published version {version} of package.")