Scheduling Support for Statutory Holidays

[nathanielatom]

Is your feature request related to a problem? Please describe.
I would like to schedule tasks to run on statutory holidays (or some timedelta offset from them).

Describe the solution you'd like
Add a new Trigger class that fires for every statutory holiday defined in a given region, based on the https://pypi.org/project/holidays/ library. Since this only gives dates, the trigger would also accept a scalar timedelta which gets added to midnight of each statutory holiday.

Describe alternatives you've considered
Manually maintaining a list of holidays and setting up many such individual task triggers for each holiday would be cumbersome, and involve a lot of boilerplate, difficult to scale; I think this is a common enough use case to consider adding direct library support.

Additional context
I came up with a prototype solution for my use case, which may be possible to base a PR off of. I would be happy to contribute, but my schedule is very uncertain and so I cannot necessarily commit to it.

Here is my code:

import datetime as dt

from apscheduler.triggers.base import BaseTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.combining import OrTrigger


class RepeatableDateTrigger(DateTrigger):

    def get_next_fire_time(self, previous_fire_time, now):
        return self.run_date if now <= self.run_date else None


class HolidayTimedeltaTrigger(BaseTrigger):

    @staticmethod
    def assert_holidays_installed():
        class HolidaysModuleNotFoundError(ModuleNotFoundError):
            """ To be raised when the holidays library is not installed. """

        try:
            import holidays
        except ModuleNotFoundError as error:
            message = 'To use HolidayTimedeltaTrigger you will need to run `pip install holidays`'
            raise HolidaysModuleNotFoundError(message)

    def __init__(self, timedelta, start_date, end_date, timezone=None, jitter=None, **holiday_options):
        self.assert_holidays_installed()
        import holidays

        self.timedelta = timedelta
        self.start_date = start_date
        self.end_date = end_date
        self.timezone = timezone
        self.jitter = jitter
        self.holiday_options = holiday_options

        years = list(range(self.start_date.year, self.end_date.year + 1))
        self.holiday_store = holidays.CountryHoliday(years=years, **holiday_options)
        self.holiday_dates = self.holiday_store[self.start_date:self.end_date]
        self.holiday_datetimes = [dt.datetime.combine(holi, dt.time()) + self.timedelta for holi in self.holiday_dates]
        self._triggor = OrTrigger([RepeatableDateTrigger(run_date=holi, timezone=self.timezone) for holi in self.holiday_datetimes],
                                 jitter=self.jitter)

    def get_next_fire_time(self, previous_fire_time, now):
        return self._triggor.get_next_fire_time(previous_fire_time, now)

and usage:

import datetime

td = datetime.timedelta(hours=10)
start = datetime.datetime(2016, 6, 8)
end = datetime.datetime(2023, 3, 3)
tz = 'UTC'
holiday_options = {'country': 'Canada', 'prov': 'ON'} # years not allowed, determined from start_datetime and end_datetime

trigger = HolidayTimedeltaTrigger(td, start, end, tz, **holiday_options)

trigger
# <aps_holidays.HolidayTimedeltaTrigger at 0x1062d1d00>

trigger.get_next_fire_time(None, datetime.datetime.now(tz=datetime.timezone.utc))
# datetime.datetime(2021, 7, 1, 10, 0, tzinfo=<UTC>)

trigger.get_next_fire_time(datetime.datetime(2021, 5, 24, 10, 0, tzinfo=datetime.timezone.utc), datetime.datetime.now(tz=datetime.timezone.utc))
# datetime.datetime(2021, 7, 1, 10, 0, tzinfo=<UTC>)

trigger.get_next_fire_time(None, datetime.datetime(2021, 8, 4, tzinfo=datetime.timezone.utc))
# datetime.datetime(2021, 9, 6, 10, 0, tzinfo=<UTC>)

trigger.get_next_fire_time(datetime.datetime(2021, 8, 2, 10, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2021, 8, 4, tzinfo=datetime.timezone.utc))
# datetime.datetime(2021, 9, 6, 10, 0, tzinfo=<UTC>)

trigger.get_next_fire_time(None, datetime.datetime(2012, 4, 4, tzinfo=datetime.timezone.utc))
# datetime.datetime(2016, 7, 1, 10, 0, tzinfo=<UTC>)

trigger.get_next_fire_time(None, datetime.datetime(2112, 4, 4, tzinfo=datetime.timezone.utc))
# None

Thanks! This is a great and very powerful/flexible and helpful library!

[agronholm]

I think this is a good idea, and it could work great with combining triggers. However, the priority for me now is to get 4.0 into usable shape, so this will have to wait.

[agronholm]

I've created #906 to track the top-level feature.