Skip to content

tanjun.schedules#

Interface and interval implementation for a Tanjun based callback scheduler.

AbstractSchedule #

Bases: abc.ABC

Abstract callback schedule class.

callback() abstractmethod property #

Return the callback attached to the schedule.

This will be an asynchronous function which takes zero positional arguments, returns None and may be relying on dependency injection.

copy() abstractmethod #

Copy the schedule.

RETURNS DESCRIPTION
Self

The copied schedule.

RAISES DESCRIPTION
RuntimeError

If the schedule is active.

force_stop() abstractmethod #

Stop the schedule while cancelling any active tasks.

RAISES DESCRIPTION
RuntimeError

If the schedule is not active.

is_alive() abstractmethod property #

Whether the schedule is alive.

start(client, /, *, loop=None) abstractmethod #

Start the schedule.

PARAMETER DESCRIPTION
client

The injector client calls should be resolved with.

TYPE: alluka.Client

loop

The event loop to use. If not provided, the current event loop will be used.

TYPE: typing.Optional[asyncio.AbstractEventLoop] DEFAULT: None

RAISES DESCRIPTION
RuntimeError

If the scheduled callback is already running. If the current or provided event loop isn't running.

stop() async abstractmethod #

Stop the schedule after waiting for any existing tasks to finish.

RAISES DESCRIPTION
RuntimeError

If the scheduled callback isn't running.

IntervalSchedule #

Bases: typing.Generic[_CallbackSigT], components.AbstractComponentLoader, AbstractSchedule

A callback schedule with an interval between calls.

__init__(callback, interval, /, *, fatal_exceptions=(), ignored_exceptions=(), max_runs=None) #

Initialise an interval schedule.

PARAMETER DESCRIPTION
callback

The callback for the schedule.

This should be an asynchronous function which takes no positional arguments, returns None and may use dependency injection.

TYPE: collections.abc.Callable[..., collections.abc.Coroutine[Any, Any, None]]

interval

The interval between calls. Passed as a timedelta, or a number of seconds.

TYPE: typing.Union[datetime.timedelta, int, float]

fatal_exceptions

A sequence of exceptions that will cause the schedule to stop if raised by the callback, start callback or stop callback.

TYPE: collections.Sequence[type[Exception]] DEFAULT: ()

ignored_exceptions

A sequence of exceptions that should be ignored if raised by the callback, start callback or stop callback.

TYPE: collections.Sequence[type[Exception]] DEFAULT: ()

max_runs

The maximum amount of times the schedule runs.

TYPE: typing.Optional[int] DEFAULT: None

interval() property #

The interval between scheduled callback calls.

set_fatal_exceptions(*exceptions) #

Set the exceptions that will stop a schedule.

If any of these exceptions are encountered, the task will stop.

PARAMETER DESCRIPTION
*exceptions

Types of the exceptions to stop the task on.

TYPE: type[Exception] DEFAULT: ()

RETURNS DESCRIPTION
Self

The schedule object to enable chianed calls.

set_ignored_exceptions(*exceptions) #

Set the exceptions that a schedule will ignore.

If any of these exceptions are encountered, there will be nothing printed to console.

PARAMETER DESCRIPTION
*exceptions

Types of the exceptions to ignore.

TYPE: type[Exception] DEFAULT: ()

RETURNS DESCRIPTION
Self

The schedule object to enable chained calls.

set_start_callback(callback) #

Set the callback executed before the schedule starts to run.

PARAMETER DESCRIPTION
callback

The callback to set.

TYPE: _CallbackSig

RETURNS DESCRIPTION
Self

The schedule instance to enable chained calls.

set_stop_callback(callback) #

Set the callback executed after the schedule is finished.

PARAMETER DESCRIPTION
callback

The callback to set.

TYPE: _CallbackSig

RETURNS DESCRIPTION
Self

The schedule instance to enable chained calls.

with_start_callback(callback) #

Set the callback executed before the schedule is finished/stopped.

