Skip to content

tanjun.schedules#

Interface and interval implementation for a Tanjun based callback scheduler.

AbstractSchedule #

Bases: ABC

Abstract callback schedule class.

callback abstractmethod property #

callback: _CallbackSig

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.

is_alive abstractmethod property #

is_alive: bool

Whether the schedule is alive.

copy abstractmethod #

copy()

Copy the schedule.

Returns:

  • Self

    The copied schedule.

Raises:

force_stop abstractmethod #

force_stop()

Stop the schedule while cancelling any active tasks.

Raises:

start abstractmethod #

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

Start the schedule.

Parameters:

Raises:

  • RuntimeError

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

stop abstractmethod async #

stop()

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

Raises:

IntervalSchedule #

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

A callback schedule with an interval between calls.

This should be loaded into a component using either Component.load_from_scope, Component.add_schedule or Component.with_schedule, and will be started and stopped with the linked tanjun client.

interval property #

interval: datetime.timedelta

The interval between scheduled callback calls.

__init__ #

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

Initialise an interval schedule.

Parameters:

  • callback (abc.abc.Callable[..., abc.abc.Coroutine[Any, Any, None]]) –

    The callback for the schedule.

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

  • interval (typing.Union[timedelta, int, float]) –

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

  • fatal_exceptions (abc.Sequence[type[Exception]], default: () ) –

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

  • ignored_exceptions (abc.Sequence[type[Exception]], default: () ) –

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

  • max_runs (typing.Optional[int], default: None ) –

    The maximum amount of times the schedule runs.

set_fatal_exceptions #

set_fatal_exceptions(*exceptions)

Set the exceptions that will stop a schedule.

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

Parameters:

  • *exceptions (type[Exception], default: () ) –

    Types of the exceptions to stop the task on.

Returns:

  • Self

    The schedule object to enable chianed calls.

set_ignored_exceptions #

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.

Parameters:

  • *exceptions (type[Exception], default: () ) –

    Types of the exceptions to ignore.

Returns:

  • Self

    The schedule object to enable chained calls.

set_start_callback #

set_start_callback(callback)

Set the callback executed before the schedule starts to run.

Parameters:

  • callback (_CallbackSig) –

    The callback to set.

Returns:

  • Self

    The schedule instance to enable chained calls.

set_stop_callback #

set_stop_callback(callback)

Set the callback executed after the schedule is finished.

Parameters:

  • callback (_CallbackSig) –

    The callback to set.

Returns:

  • Self

    The schedule instance to enable chained calls.

with_start_callback #

with_start_callback(callback)

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

Parameters:

  • callback (abc.abc.Callable[..., abc.abc.Coroutine[Any, Any, None]]) –

    The callback to set.

Returns:

  • abc.abc.Callable[..., abc.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 #

with_stop_callback(callback)

Set the callback executed after the schedule is finished.

Parameters:

  • callback (abc.abc.Callable[..., abc.abc.Coroutine[Any, Any, None]]) –

    The callback to set.

Returns:

  • abc.abc.Callable[..., abc.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], AbstractComponentLoader, AbstractSchedule

A schedule that runs at specific times.

This should be loaded into a component using either Component.load_from_scope, Component.add_schedule or Component.with_schedule and will be started and stopped with the linked tanjun client.

__init__ #

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

Initialise the time schedule.

Parameters:

  • callback (abc.abc.Callable[..., abc.abc.Coroutine[Any, Any, None]]) –

    The callback for the schedule.

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

  • months (typing.Union[int, abc.Sequence[int]], default: () ) –

    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.

  • weekly (bool, default: False ) –

    Whether the schedule should run on a weekly basis.

  • days (typing.Union[int, abc.Sequence[int]], default: () ) –

    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 run on all days.

  • hours (typing.Union[int, abc.Sequence[int]], default: () ) –

    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.

  • minutes (typing.Union[int, abc.Sequence[int]], default: () ) –

    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.

  • seconds (typing.Union[int, abc.Sequence[int]], default: 0 ) –

    Either one or multiple seconds the schedule should run on.

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

  • fatal_exceptions (abc.Sequence[type[Exception]], default: () ) –

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

  • ignored_exceptions (abc.Sequence[type[Exception]], default: () ) –

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

  • timezone (typing.Optional[datetime.timezone], default: None ) –

    The timezone to use for the schedule.

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

Raises:

  • 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 #

set_fatal_exceptions(*exceptions)

Set the exceptions that will stop a schedule.

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

Parameters:

  • *exceptions (type[Exception], default: () ) –

    Types of the exceptions to stop the task on.

Returns:

  • Self

    The schedule object to enable chianed calls.

set_ignored_exceptions #

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.

Parameters:

  • *exceptions (type[Exception], default: () ) –

    Types of the exceptions to ignore.

Returns:

  • Self

    The schedule object to enable chained calls.

as_interval #

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

Decorator to create an schedule.

Examples:

@component.with_schedule
@tanjun.as_interval(datetime.timedelta(minutes=5))  # This will run every 5 minutes
async def interval(client: alluka.Injected[tanjun.abc.Client]) -> None:
    ...

This should be loaded into a component using either Component.with_schedule or Component.load_from_scope, and will be started and stopped with the linked tanjun client.

Parameters:

  • interval (typing.Union[int, float, timedelta]) –

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

  • fatal_exceptions (abc.Sequence[type[Exception]], default: () ) –

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

  • ignored_exceptions (abc.Sequence[type[Exception]], default: () ) –

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

  • max_runs (typing.Optional[int], default: None ) –

    The maximum amount of times the schedule runs.

Returns:

  • abc.Callable[[_CallbackSigT], abc.schedules.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 #

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.

Examples:

@component.with_schedule
@tanjun.as_time_schedule(  # This will run every week day at 8:00 and 16:00 UTC.
    minutes=0, hours=[8, 16], days=range(0, 5), weekly=True, timezone=datetime.timezone.utc
)
async def interval(client: alluka.Injected[tanjun.abc.Client]) -> None:
    ...

This should be loaded into a component using either Component.with_schedule or Component.load_from_scope, and will be started and stopped with the linked tanjun client.

Parameters:

  • months (typing.Union[int, abc.Sequence[int]], default: () ) –

    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.

  • weekly (bool, default: False ) –

    Whether the schedule should run on a weekly basis.

  • days (typing.Union[int, abc.Sequence[int]], default: () ) –

    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 run on all days.

  • hours (typing.Union[int, abc.Sequence[int]], default: () ) –

    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.

  • minutes (typing.Union[int, abc.Sequence[int]], default: () ) –

    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.

  • seconds (typing.Union[int, abc.Sequence[int]], default: 0 ) –

    Either one or multiple seconds the schedule should run on.

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

  • fatal_exceptions (abc.Sequence[type[Exception]], default: () ) –

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

  • ignored_exceptions (abc.Sequence[type[Exception]], default: () ) –

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

  • timezone (typing.Optional[datetime.timezone], default: None ) –

    The timezone to use for the schedule.

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

Returns:

  • abc.Callable[[_CallbackSigT], abc.schedules.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:

  • 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).