celonis.py

`Celonis` ¶

Base object to interact with Celonis Execution Management System (EMS) API.

This class is used to configure the connection to Celonis. The basic idea of this class is the integration with Celonis resource classes, e.g. 'Datamodel'. Each resource class requires an instance of Celonis to be passed to the initializer.

Parameters:

Name	Type	Description	Default
`url`	`str`	Celonis base URL.	required
`api_token`	`str`	Celonis API token.	required
`key_type`	`Union[str, KeyType]`	KeyType of API token. One of [`APP_KEY`, `USER_KEY`] or celonis_api.utils.KeyType.	required
`read_only`	`bool`	If True only GET methods are allowed, set to False to enable PUT, POST and DELETE methods.	required
`verify_ssl`	`bool`	Requiring requests to verify the TLS certificate at the remote end. For more details see `requests.Session`.	required
`user_agent`	`str`	Session header value for `User-Agent`.	required
`timeout`	`Optional[int]`	How long to wait for the server to send data before giving up (in seconds, 300 for data push/pull, max 600). For more details see `requests.Session.request`.	`120`
`total_retry`	`Optional[int]`	Total number of retries (by default disabled, max. 10). For more details see `urllib3.Retry`.	`0`
`backoff_factor`	`Optional[float]`	Factor to apply between retry attempts after the second try (by default disabled). `[0 < backoff_factor < 5]` : (`{backoff_factor} * (2 ** ({total_retry} - 1))`). For more details see `urllib3.Retry`.	`0`
`cookie`	`Optional[str]`	Session header value for `x-celonis-api-scope`.	`None`
`connect`	`Optional[bool]`	If True connects to Celonis on initialization and provides permission information.	`True`

`url: str` `property` `readonly` ¶

Celonis base URL.

`read_only: bool` `property` `readonly` ¶

If True only GET methods are allowed, set to False to enable PUT, POST and DELETE methods.

`timeout: int` `property` `readonly` ¶

How long to wait for the server to send data before giving up (in seconds).

`permissions: Dict` `property` `readonly` ¶

Dictionary listing permissions per Celonis Resource.

`pools: CelonisCollection[Pool]` `property` `readonly` ¶

Get all Pool.

API

GET: /integration/api/pools
GET: /integration-hybrid/api/pools (hybrid)

Returns:

Type	Description
`CelonisCollection[Pool]`	Collection of Pools.

`datamodels: CelonisCollection[Datamodel]` `property` `readonly` ¶

Get all data models by iterating over all Pool and collecting all Datamodels.

Returns:

Type	Description
`CelonisCollection[Datamodel]`	Collection of Datamodels.

`analyses: CelonisCollection[Analysis]` `property` `readonly` ¶

Get all Analysis.

API

GET: /process-mining/api/analysis

Returns:

Type	Description
`CelonisCollection[Analysis]`	Collection of Analyses.

`workspaces: CelonisCollection[Workspace]` `property` `readonly` ¶

Get all Workspaces.

API

GET: /process-mining/api/processes

Returns:

Type	Description
`CelonisCollection[Workspace]`	Collection of Workspaces.

`default_space: Space` `property` `readonly` ¶

Get the Studio Default Space.

Returns:

Type	Description
`Space`	Default Space.

`spaces: CelonisCollection[Space]` `property` `readonly` ¶

Get all Studio Spaces.

API

GET: /package-manager/api/spaces

Returns:

Type	Description
`CelonisCollection[Space]`	Collection of Spaces.

`packages: CelonisCollection[Package]` `property` `readonly` ¶

Get all Studio Packages by iterating over all Spaces and collecting all Packages.

Warning

Use space.find("space_name").packages.find("package_name") for better performance.

API

GET: /package-manager/api/nodes/tree

Returns:

Type	Description
`CelonisCollection[Package]`	Collection of Studio Packages.

`storage_buckets: CelonisCollection[Bucket]` `property` `readonly` ¶

Get all Storage Buckets from file storage manager.

API

GET: /storage-manager/api/buckets

Returns:

Type	Description
`CelonisCollection[Bucket]`	Collection of Bucket.

`create_pool(self, name, safety_check=True)` ¶

Creates a new Pool.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the Pool.	required
`safety_check`	`Optional[bool]`	Weather to check if pool already exits or not.	`True`

Returns:

Type	Description
`Pool`	Instance of the newly created Pool object.

`get_datamodel(self, datamodel_id)` ¶

Get a datamodel by ID. Much faster than datamodels.find().

Parameters:

Name	Type	Description	Default
`datamodel_id`	`str`	Id of the Datamodel.	required

Returns:

Type	Description
`Datamodel`	Datamodel by ID.

