def try_enqueue()

in smallpond/execution/scheduler.py [0:0]

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

    def try_enqueue(self, tasks: Union[Iterable[Task], Task]):
        tasks = tasks if isinstance(tasks, Iterable) else [tasks]
        for task in tasks:
            task = self.copy_task_for_execution(task)
            if task.key in self.succeeded_tasks:
                logger.debug(f"task {repr(task)} already succeeded, skipping")
                self.try_enqueue(self.get_runnable_tasks(self.succeeded_tasks[task.key]))
                continue
            if task.runtime_id in self.scheduled_tasks:
                logger.debug(f"task {repr(task)} already scheduled, skipping")
                continue
            # save enqueued task
            self.scheduled_tasks[task.runtime_id] = task
            if self.standalone_mode or task.exec_on_scheduler or task.skip_when_any_input_empty:
                self.local_queue.append(task)
            else:
                self.sched_queue.append(task)