scheduling

Generic python scheduler.

class Scheduler(name)

Bases: object

Schedule the execution of coroutines and keep track of them.

When instantiating a Scheduler, a name must be provided. This name is used to distinguish the instance’s log messages from other instances. Using the name of the class or module containing the instance is suggested.

Coroutines can be scheduled immediately with schedule or in the future with schedule_at or schedule_later. A unique ID is required to be given in order to keep track of the resulting Tasks. Any scheduled task can be cancelled prematurely using cancel by providing the same ID used to schedule it.

The in operator is supported for checking if a task with a given ID is currently scheduled.

Any exception raised in a scheduled task is logged when the task is done.

__contains__(task_id)

Return True if a task with the given task_id is currently scheduled.

Parameters:

task_id (Hashable) – The task to look for.

Return type:

bool

Returns:

True if the task was found.

__init__(name)

Initialize a new Scheduler instance.

Parameters:

name (str) – The name of the Scheduler. Used in logging, and namespacing.

cancel(task_id)

Unschedule the task identified by task_id. Log a warning if the task doesn’t exist.

Parameters:

task_id (Hashable) – The task’s unique ID.

Return type:

None

cancel_all()

Unschedule all known tasks.

Return type:

None

schedule(task_id, coroutine)

Schedule the execution of a coroutine.

If a task with task_id already exists, close coroutine instead of scheduling it. This prevents unawaited coroutine warnings. Don’t pass a coroutine that’ll be re-used elsewhere.

Parameters:

• task_id (Hashable) – A unique ID to create the task with.

• coroutine (Coroutine) – The function to be called.

Return type:

None

schedule_at(time, task_id, coroutine)

Schedule coroutine to be executed at the given time.

If time is timezone aware, then use that timezone to calculate now() when subtracting. If time is naïve, then use UTC.

If time is in the past, schedule coroutine immediately.

If a task with task_id already exists, close coroutine instead of scheduling it. This prevents unawaited coroutine warnings. Don’t pass a coroutine that’ll be re-used elsewhere.

Parameters:

• time (datetime) – The time to start the task.

• task_id (Hashable) – A unique ID to create the task with.

• coroutine (Coroutine) – The function to be called.

Return type:

None

schedule_later(delay, task_id, coroutine)

Schedule coroutine to be executed after delay seconds.

If a task with task_id already exists, close coroutine instead of scheduling it. This prevents unawaited coroutine warnings. Don’t pass a coroutine that’ll be re-used elsewhere.

Parameters:

• delay (Union[int, float]) – How long to wait before starting the task.

• task_id (Hashable) – A unique ID to create the task with.

• coroutine (Coroutine) – The function to be called.

Return type:

None

create_task(coro, *, suppressed_exceptions=(), event_loop=None, **kwargs)

Wrapper for creating an asyncio.Task which logs exceptions raised in the task.

If the event_loop kwarg is provided, the task is created from that event loop, otherwise the running loop is used.

Parameters:

• coro (Coroutine[Any, Any, TypeVar(TASK_RETURN)]) – The function to call.

• suppressed_exceptions (tuple[type[Exception], ...]) – Exceptions to be handled by the task.

• event_loop (asyncio.AbstractEventLoop) – The loop to create the task from.

• kwargs – Passed to asyncio.create_task().

Returns:

The wrapped task.

Return type:

asyncio.Task