def _generate_task()

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

• 16 lines of code
• 4 McCabe index (conditional complexity)

    def _generate_task(self, action: Action,
                       retry_options: Optional[RetryOptions] = None,
                       id_: Optional[Union[int, str]] = None,
                       parent: Optional[TaskBase] = None,
                       task_constructor=AtomicTask) -> Union[AtomicTask, RetryAbleTask, TimerTask]:
        """Generate an atomic or retryable Task based on an input.

        Parameters
        ----------
        action : Action
            The action backing the Task.
        retry_options : Optional[RetryOptions]
            RetryOptions for a with-retry task, by default None

        Returns
        -------
        Union[AtomicTask, RetryAbleTask]
            Either an atomic task or a retry-able task
        """
        # Create an atomic task
        task: Union[AtomicTask, RetryAbleTask]
        action_payload: Union[Action, List[Action]]

        # TODO: find cleanear way to do this
        if self._replay_schema is ReplaySchema.V1:
            action_payload = [action]
        else:
            action_payload = action
        task = task_constructor(id_, action_payload)
        task.parent = parent

        # if task is retryable, provide the retryable wrapper class
        if not (retry_options is None):
            task = RetryAbleTask(task, retry_options, self)
        return task