Support timzones in cron

rq_scheduler/scheduler.py

@@ -8,6 +8,7 @@
 from datetime import datetime
 from itertools import repeat
 
+import dateutil
 from rq.exceptions import NoSuchJobError
 from rq.job import Job
 from rq.queue import Queue
@@ -248,6 +249,7 @@ def schedule(self, scheduled_time, func, args=None, kwargs=None,
                                timeout=timeout, meta=meta, depends_on=depends_on,
                                on_success=on_success, on_failure=on_failure)
 
+        job.meta['timezone'] = scheduled_time.tzinfo
         if interval is not None:
             job.meta['interval'] = int(interval)
         if repeat is not None:
@@ -260,12 +262,12 @@ def schedule(self, scheduled_time, func, args=None, kwargs=None,
         return job
 
     def cron(self, cron_string, func, args=None, kwargs=None, repeat=None,
-             queue_name=None, id=None, timeout=None, description=None, meta=None, use_local_timezone=False,
-             depends_on=None, on_success=None, on_failure=None):
+             queue_name=None, id=None, timeout=None, description=None, meta=None,
+             timezone=dateutil.tz.UTC, depends_on=None, on_success=None, on_failure=None):
         """
         Schedule a cronjob
         """
-        scheduled_time = get_next_scheduled_time(cron_string, use_local_timezone=use_local_timezone)
+        scheduled_time = get_next_scheduled_time(cron_string, timezone=timezone)
 
         # Set result_ttl to -1, as jobs scheduled via cron are periodic ones.
         # Otherwise the job would expire after 500 sec.
@@ -275,7 +277,7 @@ def cron(self, cron_string, func, args=None, kwargs=None, repeat=None,
                                on_success=on_success, on_failure=on_failure)
 
         job.meta['cron_string'] = cron_string
-        job.meta['use_local_timezone'] = use_local_timezone
+        job.meta['timezone'] = timezone
 
         if repeat is not None:
             job.meta['repeat'] = int(repeat)
@@ -403,7 +405,7 @@ def enqueue_job(self, job):
         interval = job.meta.get('interval', None)
         repeat = job.meta.get('repeat', None)
         cron_string = job.meta.get('cron_string', None)
-        use_local_timezone = job.meta.get('use_local_timezone', None)
+        timezone = job.meta.get('timezone', dateutil.tz.UTC)
 
         # If job is a repeated job, decrement counter
         if repeat:
@@ -419,13 +421,13 @@ def enqueue_job(self, job):
                 if job.meta['repeat'] == 0:
                     return
             self.connection.zadd(self.scheduled_jobs_key,
-                                  {job.id: to_unix(datetime.utcnow()) + int(interval)})
+                                  {job.id: to_unix(datetime.now(tz=timezone)) + int(interval)})
         elif cron_string:
             # If this is a repeat job and counter has reached 0, don't repeat
             if repeat is not None:
                 if job.meta['repeat'] == 0:
                     return
-            next_scheduled_time = get_next_scheduled_time(cron_string, use_local_timezone=use_local_timezone)
+            next_scheduled_time = get_next_scheduled_time(cron_string, timezone=timezone)
             self.connection.zadd(self.scheduled_jobs_key,
                                  {job.id: to_unix(next_scheduled_time)})
 
@@ -438,7 +440,7 @@ def enqueue_jobs(self):
         jobs = self.get_jobs_to_queue()
         for job in jobs:
             self.enqueue_job(job)
-        
+
         return jobs
 
     def heartbeat(self):

rq_scheduler/utils.py

@@ -20,14 +20,13 @@ def to_unix(dt):
     return calendar.timegm(dt.utctimetuple())
 
 
-def get_next_scheduled_time(cron_string, use_local_timezone=False):
+def get_next_scheduled_time(cron_string, timezone=dateutil.tz.UTC):
     """Calculate the next scheduled time by creating a crontab object
     with a cron string"""
     now = datetime.now()
     cron = crontab.CronTab(cron_string)
     next_time = cron.next(now=now, return_datetime=True)
-    tz = dateutil.tz.tzlocal() if use_local_timezone else dateutil.tz.UTC
-    return next_time.astimezone(tz)
+    return next_time.astimezone(timezone)
 
 
 def setup_loghandlers(level='INFO'):

setup.py

