API Docs

Builders

Builders help to create complex objects in a more readable and convenient way.

class JobBuilder(scheduler, executor, job_store=None)

at(trigger, coro_func, *args, job_id=None, **kwargs)

Create a job that will run at a specified time.

Parameters:

trigger (TriggerObject)
coro_func (Callable[..., Awaitable[Any]]) – Function which will be called
args (Any) – Positional arguments that will be passed to the coroutine function
job_id (Optional[TypeVar(IdType, bound= Hashable)]) – Job id to manually identify the job
kwargs (Any) – Keyword arguments that will be passed to the coroutine function

Return type:

DateTimeJobControl

Returns:

Created job

countdown(secs, coro_func, *args, job_id=None, **kwargs)

Create a job that count town a certain time and then execute.

Parameters:

secs (timedelta | TimeDelta | int | float | str) – countdown time in seconds
coro_func (Callable[..., Awaitable[Any]]) – Function which will be called
args (Any) – Positional arguments that will be passed to the coroutine function
job_id (Optional[TypeVar(IdType, bound= Hashable)]) – Job id to manually identify the job
kwargs (Any) – Keyword arguments that will be passed to the coroutine function

Return type:

CountdownJobControl

Returns:

Created job

once(instant, coro_func, *args, job_id=None, **kwargs)

Create a job that runs once.

Parameters:

instant (datetime | None | str | time | Time | timedelta | TimeDelta | int | float | SystemDateTime | Instant) – countdown time in seconds
coro_func (Callable[..., Awaitable[Any]]) – Function which will be called
args (Any) – Positional arguments that will be passed to the coroutine function
job_id (Optional[TypeVar(IdType, bound= Hashable)]) – Job id to manually identify the job
kwargs (Any) – Keyword arguments that will be passed to the coroutine function

Return type:

OneTimeJobControl

Returns:

Created job

class TriggerBuilder

static dawn()

Triggers at dawn.

Return type:: TriggerObject

static dusk()

Triggers at dusk.

Return type:: TriggerObject

static group(*builders)

Group multiple triggers together. The triggers will be checked and the trigger that runs next will be used

Parameters:: builders (TriggerObject) – Triggers that should be grouped together
Return type:: TriggerObject

static interval(start, interval)

Triggers at a fixed interval from a given start time.

Parameters:

start (datetime | None | str | time | Time | timedelta | TimeDelta | int | float | SystemDateTime | Instant) – When this trigger will run for the first time. Note: It’s not possible to specify a start time greater than the interval time. Since the producer is stateless it will automatically select the next appropriate run time. Example: start 90 minutes, interval 60 minutes -> first run will be in 30 minutes. This makes it easy to ensure that the job will always run at a certain time by specifying the start time isntead of a delta and the interval. E.g. start 00:15:00 and interval 1 hour
interval (timedelta | TimeDelta | int | float | str) – The interval how this trigger will be repeated

Return type:

TriggerObject

static noon()

Triggers at noon.

Return type:: TriggerObject

static sun_azimuth(azimuth)

Triggers at a specific sun azimuth

Parameters:: azimuth (float) – Sun azimuth in degrees
Return type:: TriggerObject

static sun_elevation(elevation, direction)

Triggers at a specific sun elevation

Parameters:

elevation (float) – Sun elevation in degrees
direction (Literal['rising', 'setting']) – rising or falling

Return type:

TriggerObject

static sunrise()

Triggers at sunrise.

Return type:: TriggerObject

static sunset()

Triggers at sunset.

Return type:: TriggerObject

static time(time, *, clock_forward=None, clock_backward=None)

Triggers at a specific time of day. When the time of day is during a daylight saving time transition it has to be explicitly specified how the transition should be handled.

Parameters:

time (time | Time | str) – The time of day the trigger should fire
clock_forward (Union[SkippedTimeBehavior, Literal['skip', 'earlier', 'later', 'after'], None]) – How to handle the transition when the clock moves forward
clock_backward (Union[RepeatedTimeBehavior, Literal['skip', 'earlier', 'later', 'twice'], None]) – How to handle the transition when the clock moves backward

Return type:

TriggerObject

class FilterBuilder

static all(*filters)

Passes an instant only when all the filters pass it

Return type:: FilterObject

static any(*filters)

Passes an instant as soon as any of the filters passes it

Return type:: FilterObject

static days(*days)

Let only certain days of the month pass through

Parameters:: days (int | str | Iterable[int | str]) – days of the month as str, int, or str range ('1-7', '1-5,10-15', '1,5,7')
Return type:: FilterObject

static holidays(holidays=None)

Let only holidays pass through

Parameters:: holidays (HolidayBase | None) – Optional holiday object to use. If not provided the default holiday object will be used Userful if you want to use holidays from e.g. another country or another subdivision
Return type:: FilterObject

static months(*months)

Let only certain months pass through

Parameters:: months (int | str | Iterable[int | str]) – Month as str, int, or str range ('Jan-Jun', '1-3,10-12', 'Oct-Feb', 'Jan, March')
Return type:: FilterObject

static not_(filter)

Invert a filter

