QPR ProcessAnalyzer API

Introduction

QPR ProcessAnalyzer API can be used to create integrations to other applications and automate operations in the process mining environment.

QPR ProcessAnalyzer API is a JSON-based API following the REST design principles. Most of the endpoints require a prior login to establish a session. The session is initialized with the token call with username and password, and the access token is returned as a response for a successful login. The endpoints requiring prior authenticated session, need to have a HTTP request header Authorization with value Bearer <access token> to identify the session.

Url for calling the API has the following form (replace the server hostname with a correct one):

https://customer.onqpr.com/qprpa/api/<endpointName>

Available endpoints

Following endpoints are available:

Endpoint	Description
token	Login user using username and password and get a session token as a response.
api/signout	Logs out a user session.
api/expression	Runs an expression.
api/expression/query	Runs query written using the expression language and returns result data as response.
api/filters	Get filters for all models or filters for a single model.
api/serverinfo	Returns common system information needed by UI, such as the default UI language and in whether SSO has been configured.
api/importfile	Import data into datatable from .csv, .xes or .pacm file.
api/usersettings	Save user specific settings to the server.
api/operations/terminate	Stops the defined tasks (by the task id) to save computing resources.
api/analysis/cancel	Stops currently running tasks (by the task identifier) to save computing resources.
Saml2/Acs	Identity provider (IdP) will send the SAML 2.0 assertion to this endpoint which will respond with 302 to redirect to QPR ProcessAnalyzer UI.
Saml2	Returns the SAML 2.0 service provider (SP) metadata, if SAML 2.0 authentication has been configured.

In addition, there are endpoints for

• moving and deleting workspace elements
• projects
• dashboards
• models
• datatables
• scripts
• users, groups and roles

Usage examples

Run query to fetch data

Following Python function performs a query to fetch data from QPR ProcessAnalyzer by calling the REST API. It performs following steps: (1) login to QPR ProcessAnalyzer, (2) run the query, (3) write fetched data to a file, and (4) log out.

The query is in the "json" parameter of the /api/expression/query request. For example, query for a chart can be found in the chart settings Advanced tab by clicking the Query button.

def runQuery(serverUrl: str, username: str, password: str):
  import requests
  import json
  loginData = {
    "grant_type": "password",
    "username": username,
    "password": password
  }
  loginResponse = requests.post(
    url = serverUrl + "/token",
    data = loginData
  )
  loginResponse.raise_for_status()
  sessionToken = loginResponse.json().get("access_token")
  
  queryResponse = requests.post(
    url = serverUrl + "/api/expression/query",
    headers = {
      "Authorization": "Bearer " + sessionToken
    },
    json = {
      "ProcessingMethod": "dataframe",
      "ContextType": "model",
      "ModelId": 1,
      "Root": "Cases",
      "MaximumRowCount": 10,
      "Dimensions": None,
      "Values": [
        {
          "Name": "CaseId",
          "Expression": "CaseId"
        },
        {
          "Name": "StartTime",
          "Expression": "AggregateFrom(Events, \"Min\", TimeStamp)"
        },
        {
          "Name": "EndTime",
          "Expression": "AggregateFrom(Events, \"Max\", TimeStamp)"
        },
        {
          "Name": "EventCount",
          "Expression": "AggregateFrom(Events, \"Count\")"
        },
        {
          "Name": "EventTypeCount",
          "Expression": "AggregateFrom(Events, \"CountDistinct\", EventType)"
        }
      ],
      "Ordering": [
        {
          "Name": "CaseId",
          "Direction": "Ascending"
        }
      ]
    }
  )
  queryResponse.raise_for_status()
    
  with open("QueriedData.json", "w") as f:
    json.dump(queryResponse.json(), f)

  logOutResponse = requests.post(
    url = serverUrl + "/api/signout",
    headers = {
      "Authorization": "Bearer " + sessionToken,
      "Content-type": "application/json"
    }
  )
  logOutResponse.raise_for_status()

The function can be called as follows:

runQuery(
  serverUrl = "https://server.onqpr.com/qprpa",
  username = "qpr",
  password = "demo"
)

Trigger script run

Following Python function starts a script in QPR ProcessAnalyzer by calling the REST API. It performs following steps: (1) login to QPR ProcessAnalyzer, (2) start the script, and (3) log out. The call just starts the script without waiting for it to complete (asynchronous behavior).

