CASE WHEN behavior change¶
About the change
The behavior of the CASE WHEN operator in PQL will be changed in regards to the behavior when a condition evaluates to NULL. In the future, a condition that evaluates to NULL is always treated as if it was false. The new behavior is aligned to the expectations of most users and is already desired in most cases.
Find apotentially affected CASE WHEN statements
This script can be used to find potentially affected CASE WHEN queries in your EMS team. It can search Analyses in Process Analytics, Analyses in Studio, as well as Knowledge Models for those CASE WHEN queries.
STEP 1: Login¶
The following snippet connects to Celonis by using the application key of the ML workbench. Please make sure that this application key has at least the following permissions.
Required Permissions¶
For check in Process Analytics¶
In order to check for potentially affected CASE WHEN statements in Process Analytics, assign the following permissions to the ML workbench application key under Team Settings > Permissions > Process Analytics Permissions
:
- USE ALL ANALYSES
- EDIT ALL ANALYSES
For check in Studio Assets¶
In order to check for potentially affected CASE WHEN statements in Studio Assets (Analyses and Knowledge Models), assign the following permissions to the ML workbench application key for every Studio Space that should be checked under the <Three-dot menue of the Space> > Permissions
:
- USE ALL PACKAGES
- EDIT ALL PACKAGES
In both cases, the EDIT ALL
permission is required to access the unpublished versions of Analyses. This script is read-only and will not modify any EMS application.
from pycelonis import get_celonis
from pycelonis.celonis_api import api_utils
from pycelonis.celonis_api.studio.analysis import Analysis as StudioAnalysis
from pycelonis.celonis_api.studio.knowledge_model import KnowledgeModel
import pycelonis.celonis_api.studio.node_factory as node_factory
import re
import pandas as pd
celonis = get_celonis(read_only=True)
STEP 2: Helper functions¶
These are helper functions. Simply execute the cell below. It does not do anything (yet). Please do not modify the code.
class ProcessAnalyticsResult:
def __init__(self, workspace, analysis, sheet, component, location, query):
self.workspace=workspace
self.analysis=analysis
self.sheet=sheet
self.component=component
self.location=location
self.query=query
class StudioAnalysisResult:
def __init__(self, space, package, analysis, sheet, component, location, query):
self.space=space
self.package=package
self.analysis=analysis
self.sheet=sheet
self.component=component
self.location=location
self.query=query
class KnowledgeModelResult:
def __init__(self, space, package, knowledge_model, location, id, query):
self.space=space
self.package=package
self.knowledge_model=knowledge_model
self.location=location
self.id=id
self.query=query
def result_to_csv(result: list, file_name:str):
df = pd.DataFrame([r.__dict__ for r in result ])
df["query"] = df["query"].apply( lambda x: x.replace("\n", " ") )
df.to_csv(path_or_buf=file_name, sep=",")
print("Wrote potentially affected CASE WHEN statements to file {}".format(file_name))
def has_case_when(pql: str):
return re.search(r"\s+ELSE(?!\s+NULL\s+END)", str(pql), re.IGNORECASE)
def create_analysis_result(analysis, sheet, component, location, query, space=None, package=None):
if package:
# studio analysis
return StudioAnalysisResult(space.name, package.name, analysis.name, sheet, component,location, query)
else:
# process analytics
return ProcessAnalyticsResult(analysis.workspace.name, analysis.name, sheet, component,location, query)
def find_in_analysis(analysis, space=None, package=None):
affected_case_whens=[]
for saved_formula in analysis.saved_formulas:
if (has_case_when(saved_formula.data["template"])):
affected_case_whens.append(create_analysis_result(analysis, "", "", "Saved formula {}".format(saved_formula.data["name"]), saved_formula.data["template"], space=space, package=package))
document=analysis.draft
if document is None:
return
if document.data["document"] is None:
return
if document.pql_filter is not None:
if has_case_when(document.pql_filter):
affected_case_whens.append(create_analysis_result(analysis, "", "", "Analysis Load Script", document.pql_filter, space=space, package=package))
for variable in document.variables:
if "value" in variable and has_case_when(variable["value"]):
affected_case_whens.append(create_analysis_result(analysis, "", "", "Variable {}".format(variable["name"]), variable["value"], space=space, package=package))
for sheet in document.sheets:
if sheet.pql_filter is not None:
if has_case_when(sheet.pql_filter):
affected_case_whens.append(create_analysis_result(analysis, sheet.name, "", "Sheet Filter", sheet.pql_filter, space=space, package=package))
for component in sheet.components:
if component is not None:
component_name = "Component {}".format(component.name if component.name else "<unnamed>")
if component.pql_filter is not None:
if has_case_when(component.pql_filter):
affected_case_whens.append(create_analysis_result(analysis, sheet.name, component_name, "Component Filter", component.pql_filter, space=space, package=package))
try:
if component.pql_columns is not None:
pql_columns = component.pql_columns
pql_columns = pql_columns if isinstance(pql_columns, list) else [pql_columns]
col_id = 0
for pql_column in pql_columns:
col_id = col_id +1
column_name = "Column {} ({})".format(col_id, pql_column.name if pql_column.name else "<unnamed>")
if hasattr(pql_column, "query"):
if has_case_when(pql_column.query):
affected_case_whens.append(create_analysis_result(analysis, sheet.name, component_name, column_name, pql_column.query, space=space, package=package))
except:
pass
return affected_case_whens
def find_in_process_analytics(analysis):
print("Checking Analysis {} in Workspace {}.".format(analysis.name, analysis.workspace.name))
return find_in_analysis(analysis)
def find_in_studio_analysis(analysis, space, package):
print("Checking Studio Analysis {} in Package {}/{}.".format(analysis.name, space.name, package.name))
return find_in_analysis(analysis, space, package)
def find_in_knowledge_model(knowledge_model, space, package):
print("Checking Studio Knowledge Model {} in Package {}/{}.".format(knowledge_model.name, space.name, package.name))
affected_case_whens=[]
if knowledge_model is not None:
for kpi in knowledge_model.raw_kpis:
if "pql" in kpi and has_case_when(kpi["pql"]):
affected_case_whens.append(KnowledgeModelResult(space.name, package.name, knowledge_model.name, "KPI", kpi["id"], kpi["pql"]))
for record in knowledge_model.raw_records:
if "pql" in record and has_case_when(record["pql"]):
affected_case_whens.append(KnowledgeModelResult(space.name, package.name, knowledge_model.name, "Record", record["id"], record["pql"]))
for filter in knowledge_model.raw_filters:
if "pql" in filter and has_case_when(filter["pql"]):
affected_case_whens.append(KnowledgeModelResult(space.name, package.name, knowledge_model.name, "Filter", filter["id"], filter["pql"]))
return affected_case_whens
def new_knowledge_model(package, celonis, id):
try:
return KnowledgeModel(package, celonis, id)
except:
return None
def get_studio_assets_in_package(package, asset_type):
r = celonis.api_request(package.package_nodes_url, threaded=True)
return [x for x in [
StudioAnalysis(package, celonis, data["id"]) if data["assetType"] == "ANALYSIS" else new_knowledge_model(package, celonis, data["id"]) if data["assetType"] == "SEMANTIC_MODEL" else None
for data in r
if data["nodeType"] == 'ASSET' and data["assetType"] == asset_type
] if x is not None]
def get_studio_knowledge_model_in_package(package, asset_type):
r = celonis.api_request(package.package_nodes_url, threaded=True)
return [
KnowledgeModel(package, celonis, data["id"])
for data in r
if data["nodeType"] == 'ASSET' and data["assetType"] == "SEMANTIC_MODEL"
]
## PROCESS ANALYTICS
def check_process_analytics_analyses():
analyses = celonis.analyses
result = []
for analysis in analyses:
result.extend(find_in_process_analytics(analysis))
if result:
print("{} potentially affected CASE WHEN statements found in Process Analytics.".format(len(result)))
result_to_csv(result, "process_analytics.csv")
else:
print("No potentially affected CASE WHEN statements found in Process Analytics.")
## STUDIO ANALYSES
def check_studio_analyses():
result = []
spaces = celonis.spaces
for space in spaces:
for package in space.packages:
for analysis in get_studio_assets_in_package(package, "ANALYSIS"):
result.extend(find_in_studio_analysis(analysis, space, package))
if result:
print("{} potentially affected CASE WHEN statements found in Studio Analyses.".format(len(result)))
result_to_csv(result, "studio_analyses.csv")
else:
print("No potentially affected CASE WHEN statements found in Studio Analyses.")
## KNOWLEDGE MODEL
def check_knowledge_models():
result=[]
spaces = celonis.spaces
for space in spaces:
for package in space.packages:
for knowledge_model in get_studio_assets_in_package(package, "SEMANTIC_MODEL"):
result.extend(find_in_knowledge_model(knowledge_model, space, package))
if result:
print("{} potentially affected CASE WHEN statements found in Studio Knowledge Models.".format(len(result)))
result_to_csv(result, "studio_knowledge_models.csv")
else:
print("No potentially affected CASE WHEN statements found in Studio Knowledge Models.")
STEP 3: Execute checks¶
STEP 3.1: Check Process Analytics Analyses¶
Execute the following cell if you want to find potentially affected CASE WHEN statements inside Process Analytics Analyses. It will generate a file called process_analytics.csv
with the found CASE WHEN statements and their exact locations.
check_process_analytics_analyses()
STEP 3.2: Check Studio Analyses¶
Execute the following cell if you want to find potentially affected CASE WHEN statements inside Studio Analyses. It will generate a file called studio_analyses.csv
with the found CASE WHEN statements and their exact locations.
check_studio_analyses()
STEP 3.1: Check Knowledge Models¶
Execute the following cell if you want to find potentially affected CASE WHEN statements inside Studio Knowledge Models. It will generate a file called studio_knowledge_models.csv
with the found CASE WHEN statements and their exact locations.
check_knowledge_models()