PARAMETER DESCRIPTION
callback

The callback to set.

TYPE: collections.abc.Callable[..., collections.abc.Coroutine[Any, Any, None]]

RETURNS DESCRIPTION
collections.abc.Callable[..., collections.abc.Coroutine[Any, Any, None]]

The added callback.

Examples:

@component.with_schedule()
@tanjun.as_interval(1, max_runs=20)
async def interval():
    global counter
    counter += 1
    print(f"Run #{counter}")

@interval.with_start_callback
async def pre():
    print("pre callback")

with_stop_callback(callback) #

Set the callback executed after the schedule is finished.

PARAMETER DESCRIPTION
callback

The callback to set.

TYPE: collections.abc.Callable[..., collections.abc.Coroutine[Any, Any, None]]

RETURNS DESCRIPTION
collections.abc.Callable[..., collections.abc.Coroutine[Any, Any, None]]

The added callback.

Examples:

@component.with_schedule()
@tanjun.as_interval(1, max_runs=20)
async def interval():
    global counter
    counter += 1
    print(f"Run #{counter}")


@interval.with_stop_callback
async def post():
    print("pre callback")

TimeSchedule #

Bases: typing.Generic[_CallbackSigT], components.AbstractComponentLoader, AbstractSchedule

A schedule that runs at specific times.

__init__(callback, /, *, months=(), weekly=False, days=(), hours=(), minutes=(), seconds=0, fatal_exceptions=(), ignored_exceptions=(), timezone=None) #

Initialise the time schedule.

PARAMETER DESCRIPTION
callback

The callback for the schedule.

This should be an asynchronous function which takes no positional arguments, returns None and may use dependency injection.

TYPE: collections.abc.Callable[..., collections.abc.Coroutine[Any, Any, None]]

months

Either one or multiple months the schedule shouldrun on.

If this is not specified or an empty sequence then the schedule will run on all months.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: ()

weekly

Whether the schedule should run on a weekly basis.

TYPE: bool DEFAULT: False

days

Either one or multiple days the schedule should run on.

When weekly is True, days will refer to the days of the week (range(7)).

Otherwise this will refer to the days of the month (range(32)). For months where less than 31 days exist, numbers which are too large will be ignored.

If this is not specified or an empty sequence, then the schedule will on all days.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: ()

hours

Either one or multiple hours the schedule should run on.

If this is not specified or an empty sequence then the schedule will run on all hours.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: ()

minutes

Either one or multiple minutes the schedule should run on.

If this is not specified or an empty sequence then the schedule will run on all minutes.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: ()

seconds

Either one or multiple seconds the schedule should run on.

Defaults to the start of the minute if not specified or an empty sequence.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: 0

fatal_exceptions

A sequence of exceptions that will cause the schedule to stop if raised by the callback, start callback or stop callback.

TYPE: collections.Sequence[type[Exception]] DEFAULT: ()

ignored_exceptions

A sequence of exceptions that should be ignored if raised by the callback, start callback or stop callback.

TYPE: collections.Sequence[type[Exception]] DEFAULT: ()

timezone

The timezone to use for the schedule.

If this is not specified then the system's local timezone will be used.

TYPE: typing.Optional[datetime.timezone] DEFAULT: None

RAISES DESCRIPTION
ValueError

Raises a value error for any of the following reasons:

  • If months has any values outside the range of range(1, 13).
  • If days has any values outside the range of range(1, 32) when weekly is False or outside the range of range(1, 8) when weekly is True.
  • If hours has any values outside the range of range(0, 24).
  • If minutes has any values outside the range of range(0, 60).
  • If seconds has any values outside the range of range(0, 60).

set_fatal_exceptions(*exceptions) #

Set the exceptions that will stop a schedule.

If any of these exceptions are encountered, the task will stop.

PARAMETER DESCRIPTION
*exceptions

Types of the exceptions to stop the task on.

TYPE: type[Exception] DEFAULT: ()

RETURNS DESCRIPTION
Self