Return type:: FilterObject

static not_work_days(holidays=None)

Let only days pass through that are not working days

Parameters:: holidays (HolidayBase | None) – Optional holiday object to use. If not provided the default holiday object will be used Userful if you want to use holidays from e.g. another country or another subdivision
Return type:: FilterObject

static time(lower=None, upper=None)

Restrict the trigger instant to a time range

Return type:: FilterObject

static weekdays(*weekdays)

Let only certain weekdays pass through

Parameters:: weekdays (int | str | Iterable[int | str]) – days of week as str, int, or str range ('Mon-Fri', 'Sun-Mon', 'Mo-Mi, 'Fr-So')
Return type:: FilterObject

static work_days(holidays=None)

Let only working days pass through

Parameters:: holidays (HolidayBase | None) – Optional holiday object to use. If not provided the default holiday object will be used Userful if you want to use holidays from e.g. another country or another subdivision
Return type:: FilterObject

Job Control

class OneTimeJobControl(job)

cancel()

Cancel the job

Return type:: Self

property id: Hashable: Get the job’s id

property last_run_datetime: datetime | None: Get the last run time as a naive datetime object (without timezone set) or None if yet run

property next_run_datetime: datetime | None: Get the next run time as a naive datetime object (without timezone set) or None if not scheduled

property status: JobStatusEnum: Get the status of the job

class CountdownJobControl(job)

cancel()

Cancel the job

Return type:: Self

property id: Hashable: Get the job’s id

property last_run_datetime: datetime | None: Get the last run time as a naive datetime object (without timezone set) or None if yet run

property next_run_datetime: datetime | None: Get the next run time as a naive datetime object (without timezone set) or None if not scheduled

reset()

Start the countdown again

Return type:: Self

set_countdown(secs)

Set the countdown time

Parameters:: secs (float) – Seconds that will be used for the next reset call
Return type:: Self

property status: JobStatusEnum: Get the status of the job

stop()

Stop the countdown

Return type:: Self

class DateTimeJobControl(job)

cancel()

Cancel the job

Return type:: Self

property id: Hashable: Get the job’s id

property last_run_datetime: datetime | None: Get the last run time as a naive datetime object (without timezone set) or None if yet run

property next_run_datetime: datetime | None: Get the next run time as a naive datetime object (without timezone set) or None if not scheduled

pause()

Stop executing this job

Return type:: Self

resume()

Resume executing this job

Return type:: Self

property status: JobStatusEnum: Get the status of the job

Exception Handling

set_exception_handler(handler)

Return type:: None

Sun Position

set_location(latitude, longitude, elevation=0.0)

Latitude and longitude can be set either as a float or as a string. For strings they must be of the form

degrees°minutes’seconds”[N|S|E|W] e.g. 51°31’N

minutes & seconds are optional.

Parameters:

latitude (float | str) – Latitude - Northern latitudes should be positive (e.g. 52.5185537)
longitude (float | str) – Longitude - Eastern longitudes should be positive (e.g. 13.3758636)
elevation (float | tuple[float, float]) – Elevation above sea level (e.g. 43.0)

Return type:

None

Returns:

get_sun_position(instant)

Return the sun position (azimuth and elevation) at a given instant.

Parameters:: instant (datetime | None | str | time | Time | timedelta | TimeDelta | int | float | SystemDateTime | Instant) – Instant to get the sun position at or None for now
Return type:: tuple[float, float]
Returns:: Azimuth and elevation of the sun at the specified instant

Holidays

setup_holidays(country, subdiv=None, *, observed=True, language=None, categories=None)

Set up the holidays

Parameters:

country (str) – An ISO 3166-1 Alpha-2 country code.
subdiv (str | None) – The subdivision (e.g. state or province) as a ISO 3166-2 code or its alias; not implemented for all countries (see documentation).
observed (bool) – Whether to include the dates of when public holiday are observed (e.g. a holiday falling on a Sunday being observed the following Monday). False may not work for all countries.
language (str | None) – The language which the returned holiday names will be translated into. It must be an ISO 639-1 (2-letter) language code. If the language translation is not supported the original holiday names will be used.
categories (str | Iterable[str] | None) – Requested holiday categories.

Return type:

None

Returns:

add_holiday(date, name=None)

Add a new holiday. If the date is already a holiday the names will be joined together with a semicolon as separator.

Parameters:

date (date | datetime | str | None | Date | SystemDateTime | Instant) – Date for the new holiday.
name (str | None) – Name of the Holiday, if not provided it will be set to “Holiday”

Return type:

None

get_holiday_name(date, default=None)

Get the holiday name of a given date if the date is a holiday else return the given default.

Parameters:

date – Date
default – Default value to return if the date is not a holiday

Return type:

str

get_holidays_by_name(name, *, lookup='icontains')

Return a list of all holiday dates matching the provided holiday name. The match will be made case insensitively and partial matches will be included by default

Parameters:

name (str) – The holiday’s name to try to match.
lookup (Literal['contains', 'exact', 'startswith', 'icontains', 'iexact', 'istartswith']) –

The holiday name lookup type:
contains - case sensitive contains match; exact - case sensitive exact match; startswith - case sensitive starts with match; icontains - case insensitive contains match; iexact - case insensitive exact match; istartswith - case insensitive starts with match;

Return type:

list[date]

Returns:

A list of all holiday dates matching the provided holiday name.

is_holiday(date)

Check if a given date is a holiday.

Parameters:: date (date | datetime | str | None | Date | SystemDateTime | Instant) – Date to check or None for today
Return type:: bool
Returns:: True if the date is a holiday, False otherwise

pop_holiday(date, default=None)

Delete a holiday and return the name of the deleted holiday

Parameters:

date – Date to delete or None for today
default – Default value to return if the date is not a holiday

Returns:

Name of the deleted holiday or the default value

TriggerObject

class TriggerObject(producer)

earliest(earliest, *, clock_forward=None, clock_backward=None)

Set the earliest time of day the trigger can fire. If the trigger would fire before the earliest time, the earliest time will be used instead.

Parameters:

earliest (time | Time | str) – The time of day before the trigger can not fire
clock_forward (Union[SkippedTimeBehavior, Literal['skip', 'earlier', 'later', 'after'], None]) – How to handle the transition when the clock moves forward
clock_backward (Union[RepeatedTimeBehavior, Literal['skip', 'earlier', 'later', 'twice'], None]) – How to handle the transition when the clock moves backward

Return type:

Self

jitter(low, high=None)

Add jitter to the time returned by the trigger.

Parameters:

low (timedelta | TimeDelta | int | float | str) – The lower bound of the jitter
high (timedelta | TimeDelta | int | float | str | None) – The upper bound of the jitter. If not specified the jitter will be 0 .. low

Return type:

Self

latest(latest, *, clock_forward=None, clock_backward=None)

Set the latest time of day the trigger can fire. If the trigger would fire after the latest time, the latest time will be used instead.

Parameters:

latest (time | Time | str) – The time of day before the trigger can not fire
clock_forward (Union[SkippedTimeBehavior, Literal['skip', 'earlier', 'later', 'after'], None]) – How to handle the transition when the clock moves forward
clock_backward (Union[RepeatedTimeBehavior, Literal['skip', 'earlier', 'later', 'twice'], None]) – How to handle the transition when the clock moves backward

Return type:

Self

offset(offset)

Offset the time returned by the trigger

Parameters:: offset (timedelta | TimeDelta | int | float | str) – The offset (positive or negative)
Return type:: Self

only_at(filter)

Add a filter to the trigger which can be used to allow or disallow certain times.

Parameters:: filter (FilterObject) – The filter to apply to the trigger
Return type:: Self

only_on(filter)

Add a filter to the trigger which can be used to allow or disallow certain times.

Parameters:: filter (FilterObject) – The filter to apply to the trigger
Return type:: Self

Job status

class JobStatusEnum(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

property is_finished: bool: True if the job is finished

property is_paused: bool: True if the job is paused

property is_running: bool: True if the job is running

Task Managers

Parallel

class ParallelTaskManager

create_task(coro, *, name=None)

Create a new task. All tasks will run parallel

Parameters:

coro (Coroutine) – Coro
name (str | None) – Task name

Return type:

Task | None

class LimitingParallelTaskManager(parallel, action=ParallelTaskPolicy.SKIP)

__init__(parallel, action=ParallelTaskPolicy.SKIP)

Parameters:

parallel (int) – Maximum parallel tasks
action (Union[ParallelTaskPolicy, Literal['skip', 'cancel_first', 'cancel_last']]) – What to do when the limit is exceeded

create_task(coro, *, name=None)

Create a new task. All tasks will run until the max parallel limit is reached. Depending on the action, the oldest or the newest the task will be cancelled or the to be created task will be skipped.

Parameters:

coro (Coroutine) – Coro
name (str | None) – Task name

Return type:

Task | None

Sequential

class SequentialTaskManager

create_task(coro, *, name=None)

Create a new task. All tasks will sequentially

Parameters:

coro (Coroutine) – Coro
name (str | None) – Task name

Return type:

Task | None

class LimitingSequentialTaskManager(max_queue, action=SequentialTaskPolicy.SKIP)

__init__(max_queue, action=SequentialTaskPolicy.SKIP)

Parameters:

max_queue (int) – maximum number of tasks in the queue
action (Union[SequentialTaskPolicy, Literal['skip', 'skip_first', 'skip_last']]) – what to do when the limit will be exceeded

create_task(coro, *, name=None)

Create a new task. All tasks will sequentially. When the limit is exceeded either the first task in the queue, the last task in the queue or the to be created task will be skipped.

Parameters:

coro (Coroutine) – Coro
name (str | None) – Task name

Return type:

Task | None

class SequentialDeduplicatingTaskManager

create_task(coro, key, *, name=None)

Create a new task. All tasks will sequentially. A key is used to deduplicate the tasks. If a task with the same key is already in the queue it will be skipped.

Parameters:

coro (Coroutine) – Coro
key (Hashable) – Key used for deduplicating tasks
name (str | None) – Task name

Return type:

Task | None