`create_datamodel(self, name, pool)` ¶

Creates a new Datamodel.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the Datamodel.	required
`pool`	`Union[str, pycelonis.celonis_api.event_collection.data_pool.Pool]`	Name or ID of the Pool the new Datamodel should be linked to.	required

Returns:

Type	Description
`Datamodel`	The newly created Datamodel object.

`create_analysis(self, name, workspace, **kwargs)` ¶

Creates a new Analysis.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the analysis.	required
`workspace`	`Workspace`	Workspace object the new Analysis should be linked to.	required
`**kwargs`		Additional arguments which are passed to celonis_api.process_analytics.workspace.Workspace.create_analysis	`{}`

Returns:

Type	Description
`Analysis`	The newly created Analysis object.

`create_workspace(self, name, datamodel)` ¶

Creates a new Workspace.

Parameters:

Name	Type	Description	Default
`name`		Name of the Workspace.	required
`datamodel`		Name or ID of the Datamodel the Workspace should be linked to.	required

Returns:

Type	Description
`Workspace`	The newly created Workspace object.

`create_space(self, name, icon_reference='earth')` ¶

Creates a new Studio Space.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the Space.	required
`icon_reference`	`str`	Name of the icon to use for the Space. One of: `earth`, `binocular`, `box-archive`, `building`, `building-2`, `catalog`, `earth-2`, `factory`, `finance`, `horn`, `inventory`, `logistics`, `store`, `suitcase`, `wallet`.	`'earth'`

Returns:

Type	Description
`Space`	The newly created Space object.

`create_package(self, name, key=None, package_type='APP', space_id=None)` ¶

Creates a new Studio Package.

Warning

This method is deprecated. Use celonis_api.studio.space.Space.create_package:

space.create_package("package_name", "package_key", "APP")

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the Package.	required
`key`	`str`	Key of the Package, if left blank defaults to the name of Package.	`None`
`package_type`	`str`	Package type. Can be one of "APP", "LIBRARY", "INSTRUMENT".	`'APP'`
`space_id`	`str`	ID of Space where package will be created. If None will use the 'Default' Space.	`None`

Returns:

Type	Description
`Package`	The newly created Studio Package object.

`get_knowledge_model_by_full_key(self, knowledge_model_key)` ¶

Get a Knowledge Model by key (PACKAGE_KEY.KNOWLEDGE_MODEL_KEY)

Parameters:

Name	Type	Description	Default
`knowledge_model_key`	`str`	Full Knowledge Model key consisting of "PACKAGE_KEY.KNOWLEDGE_MODEL_KEY"	required

Returns:

Type	Description
`KnowledgeModel`	The KnowledgeModel.

`create_storage_bucket(self, name, features=None)` ¶

Creates a new Bucket in the file storage manager.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the Bucket.	required
`features`	`List[str]`	Features used for the Bucket. Currently only supports SFTP.	`None`

Returns:

Type	Description
`Bucket`	The newly created Bucket object.

`api_request(self, url, message=None, method='auto', timeout='default', get_json=True, **extra)` ¶

Wrapper method for requests.Session.request. The session object is created and configured during Celonis class initialization.

Parameters:

Name	Type	Description	Default
`url`	`str`	Absolut URL pointing to a Celonis resource.	required
`message`	`Union[str, Dict, pathlib.Path]`	If no message is given, performs `GET` request. If message is `"DELETE"` performs `DELETE` request. If message is `pathlib.Path`: If the path refers to an existing file performs `POST` request with file attached. Otherwise performs `GET` request and streams payload into the path. If message is a `list`, `dict` or `str`, performs `POST` request with message as JSON payload.	`None`
`method`	`str`	Can be used to override `auto` detection of request method. One of [`auto`, `get`, `put`, `post`, `delete`]	`'auto'`
`timeout`	`Union[str, int]`	How long to wait for the server to send data before giving up (in seconds, max. 600): `default` = 120 `connect` = 6 In the case of a file transfer, the timeout must be 300 or higher. If not set defaults to 300.	`'default'`
`get_json`	`bool`	If True parses requires the response to be in JSON format and converts to dictionary.	`True`
`**extra`		Additional arguments which are passed to `requests.Session.request`.	`{}`

Exceptions:

Type	Description
`PyCelonisPermissionError`	When the session is read-only or the API key has not sufficient permissions.
`PyCelonisHTTPError`	When timeout occurs or unexpected HTTP error.
`PyCelonisValueError`	When timeout is invalid.

Returns:

Type	Description
`dict \| list \| str \| pathlib.Path`	Celonis resource response.

