AsyncioScheduler

__init__(loop) -> None

def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
    super().__init__()
    self._loop = loop
    self._thread = threading.get_ident()
    self._ready: deque[Task] = deque()
    self._draining_scheduled = False

current() -> AsyncioScheduler

@classmethod
def current(cls) -> AsyncioScheduler:
    """The scheduler for the running loop. Must be called on the loop."""
    return cls(asyncio.get_running_loop())

schedule(task) -> None

def schedule(self, task: Task) -> None:
    if threading.get_ident() != self._thread:
        self._loop.call_soon_threadsafe(self.schedule, task)
        return
    self._ready.append(self._tracked(task))
    self._drain_soon()

call_later(delay, task, *, daemon=False) -> Cancel

def call_later(self, delay: Seconds, task: Task, *, daemon: bool = False) -> Cancel:
    return self._loop.call_later(max(delay, 0.0), task).cancel

track(tracker, f) -> None

def track(self, tracker: WakeTracker, f: Callable[[], None]) -> None:
    """Run ``f``, counting what it schedules against ``tracker``."""
    previous = self._tracking.tracker
    self._tracking.tracker = tracker
    try:
        f()
    finally:
        self._tracking.tracker = previous