1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
from abc import ABC, abstractmethod
from typing import Callable, Optional, TypeVar, Union
from .disposable import DisposableBase
from .scheduler import RelativeTime, ScheduledAction
_TState = TypeVar("_TState") # Can be anything
ScheduledPeriodicAction = Callable[[Optional[_TState]], Optional[_TState]]
ScheduledSingleOrPeriodicAction = Union[
ScheduledAction[_TState], ScheduledPeriodicAction[_TState]
]
class PeriodicSchedulerBase(ABC):
"""PeriodicScheduler abstract base class."""
__slots__ = ()
@abstractmethod
def schedule_periodic(
self,
period: RelativeTime,
action: ScheduledPeriodicAction[_TState],
state: Optional[_TState] = None,
) -> DisposableBase:
"""Schedules a periodic piece of work.
Args:
period: Period in seconds or timedelta for running the
work periodically.
action: Action to be executed.
state: [Optional] Initial state passed to the action upon
the first iteration.
Returns:
The disposable object used to cancel the scheduled
recurring action (best effort).
"""
return NotImplemented
__all__ = [
"PeriodicSchedulerBase",
"ScheduledPeriodicAction",
"ScheduledSingleOrPeriodicAction",
"RelativeTime",
]
|
