workspace.py

Workspace (CelonisApiObject)

Workspace object to interact with Celonis Process Analytics API.

Source code in celonis_api/process_analytics/workspace.py

class Workspace(CelonisApiObject):
    """Workspace object to interact with Celonis Process Analytics API."""

    @property
    def url(self) -> str:
        """
        !!! api "API"
            - `/process-mining/api/processes/{workspace_id}`
        """
        return f"{self.celonis.url}/process-mining/api/processes/{self.id}"

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

        !!! api "API"
            - `GET: /process-mining/api/analysis/{analysis_id}?processId={workspace_id}`

        Returns:
            Collection of Analyses in the Workspace.
        """
        params = {"processId": self.id}
        response = self.celonis.api_request(f"{self.celonis.url}/process-mining/api/analysis", params=params)
        return CelonisCollection([Analysis(self.celonis, data) for data in response])

    def create_analysis(self, name: str = None, content_backup: typing.Union[str, pathlib.Path] = None) -> 'Analysis':
        """Creates a new Analysis in the Workspace

        Args:
            name: Name of the Analysis.
            content_backup: Path to an Analysis Backup that will be restored to this Analysis.

        Returns:
            The newly created Analysis object.
        """
        if name is None and content_backup is None:
            raise PyCelonisValueError("At least one of name and content_backup must not be None.")
        elif name is None:
            name = "tempname"
            keep_analysis_name = False
        else:
            keep_analysis_name = True

        payload = {"name": name, "processId": self.id, "applicationId": None}
        response = self.celonis.api_request(f"{self.celonis.url}/process-mining/api/analysis", payload)
        analysis = Analysis(self.celonis, response)
        if content_backup:
            analysis.rebuild_content_from_backup(content_backup, keep_analysis_name)

        return analysis

    @property
    def pool(self) -> 'Pool':
        """Get the Pool the Workspace is connected to."""
        return data_pool.Pool(self.celonis, self.data.get("poolId"))

    @property
    def datamodel(self) -> 'Datamodel':
        """Get the Datamodel the Workspace is connected to.

        Raises:
            PyCelonisNotFoundError: If the Datamodel was not found or needs to reloaded.
        """
        dm_id = self.data.get("dataModelId")
        if dm_id:
            return self.celonis.get_datamodel(dm_id)
        else:
            raise PyCelonisNotFoundError(
                "Datamodel of workspace can't be found! Reloading the datamodel manually might help."
            )

    def move(self, to: str, target_workspace: 'Workspace' = None):
        """Moves the Workspace to another team.

        !!! api "API"
            - `POST: /process-mining/api/processes/move`
                ```json
                {
                    "analysisIdsToReplace": [],
                    "sourceWorkspaceId": self.id,
                    "targetWorkspaceId": target_workspace,
                    "teamDomain": to
                }
                ```

        Args:
            to: Team name (e.g. `move` for https://move.eu-1.celonis.cloud)
            target_workspace: Workspace ID or Workspace object where the analysis should be copied to.
        """
        if isinstance(target_workspace, Workspace):
            target_workspace = target_workspace.id

        payload = {
            "analysisIdsToReplace": [],
            "sourceWorkspaceId": self.id,
            "targetWorkspaceId": target_workspace,
            "teamDomain": to,
        }
        self.celonis.api_request(f"{self.celonis.url}/process-mining/api/processes/move", payload)

analyses: CelonisCollection[Analysis] property readonly

Get all Analyses in the Workspace.

API

• GET: /process-mining/api/analysis/{analysis_id}?processId={workspace_id}

Returns:

Type	Description
CelonisCollection[Analysis]	Collection of Analyses in the Workspace.

datamodel: Datamodel property readonly

Get the Datamodel the Workspace is connected to.

Exceptions:

Type	Description
PyCelonisNotFoundError	If the Datamodel was not found or needs to reloaded.

pool: Pool property readonly

Get the Pool the Workspace is connected to.

url: str property readonly

API

• /process-mining/api/processes/{workspace_id}

create_analysis(self, name=None, content_backup=None)

Creates a new Analysis in the Workspace

Parameters:

Name	Type	Description	Default
name	str	Name of the Analysis.	None
content_backup	Union[str, pathlib.Path]	Path to an Analysis Backup that will be restored to this Analysis.	None

Returns:

Type	Description
Analysis	The newly created Analysis object.

Source code in celonis_api/process_analytics/workspace.py

def create_analysis(self, name: str = None, content_backup: typing.Union[str, pathlib.Path] = None) -> 'Analysis':
    """Creates a new Analysis in the Workspace

    Args:
        name: Name of the Analysis.
        content_backup: Path to an Analysis Backup that will be restored to this Analysis.

    Returns:
        The newly created Analysis object.
    """
    if name is None and content_backup is None:
        raise PyCelonisValueError("At least one of name and content_backup must not be None.")
    elif name is None:
        name = "tempname"
        keep_analysis_name = False
    else:
        keep_analysis_name = True

    payload = {"name": name, "processId": self.id, "applicationId": None}
    response = self.celonis.api_request(f"{self.celonis.url}/process-mining/api/analysis", payload)
    analysis = Analysis(self.celonis, response)
    if content_backup:
        analysis.rebuild_content_from_backup(content_backup, keep_analysis_name)

    return analysis

move(self, to, target_workspace=None)

Moves the Workspace to another team.

API

• POST: /process-mining/api/processes/move

  {
      "analysisIdsToReplace": [],
      "sourceWorkspaceId": self.id,
      "targetWorkspaceId": target_workspace,
      "teamDomain": to
  }
  

Parameters:

Name	Type	Description	Default
to	str	Team name (e.g. move for https://move.eu-1.celonis.cloud)	required
target_workspace	Workspace	Workspace ID or Workspace object where the analysis should be copied to.	None

Source code in celonis_api/process_analytics/workspace.py

def move(self, to: str, target_workspace: 'Workspace' = None):
    """Moves the Workspace to another team.

    !!! api "API"
        - `POST: /process-mining/api/processes/move`
            ```json
            {
                "analysisIdsToReplace": [],
                "sourceWorkspaceId": self.id,
                "targetWorkspaceId": target_workspace,
                "teamDomain": to
            }
            ```

    Args:
        to: Team name (e.g. `move` for https://move.eu-1.celonis.cloud)
        target_workspace: Workspace ID or Workspace object where the analysis should be copied to.
    """
    if isinstance(target_workspace, Workspace):
        target_workspace = target_workspace.id

    payload = {
        "analysisIdsToReplace": [],
        "sourceWorkspaceId": self.id,
        "targetWorkspaceId": target_workspace,
        "teamDomain": to,
    }
    self.celonis.api_request(f"{self.celonis.url}/process-mining/api/processes/move", payload)