diff --git a/src/apscheduler/triggers/cron/expressions.py b/src/apscheduler/triggers/cron/expressions.py index 71966b47..e04f0a72 100644 --- a/src/apscheduler/triggers/cron/expressions.py +++ b/src/apscheduler/triggers/cron/expressions.py @@ -251,3 +251,20 @@ def get_next_value(self, dateval: datetime, field: BaseField) -> int | None: def __str__(self) -> str: return "last" + + +class LastNDayOfMonthExpression(AllExpression): + value_re = re.compile(r"last-(?P[0-9]+)", re.IGNORECASE) + + def __init__(self, last_day: str): + super().__init__(None) + self.last_day = as_int(last_day) + + def get_next_value(self, dateval: datetime, field: BaseField) -> int | None: + currval = field.get_value(dateval) + nextval = monthrange(dateval.year, dateval.month)[1] - self.last_day + + return nextval if currval <= nextval else None + + def __str__(self) -> str: + return f"last-{self.last_day}" diff --git a/src/apscheduler/triggers/cron/fields.py b/src/apscheduler/triggers/cron/fields.py index 4a2e44b8..a777fa62 100644 --- a/src/apscheduler/triggers/cron/fields.py +++ b/src/apscheduler/triggers/cron/fields.py @@ -14,6 +14,7 @@ WEEKDAYS, AllExpression, LastDayOfMonthExpression, + LastNDayOfMonthExpression, MonthRangeExpression, RangeExpression, WeekdayPositionExpression, @@ -122,7 +123,12 @@ def get_value(self, dateval: datetime) -> int: class DayOfMonthField( - BaseField, extra_compilers=(WeekdayPositionExpression, LastDayOfMonthExpression) + BaseField, + extra_compilers=( + WeekdayPositionExpression, + LastNDayOfMonthExpression, + LastDayOfMonthExpression, + ), ): __slots__ = () diff --git a/tests/triggers/test_cron.py b/tests/triggers/test_cron.py index 58b46291..ae1661b1 100644 --- a/tests/triggers/test_cron.py +++ b/tests/triggers/test_cron.py @@ -160,6 +160,22 @@ def test_cron_trigger_4(timezone, serializer): ) +def test_cron_trigger_5(timezone, serializer): + start_time = datetime(2012, 2, 1, tzinfo=timezone) + trigger = CronTrigger( + year="2012", month="2", day="last-6", start_time=start_time, timezone=timezone + ) + if serializer: + trigger = serializer.deserialize(serializer.serialize(trigger)) + + assert trigger.next() == datetime(2012, 2, 23, tzinfo=timezone) + assert repr(trigger) == ( + "CronTrigger(year='2012', month='2', day='last-6', week='*', " + "day_of_week='*', hour='0', minute='0', second='0', " + "start_time='2012-02-01T00:00:00+01:00', timezone='Europe/Berlin')" + ) + + @pytest.mark.parametrize("expr", ["3-5", "wed-fri"], ids=["numeric", "text"]) def test_weekday_overlap(timezone, serializer, expr): start_time = datetime(2009, 1, 1, tzinfo=timezone)