def startQprProcessAnalyzerScript(serverUrl: str, username: str, password: str, scriptId: int):
  import requests
  loginData = {
    "grant_type": "password",
    "username": username,
    "password": password
  }
  loginResponse = requests.post(
    url = serverUrl + "/token",
    data = loginData
  )
  loginResponse.raise_for_status()
  sessionToken = loginResponse.json().get("access_token")
  
  startScriptResponse = requests.post(
    url = serverUrl + "/api/scripts/run/" + str(scriptId),
    headers = {
      "Authorization": "Bearer " + sessionToken,
      "Content-type": "application/json"
    }
  )
  startScriptResponse.raise_for_status()

  logOutResponse = requests.post(
    url = serverUrl + "/api/signout",
    headers = {
      "Authorization": "Bearer " + sessionToken,
      "Content-type": "application/json"
    }
  )
  logOutResponse.raise_for_status()

The function can be called as follows:

startQprProcessAnalyzerScript(
  serverUrl = "https://server.onqpr.com/qprpa",
  username = "qpr",
  password = "demo",
  scriptId = 1
)

The script id can be found in the scripts list in the Workspace.

Synchronize users and groups

Following script written in Python updates QPR ProcessAnalyzer users and groups based on the provided dataset. This script can be extended to fetch the user data from an external source (e.g., Azure AD) to implement a complete user management integration between the systems.

This script performs following steps:

1. Read the provided dataset and store it to in-memory structures.
2. Read all users from QPR ProcessAnalyzer (including which groups the users belong to). (POST /api/expression/query)
3. Read all groups from QPR ProcessAnalyzer. (POST /api/expression/query)
4. Determine the gap between the current state in the user management and the provided dataset.
5. Create new users appearing in the dataset to QPR ProcessAnalyzer. (POST /api/users)
6. Inactivate non-existing users in the dataset from QPR ProcessAnalyzer. (PUT /api/users)
7. Activate existing inactive users in QPR ProcessAnalyzer that exist in the dataset. (PUT /api/users)
8. Add users to groups and remove from groups based on the determined gap in the state. (PUT/DELETE /api/users/memberships)

Python script code:

