agronholm / agronholm/apscheduler

Scheduling Support for Statutory Holidays

Offen
#520 2 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen
enhancement
Vorherrschende Sprache
Python
Sterne
7.6k
Forks
783
Ø Merge
4 T. 8 Std.
Gemergte PRs (30 T.)
5

Beschreibung

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

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

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

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

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

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

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!

Beitragsleitfaden

Beitragsleitfaden öffnen

Bewertung

Dieses Issue wurde noch nicht bewertet.

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.