The schedule object to enable chianed calls.

set_ignored_exceptions(*exceptions) #

Set the exceptions that a schedule will ignore.

If any of these exceptions are encountered, there will be nothing printed to console.

PARAMETER DESCRIPTION
*exceptions

Types of the exceptions to ignore.

TYPE: type[Exception] DEFAULT: ()

RETURNS DESCRIPTION
Self

The schedule object to enable chained calls.

as_interval(interval, /, *, fatal_exceptions=(), ignored_exceptions=(), max_runs=None) #

Decorator to create an schedule.

PARAMETER DESCRIPTION
interval

The interval between calls. Passed as a timedelta, or a number of seconds.

TYPE: typing.Union[int, float, datetime.timedelta]

fatal_exceptions

A sequence of exceptions that will cause the schedule to stop if raised by the callback, start callback or stop callback.

TYPE: collections.Sequence[type[Exception]] DEFAULT: ()

ignored_exceptions

A sequence of exceptions that should be ignored if raised by the callback, start callback or stop callback.

TYPE: collections.Sequence[type[Exception]] DEFAULT: ()

max_runs

The maximum amount of times the schedule runs.

TYPE: typing.Optional[int] DEFAULT: None

RETURNS DESCRIPTION
collections.Callable[[_CallbackSigT], tanjun.scheduling.IntervalSchedule[_CallbackSigT]]

The decorator used to create the schedule.

This should be decorating an asynchronous function which takes no positional arguments, returns None and may use dependency injection.

as_time_schedule(*, months=(), weekly=False, days=(), hours=(), minutes=(), seconds=0, fatal_exceptions=(), ignored_exceptions=(), timezone=None) #

Create a time schedule through a decorator call.

PARAMETER DESCRIPTION
months

Either one or multiple months the schedule should run on.

If this is not specified or an empty sequence then the schedule will run on all months.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: ()

weekly

Whether the schedule should run on a weekly basis.

TYPE: bool DEFAULT: False

days

Either one or multiple days the schedule should run on.

When weekly is True, days will refer to the days of the week (range(7)).

Otherwise this will refer to the days of the month (range(32)). For months where less than 31 days exist, numbers which are too large will be ignored.

If this is not specified or an empty sequence, then the schedule will on all days.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: ()

hours

Either one or multiple hours the schedule should run on.

If this is not specified or an empty sequence then the schedule will run on all hours.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: ()

minutes

Either one or multiple minutes the schedule should run on.

If this is not specified or an empty sequence then the schedule will run on all minutes.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: ()

seconds

Either one or multiple seconds the schedule should run on.

Defaults to the start of the minute if not specified or an empty sequence.

TYPE: typing.Union[int, collections.Sequence[int]] DEFAULT: 0

fatal_exceptions

A sequence of exceptions that will cause the schedule to stop if raised by the callback, start callback or stop callback.

TYPE: collections.Sequence[type[Exception]] DEFAULT: ()

ignored_exceptions

A sequence of exceptions that should be ignored if raised by the callback, start callback or stop callback.

TYPE: collections.Sequence[type[Exception]] DEFAULT: ()

timezone

The timezone to use for the schedule.

If this is not specified then the system's local timezone will be used.

TYPE: typing.Optional[datetime.timezone] DEFAULT: None

RETURNS DESCRIPTION
collections.Callable[[_CallbackSigT], tanjun.scheduling.TimeSchedule[_CallbackSigT]]

The decorator used to create the schedule.

This should be decorating an asynchronous function which takes no positional arguments, returns None and may use dependency injection.

RAISES DESCRIPTION
ValueError

Raises a value error for any of the following reasons:

  • If months has any values outside the range of range(1, 13).
  • If days has any values outside the range of range(1, 32) when weekly is False or outside the range of range(1, 7) when weekly is True.
  • If hours has any values outside the range of range(0, 24).
  • If minutes has any values outside the range of range(0, 60).
  • If seconds has any values outside the range of range(0, 60).