def __init__()

in azure/durable_functions/models/TaskOrchestrationExecutor.py [0:0]

• 14 lines of code
• 2 McCabe index (conditional complexity)

    def __init__(self):
        """Initialize TaskOrchestrationExecutor."""
        # A mapping of event types to a tuple of
        #   (1) whether the event type represents a task success
        #   (2) the attribute in the corresponding event object that identifies the Task
        # this mapping is used for processing events that transition a Task from its running state
        # to a terminal one
        SetTaskValuePayload = namedtuple("SetTaskValuePayload", ("is_success", "task_id_key"))
        self.event_to_SetTaskValuePayload = dict([
            (HistoryEventType.TASK_COMPLETED, SetTaskValuePayload(True, "TaskScheduledId")),
            (HistoryEventType.TIMER_FIRED, SetTaskValuePayload(True, "TimerId")),
            (HistoryEventType.SUB_ORCHESTRATION_INSTANCE_COMPLETED,
             SetTaskValuePayload(True, "TaskScheduledId")),
            (HistoryEventType.EVENT_RAISED, SetTaskValuePayload(True, "Name")),
            (HistoryEventType.TASK_FAILED, SetTaskValuePayload(False, "TaskScheduledId")),
            (HistoryEventType.SUB_ORCHESTRATION_INSTANCE_FAILED,
             SetTaskValuePayload(False, "TaskScheduledId"))
        ])
        self.task_completion_events = set(self.event_to_SetTaskValuePayload.keys())
        self.initialize()