add support for crontab expressions

README.rst

@@ -43,6 +43,7 @@ Usage
     schedule.every().wednesday.at("13:15").do(job)
     schedule.every().day.at("12:42", "Europe/Amsterdam").do(job)
     schedule.every().minute.at(":17").do(job)
+    schedule.every().crontab_expression("5 4 * * 2,5", "Europe/Amsterdam").do(job)
 
     def job_with_argument(name):
         print(f"I am {name}")

docs/examples.rst

@@ -42,6 +42,10 @@ Run a job every x minute
     schedule.every().monday.do(job)
     schedule.every().wednesday.at("13:15").do(job)
 
+    # Run job based on a crontab expression
+    schedule.every().crontab_expression("5 4 * * 2,5").do(job)
+    schedule.every().crontab_expression("5 4 * * 2,5", "Europe/Amsterdam").do(job)
+
     while True:
         schedule.run_pending()
         time.sleep(1)

requirements-dev.txt

@@ -9,4 +9,5 @@ black==20.8b1
 click==8.0.4
 mypy
 pytz
-types-pytz
+types-pytz
+freezegun

schedule/__init__.py

@@ -46,6 +46,9 @@
 import time
 from typing import Set, List, Optional, Callable, Union
 
+from .crontab import Crontab
+
+
 logger = logging.getLogger("schedule")
 
 
@@ -228,7 +231,7 @@ def __init__(self, interval: int, scheduler: Optional[Scheduler] = None):
         self.latest: Optional[int] = None  # upper limit to the interval
         self.job_func: Optional[functools.partial] = None  # the job job_func to run
 
-        # time units, e.g. 'minutes', 'hours', ...
+        # time units, e.g. 'minutes', 'hours', ... Only relevant when not using crontab
         self.unit: Optional[str] = None
 
         # optional time at which this job runs
@@ -237,6 +240,9 @@ def __init__(self, interval: int, scheduler: Optional[Scheduler] = None):
         # optional time zone of the self.at_time field. Only relevant when at_time is not None
         self.at_time_zone = None
 
+        # crontab
+        self.crontab: Optional[Crontab] = None
+
         # datetime of the last run
         self.last_run: Optional[datetime.datetime] = None
 
@@ -469,6 +475,18 @@ def tag(self, *tags: Hashable):
         self.tags.update(tags)
         return self
 
+    def crontab_expression(self, expression: str, tz: Optional[str] = None) -> "Job":
+        if self.unit is not None:
+            raise ScheduleValueError(
+                "crontab expression cannot be used if a unit is already defined")
+        if self.at_time is not None:
+            raise ScheduleValueError(
+                "crontab expression cannot be used if an `at` time is already defined")
+
+        c = Crontab.from_expression(expression, tz=tz)
+        self.crontab = c
+        return self
+
     def at(self, time_str: str, tz: Optional[str] = None):
 
         """
@@ -497,6 +515,11 @@ def at(self, time_str: str, tz: Optional[str] = None):
                 "Invalid unit (valid units are `days`, `hours`, and `minutes`)"
             )
 
+        if self.crontab is not None:
+            raise ScheduleValueError(
+                "`at` function cannot be used if a crontab expression is already defined"
+            )
+
         if tz is not None:
             import pytz
 
@@ -703,6 +726,13 @@ def _schedule_next_run(self) -> None:
         """
         Compute the instant when this job should run next.
         """
+        if self.crontab is not None:
+            self.next_run = self.crontab.next_run_time()
+            # the whole library lacks proper timezone support
+            # -> convert to unaware local time.
+            self.next_run = self.next_run.astimezone().replace(tzinfo=None)
+            return
+
         if self.unit not in ("seconds", "minutes", "hours", "days", "weeks"):
             raise ScheduleValueError(
                 "Invalid unit (valid units are `seconds`, `minutes`, `hours`, "