`from_url(url, api_token, key_type, read_only, verify_ssl, user_agent, timeout=None, total_retry=None, backoff_factor=None, cookie=None, connect=True)` `classmethod` ¶

Convenient Celonis class method to interact directly with a the specified resource by URL.

Parameters:

Name	Type	Description	Default
`url`	`str`	Celonis base URL.	required
`api_token`	`str`	Celonis API token.	required
`key_type`	`Union[str, KeyType]`	KeyType of API token. One of [`APP_KEY`, `USER_KEY`] or celonis_api.utils.KeyType.	required
`read_only`	`bool`	If True only GET methods are allowed, set to False to enable PUT, POST and DELETE methods.	required
`verify_ssl`	`bool`	Requiring requests to verify the TLS certificate at the remote end. For more details see `requests.Session`.	required
`user_agent`	`str`	Session header value for `User-Agent`.	required
`timeout`	`Optional[int]`	How long to wait for the server to send data before giving up (in seconds, 300 for data push/pull, max 600). For more details see `requests.Session.request`.	`None`
`total_retry`	`Optional[int]`	Total number of retries (by default disabled, max. 10). For more details see `urllib3.Retry`.	`None`
`backoff_factor`	`Optional[float]`	Factor to apply between retry attempts after the second try (by default disabled). `[0 < backoff_factor < 5]` : (`{backoff_factor} * (2 ** ({total_retry} - 1))`). For more details see `urllib3.Retry`.	`None`
`cookie`	`Optional[str]`	Session header value for `x-celonis-api-scope`.	`None`
`connect`	`Optional[bool]`	If True connects to Celonis on initialization and provides permission information.	`True`

celonis.py

Celonis ¶

url: str property readonly ¶

read_only: bool property readonly ¶

timeout: int property readonly ¶

permissions: Dict property readonly ¶

pools: CelonisCollection[Pool] property readonly ¶

datamodels: CelonisCollection[Datamodel] property readonly ¶

analyses: CelonisCollection[Analysis] property readonly ¶

workspaces: CelonisCollection[Workspace] property readonly ¶

default_space: Space property readonly ¶

spaces: CelonisCollection[Space] property readonly ¶

packages: CelonisCollection[Package] property readonly ¶

storage_buckets: CelonisCollection[Bucket] property readonly ¶

create_pool(self, name, safety_check=True) ¶

get_datamodel(self, datamodel_id) ¶

create_datamodel(self, name, pool) ¶

create_analysis(self, name, workspace, **kwargs) ¶

create_workspace(self, name, datamodel) ¶

create_space(self, name, icon_reference='earth') ¶

create_package(self, name, key=None, package_type='APP', space_id=None) ¶

get_knowledge_model_by_full_key(self, knowledge_model_key) ¶

create_storage_bucket(self, name, features=None) ¶

api_request(self, url, message=None, method='auto', timeout='default', get_json=True, **extra) ¶

from_url(url, api_token, key_type, read_only, verify_ssl, user_agent, timeout=None, total_retry=None, backoff_factor=None, cookie=None, connect=True) classmethod ¶

`Celonis` ¶

`url: str` `property` `readonly` ¶

`read_only: bool` `property` `readonly` ¶

`timeout: int` `property` `readonly` ¶

`permissions: Dict` `property` `readonly` ¶

`pools: CelonisCollection[Pool]` `property` `readonly` ¶

`datamodels: CelonisCollection[Datamodel]` `property` `readonly` ¶

`analyses: CelonisCollection[Analysis]` `property` `readonly` ¶

`workspaces: CelonisCollection[Workspace]` `property` `readonly` ¶

`default_space: Space` `property` `readonly` ¶

`spaces: CelonisCollection[Space]` `property` `readonly` ¶

`packages: CelonisCollection[Package]` `property` `readonly` ¶

`storage_buckets: CelonisCollection[Bucket]` `property` `readonly` ¶

`create_pool(self, name, safety_check=True)` ¶

`get_datamodel(self, datamodel_id)` ¶

`create_datamodel(self, name, pool)` ¶

`create_analysis(self, name, workspace, **kwargs)` ¶

`create_workspace(self, name, datamodel)` ¶

`create_space(self, name, icon_reference='earth')` ¶

`create_package(self, name, key=None, package_type='APP', space_id=None)` ¶

`get_knowledge_model_by_full_key(self, knowledge_model_key)` ¶

`create_storage_bucket(self, name, features=None)` ¶

`api_request(self, url, message=None, method='auto', timeout='default', get_json=True, **extra)` ¶

`from_url(url, api_token, key_type, read_only, verify_ssl, user_agent, timeout=None, total_retry=None, backoff_factor=None, cookie=None, connect=True)` `classmethod` ¶