tanjun.schedules#

Interface and interval implementation for a Tanjun based callback scheduler.

AbstractSchedule #

Bases: abc.ABC

Abstract callback schedule class.

callback `property` `abstractmethod` #

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 `property` `abstractmethod` #

is_alive: bool

Whether the schedule is alive.

copy `abstractmethod` #

copy()

Copy the schedule.

RETURNS	DESCRIPTION
`Self`	The copied schedule.

RAISES	DESCRIPTION
`RuntimeError`	If the schedule is active.

force_stop `abstractmethod` #

force_stop()

Stop the schedule while cancelling any active tasks.

RAISES	DESCRIPTION
`RuntimeError`	If the schedule is not active.

start `abstractmethod` #

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

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 `abstractmethod` `async` #

stop()

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.

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.

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`

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.

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 #

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 #

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 #

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 #

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 #

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.

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.

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 #

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 #

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 #

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.

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 #

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.

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

tanjun.schedules#

AbstractSchedule #

callback property abstractmethod #

is_alive property abstractmethod #

copy abstractmethod #

force_stop abstractmethod #

start abstractmethod #

stop abstractmethod async #

IntervalSchedule #

interval property #

__init__ #

set_fatal_exceptions #

set_ignored_exceptions #

set_start_callback #

set_stop_callback #

with_start_callback #

with_stop_callback #

TimeSchedule #

__init__ #

set_fatal_exceptions #

set_ignored_exceptions #

as_interval #

as_time_schedule #

callback `property` `abstractmethod` #

is_alive `property` `abstractmethod` #

copy `abstractmethod` #

force_stop `abstractmethod` #

start `abstractmethod` #

stop `abstractmethod` `async` #

interval `property` #

init #

init #