Import fixes for lack of datetime.timezone in python 2.7 (simple UTC)

Signed-off-by: Steve Arnold <[email protected]>

Author: sarnold

schedule/__init__.py

@@ -39,12 +39,19 @@
 """
 import collections
 import datetime
-from datetime import timezone
 import functools
 import logging
 import random
 import time
 
+try:
+    from datetime import timezone
+    utc = timezone.utc
+except ImportError:
+    from schedule.timezone import UTC
+    utc = UTC()
+
+
 logger = logging.getLogger('schedule')
 
 
@@ -150,7 +157,7 @@ def idle_seconds(self):
         :return: Number of seconds until
                  :meth:`next_run <Scheduler.next_run>`.
         """
-        return (self.next_run - datetime.datetime.now(timezone.utc)).total_seconds()
+        return (self.next_run - datetime.datetime.now(utc)).total_seconds()
 
 
 class Job(object):
@@ -397,7 +404,7 @@ def should_run(self):
         """
         :return: ``True`` if the job should be run now.
         """
-        return datetime.datetime.now(timezone.utc) >= self.next_run
+        return datetime.datetime.now(utc) >= self.next_run
 
     def run(self):
         """
@@ -407,7 +414,7 @@ def run(self):
         """
         logger.info('Running job %s', self)
         ret = self.job_func()
-        self.last_run = datetime.datetime.now(timezone.utc)
+        self.last_run = datetime.datetime.now(utc)
         self._schedule_next_run()
         return ret
 
@@ -424,7 +431,7 @@ def _schedule_next_run(self):
             interval = self.interval
 
         self.period = datetime.timedelta(**{self.unit: interval})
-        self.next_run = datetime.datetime.now(timezone.utc) + self.period
+        self.next_run = datetime.datetime.now(utc) + self.period
         if self.start_day is not None:
             assert self.unit == 'weeks'
             weekdays = (
@@ -455,15 +462,15 @@ def _schedule_next_run(self):
             # If we are running for the first time, make sure we run
             # at the specified time *today* (or *this hour*) as well
             if not self.last_run:
-                now = datetime.datetime.now(timezone.utc)
+                now = datetime.datetime.now(utc)
                 if (self.unit == 'days' and self.at_time > now.time() and
                         self.interval == 1):
                     self.next_run = self.next_run - datetime.timedelta(days=1)
                 elif self.unit == 'hours' and self.at_time.minute > now.minute:
                     self.next_run = self.next_run - datetime.timedelta(hours=1)
         if self.start_day is not None and self.at_time is not None:
             # Let's see if we will still make that time we specified today
-            if (self.next_run - datetime.datetime.now(timezone.utc)).days >= 7:
+            if (self.next_run - datetime.datetime.now(utc)).days >= 7:
                 self.next_run -= self.period
 
 

schedule/timezone.py

@@ -0,0 +1,18 @@
+import datetime
+
+
+class UTC(datetime.tzinfo):
+    """tzinfo derived concrete class named "UTC" with offset of 0"""
+    # can be configured here
+    _offset = datetime.timedelta(seconds=0)
+    _dst = datetime.timedelta(0)
+    _name = "UTC"
+
+    def utcoffset(self, dt):
+        return self.__class__._offset
+
+    def dst(self, dt):
+        return self.__class__._dst
+
+    def tzname(self, dt):
+        return self.__class__._name

test_schedule.py

@@ -11,6 +11,13 @@
 import schedule
 from schedule import every
 
+try:
+    from datetime import timezone
+    utc = timezone.utc
+except ImportError:
+    from schedule.timezone import UTC
+    utc = UTC()
+
 
 def make_mock_job(name=None):
     job = mock.Mock()
@@ -260,7 +267,7 @@ def test_next_run_property(self):
             assert len(schedule.jobs) == 2
             # Make sure the hourly job is first
             assert schedule.next_run() == original_datetime(2010, 1, 6, 14, 16,
-                                                            tzinfo=datetime.timezone.utc)
+                                                            tzinfo=utc)
             assert schedule.idle_seconds() == 60 * 60
 
     def test_cancel_job(self):