@@ -4,7 +4,7 @@
 
 setup(
     name='rq-scheduler',
-    version='0.11.0',
+    version='0.12.0',
     author='Selwin Ong',
     author_email='[email protected]',
     packages=['rq_scheduler'],

tests/test_scheduler.py

@@ -1,6 +1,7 @@
 import os
 import signal
 import time
+import dateutil.tz
 from datetime import datetime
 from datetime import timedelta
 from threading import Thread
@@ -511,25 +512,27 @@ def test_crontab_persisted_correctly(self):
         assert datetime_time.second == 0
         assert datetime_time - datetime.utcnow() < timedelta(hours=1)
 
-    def test_crontab_persisted_correctly_with_local_timezone(self):
+    def test_crontab_persisted_correctly_with_timezone(self):
         """
         Ensure that crontab attribute gets correctly saved in Redis when using local TZ.
         """
         # create a job that runs one minute past each whole hour
-        job = self.scheduler.cron("0 15 * * *", say_hello, use_local_timezone=True)
+        timezone = dateutil.tz.gettz('Europe/Kiev')
+        job = self.scheduler.cron("0 15 * * *", say_hello, timezone=timezone)
         job_from_queue = Job.fetch(job.id, connection=self.testconn)
         self.assertEqual(job_from_queue.meta['cron_string'], "0 15 * * *")
+        self.assertEqual(job_from_queue.meta['timezone'], timezone)
 
         # get the scheduled_time and convert it to a datetime object
         unix_time = self.testconn.zscore(self.scheduler.scheduled_jobs_key, job.id)
         datetime_time = from_unix(unix_time)
 
-        expected_datetime_in_local_tz = datetime.now(tzlocal()).replace(hour=15,minute=0,second=0,microsecond=0)
-        assert datetime_time.time() == expected_datetime_in_local_tz.astimezone(UTC).time()
+        assert datetime_time.time() == datetime.utcnow().replace(hour=12, minute=0, second=0, microsecond=0).time()
 
-    def test_crontab_rescheduled_correctly_with_local_timezone(self):
+    def test_crontab_rescheduled_correctly_with_timezone(self):
         # Create a job with a cronjob_string
-        job = self.scheduler.cron("1 15 * * *", say_hello, use_local_timezone=True)
+        timezone = dateutil.tz.gettz('Europe/Kiev')
+        job = self.scheduler.cron("1 15 * * *", say_hello, timezone=timezone)
 
         # change crontab
         job.meta['cron_string'] = "2 15 * * *"
@@ -541,8 +544,7 @@ def test_crontab_rescheduled_correctly_with_local_timezone(self):
         unix_time = self.testconn.zscore(self.scheduler.scheduled_jobs_key, job.id)
         datetime_time = from_unix(unix_time)
 
-        expected_datetime_in_local_tz = datetime.now(tzlocal()).replace(hour=15,minute=2,second=0,microsecond=0)
-        assert datetime_time.time() == expected_datetime_in_local_tz.astimezone(UTC).time()
+        assert datetime_time.time() == datetime.utcnow().replace(hour=12, minute=2, second=0, microsecond=0).time()
 
     def test_crontab_schedules_correctly(self):
         # Create a job with a cronjob_string
@@ -609,7 +611,7 @@ def test_job_with_intervals_get_rescheduled(self):
         """
         Ensure jobs with interval attribute are put back in the scheduler
         """
-        time_now = datetime.utcnow()
+        time_now = datetime.now(tz=dateutil.tz.gettz('Europe/Kiev'))
         interval = 10
         job = self.scheduler.schedule(time_now, say_hello, interval=interval)
         self.scheduler.enqueue_job(job)
@@ -711,8 +713,8 @@ def test_periodic_jobs_sets_meta(self):
         Ensure periodic jobs sets correctly meta.
         """
         meta = {'say': 'hello'}
-        job = self.scheduler.schedule(datetime.utcnow(), say_hello, interval=5, meta=meta)
-        self.assertEqual(meta, job.meta)
+        job = self.scheduler.schedule(datetime.now(tz=dateutil.tz.UTC), say_hello, interval=5, meta=meta)
+        self.assertEqual({'timezone': dateutil.tz.UTC, 'interval': 5, 'say': 'hello'}, job.meta)
 
     def test_periodic_job_sets_id(self):
         """