import json
def handleRocketlaneEvent(payloadJson):
payloadData = json.loads(payloadJson)
eventType = payloadData["eventType"]
if eventType == 'TASK_CEATED':
taskId = payloadData["data"]["task"]["taskId"]
taskName = payloadData["data"]["task"]["taskName"]
print(f'Task created with id {taskId} and name {taskName}')
3. Extract old and new values, when task field is updated
import json
def handleRocketlaneEvent(payloadJson):
payloadData = json.loads(payloadJson)
eventType = payloadData["eventType"]
if eventType == 'TASK_UPDATED':
changedFields = payloadData["changeLog"]["changedFields"]
# Consuming change log if assignees is present in changedFields
if 'assignees' in changedFields:
taskName = payloadData["data"]["task"]["taskName"]
assigneesFrom = payloadData["changeLog"]["from"]["assignees"]
assigneeFromMembers = assigneesFrom["members"] if assigneesFrom else ''
isAssigneeFromMemsPresent= assigneeFromMembers and len(assigneeFromMembers) > 0
assigneesFromMemEmails= [member["emailId"] for member in assigneeFromMembers] if isAssigneeFromMemsPresent else ''
assigneesTo = payloadData["changeLog"]["to"]["assignees"]
assigneeToMembers = assigneesTo["members"] if assigneesTo else ''
isAssigneeToMemsPresent= assigneeToMembers and len(assigneeToMembers) > 0
assigneesToMemEmails= [member["emailId"] for member in assigneeToMembers] if isAssigneeToMemsPresent else ''
new_assignees = list(filter(lambda item: item not in assigneesFromMemEmails, assigneesToMemEmails))
if '[email protected]' in new_assignees:
print( f'Aaron Paul is assigned to the task : {taskName}')