def synchronizeUsersAndGroups(serverUrl: str, username: str, password: str, userData: list, userGroupData: list):
    """
    Synchronizes QPR ProcessAnalyzer users and groups based on the provided dataset.
    
    Parameters:
    - serverUrl: QPR ProcessAnalyzer server URL
    - username: Admin username for authentication
    - password: Admin password for authentication
    - userData: List of dicts with keys "Name" and "Groups" (list of group names)
    - userGroupData: List of group names that should exist
    """
    import requests
    import json

    # Step 1: Login to QPR ProcessAnalyzer
    loginData = {
        "grant_type": "password",
        "username": username,
        "password": password
    }
    loginResponse = requests.post(
        url=serverUrl + "/token",
        data=loginData
    )
    loginResponse.raise_for_status()
    sessionToken = loginResponse.json().get("access_token")

    headers = {
        "Authorization": "Bearer " + sessionToken,
        "Content-type": "application/json"
    }

    # Step 2: Read all users from QPR ProcessAnalyzer (including group memberships)
    usersQueryResponse = requests.post(
        url=serverUrl + "/api/expression/query",
        headers=headers,
        json={
            "Dimensions": None,
            "Values": [
                {
                    "Name": "Id",
                    "Expression": "Id"
                },
                {
                    "Name": "Name",
                    "Expression": "Name"
                },
                {
                    "Name": "IsActive",
                    "Expression": "IsActive"
                },
                {
                    "Name": "Groups",
                    "Expression": "ToJson(OrderByValue(Groups.Id))"
                }
            ],
            "Root": "Users",
            "ContextType": "generic"
        }
    )
    usersQueryResponse.raise_for_status()
    existingUsers = usersQueryResponse.json()

    # Step 3: Read all groups from QPR ProcessAnalyzer
    groupsQueryResponse = requests.post(
        url=serverUrl + "/api/expression/query",
        headers=headers,
        json={
            "Dimensions": None,
            "Values": [
                {
                    "Name": "Id",
                    "Expression": "Id"
                },
                {
                    "Name": "Name",
                    "Expression": "Name"
                }
            ],
            "Root": "UserGroups",
            "ContextType": "generic"
        }
    )
    groupsQueryResponse.raise_for_status()
    existingGroups = groupsQueryResponse.json()

    # Build lookup dictionaries
    existingUsersByName = {}
    for user in existingUsers:
        existingUsersByName[user["Name"]] = user

    existingGroupsByName = {}
    for group in existingGroups:
        existingGroupsByName[group["Name"]] = group

    # Build set of desired usernames from the provided dataset
    desiredUserNames = set()
    for user in userData:
        desiredUserNames.add(user["Name"])

    # Build desired group memberships: username -> set of group names
    desiredMemberships = {}
    for user in userData:
        desiredMemberships[user["Name"]] = set(user.get("Groups", []))

    # Step 4: Determine the gap between current state and desired state

    existingUserNames = set(existingUsersByName.keys())
    usersToCreate = desiredUserNames - existingUserNames
    usersToInactivate = existingUserNames - desiredUserNames
    usersToActivate = set()

    for userName in desiredUserNames:
        if userName in existingUsersByName:
            user = existingUsersByName[userName]
            if not user["IsActive"]:
                usersToActivate.add(userName)

    # Step 5: Create new users appearing in the dataset
    for userName in usersToCreate:
        createUserResponse = requests.post(
            url=serverUrl + "/api/users",
            headers=headers,
            json={
                "Name": userName,
                "IsActive": True
            }
        )
        createUserResponse.raise_for_status()
        createdUser = createUserResponse.json()
        # Add the newly created user to our lookup
        existingUsersByName[userName] = {
            "Id": createdUser["Id"],
            "Name": userName,
            "IsActive": True,
            "Groups": "[]"
        }

    # Step 6: Inactivate users not in the dataset
    for userName in usersToInactivate:
        user = existingUsersByName[userName]
        if user["IsActive"]:
            inactivateResponse = requests.put(
                url=serverUrl + "/api/users",
                headers=headers,
                json={
                    "Id": user["Id"],
                    "IsActive": False
                }
            )
            inactivateResponse.raise_for_status()

    # Step 7: Activate existing inactive users that are in the dataset
    for userName in usersToActivate:
        user = existingUsersByName[userName]
        activateResponse = requests.put(
            url=serverUrl + "/api/users",
            headers=headers,
            json={
                "Id": user["Id"],
                "IsActive": True
            }
        )
        activateResponse.raise_for_status()

    # Step 8: Add users to groups and remove from groups based on the gap
    for userName in desiredUserNames:
        user = existingUsersByName[userName]
        userId = user["Id"]

        # Parse current group IDs from the user's Groups field
        currentGroupIds = set()
        groupsJson = user.get("Groups", "[]")
        if groupsJson:
            currentGroupIds = set(json.loads(groupsJson))

        # Determine desired group IDs for this user
        desiredGroupIds = set()
        for groupName in desiredMemberships.get(userName, []):
            if groupName in existingGroupsByName:
                desiredGroupIds.add(existingGroupsByName[groupName]["Id"])

        # Groups to add the user to
        groupsToAdd = desiredGroupIds - currentGroupIds
        # Groups to remove the user from
        groupsToRemove = currentGroupIds - desiredGroupIds

        # Add user to new groups
        for groupId in groupsToAdd:
            addMembershipResponse = requests.put(
                url=serverUrl + "/api/users/memberships",
                headers=headers,
                json={
                    "GroupId": groupId,
                    "MemberId": userId,
                    "RoleName": "Member"
                }
            )
            addMembershipResponse.raise_for_status()

        # Remove user from groups
        for groupId in groupsToRemove:
            removeMembershipResponse = requests.delete(
                url=serverUrl + "/api/users/memberships",
                headers=headers,
                json={
                    "GroupId": groupId,
                    "MemberId": userId,
                    "RoleName": "Member"
                }
            )
            removeMembershipResponse.raise_for_status()

    # Log out
    logOutResponse = requests.post(
        url=serverUrl + "/api/signout",
        headers=headers
    )
    logOutResponse.raise_for_status()

The script can be called as follows:

synchronizeUsersAndGroups(
    serverUrl="https://server.onqpr.com/qprpa",
    username="qpr",
    password="demo",
    userData=[
        {"Name": "user1@company.com", "Groups": ["Analysts", "Managers"]},
        {"Name": "user2@company.com", "Groups": ["Analysts"]},
        {"Name": "user3@company.com", "Groups": ["Managers", "Admins"]}
    ],
    userGroupData=["Analysts", "Managers", "Admins"]
)