Skip to content

Scheduler Internals

These methods help the scheduler plan execution, find tasks, record statistics, and run individual tasks with retries. Most users only need the external execute() method.

astrum.scheduler.DynamicScheduler.detect_cycle 

detect_cycle() -> None
源代码位于： src/astrum/scheduler.py 
71
72
def detect_cycle(self) -> None:
    ExecutionPlanner(self.task_order).detect_cycle()

astrum.scheduler.DynamicScheduler.get_execute_timeline 

get_execute_timeline()
源代码位于： src/astrum/scheduler.py 
74
75
def get_execute_timeline(self):
    return ExecutionPlanner(self.task_order).get_execute_timeline()

astrum.scheduler.DynamicScheduler.find_task_by_name 

find_task_by_name(task_name: str) -> TaskCallable | None
源代码位于： src/astrum/scheduler.py 
77
78
79
80
81
def find_task_by_name(self, task_name: str) -> TaskCallable | None:
    for name, task in self.tasks:
        if name == task_name:
            return task
    return None