Adds option to control transition normalization behavior.

Author: sdispater

docs/index.rst

@@ -1008,6 +1008,31 @@ given timezone to properly handle any transition that might have occurred.
     '2013-10-27T02:30:00+01:00'
 
 
+.. versionadded:: 0.6
+
+    You can now control the normalization behavior:
+
+    .. code-block:: python
+
+        import pendulum
+
+        pendulum.set_transition_rule(pendulum.PRE_TRANSITION)
+
+        pendulum.create(2013, 3, 31, 2, 30, 0, 0, 'Europe/Paris')
+        '2013-03-31T02:30:00+01:00'
+        pendulum.create(2013, 10, 27, 2, 30, 0, 0, 'Europe/Paris')
+        '2013-10-27T02:30:00+02:00'
+
+        pendulum.set_transition_rule(pendulum.TRANSITION_ERROR)
+
+        pendulum.create(2013, 3, 31, 2, 30, 0, 0, 'Europe/Paris')
+        # NonExistingTime: The datetime 2013-03-31 02:30:00 does not exist
+        pendulum.create(2013, 10, 27, 2, 30, 0, 0, 'Europe/Paris')
+        # AmbiguousTime: The datetime 2013-10-27 02:30:00 is ambiguous.
+
+    Note that it only affects instances at creation time. Shifting time around
+    transition times still behaves the same.
+
 Shifting time to transition
 ---------------------------
 

pendulum/__init__.py

@@ -14,6 +14,12 @@
     SECONDS_PER_HOUR, SECONDS_PER_DAY
 )
 
+from .tz.timezone import Timezone
+
+PRE_TRANSITION = Timezone.PRE_TRANSITION
+POST_TRANSITION = Timezone.POST_TRANSITION
+TRANSITION_ERROR = Timezone.TRANSITION_ERROR
+
 # Helpers
 instance = Pendulum.instance
 parse = Pendulum.parse
@@ -38,6 +44,8 @@
 set_transaltor = Pendulum.set_translator
 set_to_string_format = Pendulum.set_to_string_format
 reset_to_string_format = Pendulum.reset_to_string_format
+set_transition_rule = Pendulum.set_transition_rule
+get_transition_rule = Pendulum.get_transition_rule
 
 # Standard helpers
 min = Pendulum.min

pendulum/pendulum.py

@@ -83,6 +83,8 @@ class Pendulum(datetime.datetime, TranslatableMixin):
 
     _MODIFIERS_VALID_UNITS = ['day', 'week', 'month', 'year', 'decade', 'century']
 
+    _TRANSITION_RULE = Timezone.POST_TRANSITION
+
     @classmethod
     def _safe_create_datetime_zone(cls, obj):
         """
@@ -170,7 +172,7 @@ def __init__(self, year, month, day,
             self._datetime = self._tz.convert(datetime.datetime(
                 year, month, day,
                 hour, minute, second, microsecond
-            ))
+            ), dst_rule=self._TRANSITION_RULE)
 
     @classmethod
     def instance(cls, dt, tz=UTC):
@@ -369,7 +371,7 @@ def create(cls, year=None, month=None, day=None,
         dt = datetime.datetime(*cls._create_datetime(
             tz, year, month, day, hour, minute, second, microsecond
         )[:-1])
-        dt = tz.convert(dt)
+        dt = tz.convert(dt, dst_rule=cls._TRANSITION_RULE)
 
         return cls.instance(dt)
 
@@ -799,6 +801,20 @@ def set_weekend_days(cls, value):
         """
         cls._weekend_days = value
 
+    # Normalization Rule
+    @classmethod
+    def set_transition_rule(cls, rule):
+        if rule not in [Timezone.PRE_TRANSITION,
+                        Timezone.POST_TRANSITION,
+                        Timezone.TRANSITION_ERROR]:
+            raise ValueError('Invalid transition rule: {}'.format(rule))
+
+        cls._TRANSITION_RULE = rule
+
+    @classmethod
+    def get_transition_rule(cls):
+        return cls._TRANSITION_RULE
+
     # Testing aids
 
     @classmethod

pendulum/tz/exceptions.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+
+
+class TimezoneError(ValueError):
+
+    pass
+
+
+class NonExistingTime(TimezoneError):
+
+    message = 'The datetime {} does not exist.'
+
+    def __init__(self, dt):
+        message = self.message.format(dt)
+
+        super(NonExistingTime, self).__init__(message)
+
+
+class AmbiguousTime(TimezoneError):
+
+    message = 'The datetime {} is ambiguous.'
+
+    def __init__(self, dt):
+        message = self.message.format(dt)
+
+        super(AmbiguousTime, self).__init__(message)

pendulum/tz/timezone.py

@@ -7,12 +7,17 @@
 from .timezone_info import TimezoneInfo, UTC
 from .breakdown import local_time as _local_time
 from .transition_type import TransitionType
+from .exceptions import NonExistingTime, AmbiguousTime
 
 
 class Timezone(object):
 
     _cache = {}
 
+    PRE_TRANSITION = 'pre'
+    POST_TRANSITION = 'post'
+    TRANSITION_ERROR = 'error'
+
     def __init__(self, name, transitions,
                  transition_types, default_transition_type):
         self._name = name
@@ -58,7 +63,7 @@ def load(cls, name):
 
         return cls._cache[name]
 
-    def convert(self, dt):
+    def convert(self, dt, dst_rule=POST_TRANSITION):
         """
         Converts or normalizes a datetime.
 
@@ -69,7 +74,7 @@ def convert(self, dt):
         """
         if dt.tzinfo is None:
             # we assume local time
-            converted = self._normalize(dt)
+            converted = self._normalize(dt, dst_rule=dst_rule)
 
         else:
             converted = self._convert(dt)
@@ -79,7 +84,7 @@ def convert(self, dt):
 
         return dt.__class__(*converted)
 
-    def _normalize(self, dt):
+    def _normalize(self, dt, dst_rule=POST_TRANSITION):
         # if tzinfo is set, something wrong happened
         if dt.tzinfo is not None:
             raise ValueError(
@@ -128,36 +133,58 @@ def _normalize(self, dt):
             else:
                 # tr.pre_time < dt < tr.time
                 # Skipped time
-                unix_time = tr.unix_time - (tr.pre_time - dt).total_seconds()
+                if dst_rule == self.TRANSITION_ERROR:
+                    raise NonExistingTime(dt)
+                elif dst_rule == self.PRE_TRANSITION:
+                    # We do not apply the transition
+                    (unix_time,
+                     transition_type) = self._get_previous_transition_time(tr, dt)
+                else:
+                    unix_time = tr.unix_time - (tr.pre_time - dt).total_seconds()
         elif tr is end:
             if tr.pre_time < dt:
                 # After the last transition.
                 unix_time = tr.unix_time + (dt - tr.time).total_seconds()
             else:
                 # tr.time <= dt <= tr.pre_time
                 # Repeated time
-                unix_time = tr.unix_time + (dt - tr.time).total_seconds()
+                if dst_rule == self.TRANSITION_ERROR:
+                    raise AmbiguousTime(dt)
+                elif dst_rule == self.PRE_TRANSITION:
+                    # We do not apply the transition
+                    (unix_time,
+                     transition_type) = self._get_previous_transition_time(tr, dt)
+                else:
+                    unix_time = tr.unix_time + (dt - tr.time).total_seconds()
         else:
             if tr.pre_time <= dt < tr.time:
                 # tr.pre_time <= dt < tr.time
                 # Skipped time
-                unix_time = tr.unix_time - (tr.pre_time - dt).total_seconds()
+                if dst_rule == self.TRANSITION_ERROR:
+                    raise NonExistingTime(dt)
+                elif dst_rule == self.PRE_TRANSITION:
+                    # We do not apply the transition
+                    (unix_time,
+                     transition_type) = self._get_previous_transition_time(tr, dt)
+                else:
+                    unix_time = tr.unix_time - (tr.pre_time - dt).total_seconds()
             elif tr.time <= dt <= tr.pre_time:
                 # tr.time <= dt <= tr.pre_time
                 # Repeated time
-                unix_time = tr.unix_time + (dt - tr.time).total_seconds()
+                if dst_rule == self.TRANSITION_ERROR:
+                    raise AmbiguousTime(dt)
+                elif dst_rule == self.PRE_TRANSITION:
+                    # We do not apply the transition
+                    (unix_time,
+                     transition_type) = self._get_previous_transition_time(tr, dt)
+                else:
+                    unix_time = tr.unix_time + (dt - tr.time).total_seconds()
             else:
                 # In between transitions
                 # The actual transition type is the previous transition one
 
-                # Fix for negative microseconds for negative timestamps
-                diff = (dt - tr.pre_time).total_seconds()
-                if -1 < diff < 0 and tr.unix_time < 0:
-                    diff -= 1
-
-                unix_time = tr.unix_time + diff
-
-                transition_type = tr.pre_transition_type
+                (unix_time,
+                 transition_type) = self._get_previous_transition_time(tr, dt)
 
         return self._to_local_time(unix_time, transition_type)
 
@@ -232,6 +259,17 @@ def _find_transition_index(self, dt, prop='_time'):
 
         return lo
 
+    def _get_previous_transition_time(self, tr, dt):
+        diff = (dt - tr.pre_time).total_seconds()
+        if -1 < diff < 0 and tr.unix_time < 0:
+            diff -= 1
+
+        unix_time = tr.unix_time + diff
+
+        transition_type = tr.pre_transition_type
+
+        return unix_time, transition_type
+
     def __repr__(self):
         return '<Timezone [{}]>'.format(self._name)
 

tests/__init__.py

@@ -4,7 +4,7 @@
 from contextlib import contextmanager
 
 from pendulum import Pendulum, Interval
-from pendulum.tz import LocalTimezone, timezone
+from pendulum.tz import LocalTimezone, timezone, Timezone
 
 
 class AbstractTestCase(TestCase):
@@ -17,6 +17,7 @@ def setUp(self):
     def tearDown(self):
         LocalTimezone.set_local_timezone()
         Pendulum.reset_to_string_format()
+        Pendulum.set_transition_rule(Timezone.POST_TRANSITION)
 
     def assertPendulum(self, d, year, month, day,
                        hour=None, minute=None, second=None, microsecond=None):

tests/tz_tests/test_timezone.py

@@ -3,6 +3,7 @@
 import pendulum
 from datetime import datetime
 from pendulum import timezone
+from pendulum.tz.exceptions import NonExistingTime, AmbiguousTime
 
 from .. import AbstractTestCase
 
@@ -41,6 +42,27 @@ def test_skipped_time(self):
         self.assertEqual(7200, dt.tzinfo.offset)
         self.assertTrue(dt.tzinfo.is_dst)
 
+    def test_skipped_time_with_pre_rule(self):
+        dt = datetime(2013, 3, 31, 2, 30, 45, 123456)
+        tz = timezone('Europe/Paris')
+        dt = tz.convert(dt, dst_rule=tz.PRE_TRANSITION)
+
+        self.assertEqual(2013, dt.year)
+        self.assertEqual(3, dt.month)
+        self.assertEqual(31, dt.day)
+        self.assertEqual(2, dt.hour)
+        self.assertEqual(30, dt.minute)
+        self.assertEqual(45, dt.second)
+        self.assertEqual(123456, dt.microsecond)
+        self.assertEqual('Europe/Paris', dt.tzinfo.tz.name)
+        self.assertEqual(3600, dt.tzinfo.offset)
+        self.assertFalse(dt.tzinfo.is_dst)
+
+    def test_skipped_time_with_error(self):
+        dt = datetime(2013, 3, 31, 2, 30, 45, 123456)
+        tz = timezone('Europe/Paris')
+        self.assertRaises(NonExistingTime, tz.convert, dt, tz.TRANSITION_ERROR)
+
     def test_repeated_time(self):
         dt = datetime(2013, 10, 27, 2, 30, 45, 123456)
         tz = timezone('Europe/Paris')
@@ -57,6 +79,27 @@ def test_repeated_time(self):
         self.assertEqual(3600, dt.tzinfo.offset)
         self.assertFalse(dt.tzinfo.is_dst)
 
+    def test_repeated_time_pre_rule(self):
+        dt = datetime(2013, 10, 27, 2, 30, 45, 123456)
+        tz = timezone('Europe/Paris')
+        dt = tz.convert(dt, dst_rule=tz.PRE_TRANSITION)
+
+        self.assertEqual(2013, dt.year)
+        self.assertEqual(10, dt.month)
+        self.assertEqual(27, dt.day)
+        self.assertEqual(2, dt.hour)
+        self.assertEqual(30, dt.minute)
+        self.assertEqual(45, dt.second)
+        self.assertEqual(123456, dt.microsecond)
+        self.assertEqual('Europe/Paris', dt.tzinfo.tz.name)
+        self.assertEqual(7200, dt.tzinfo.offset)
+        self.assertTrue(dt.tzinfo.is_dst)
+
+    def test_repeated_time_with_error(self):
+        dt = datetime(2013, 10, 27, 2, 30, 45, 123456)
+        tz = timezone('Europe/Paris')
+        self.assertRaises(AmbiguousTime, tz.convert, dt, tz.TRANSITION_ERROR)
+
     def test_pendulum_create_basic(self):
         dt = pendulum.create(2016, 6, 1, 12, 34, 56, 123456, 'Europe/Paris')
 
@@ -235,3 +278,34 @@ def test_convert_accept_pendulum_instance(self):
 
         self.assertIsInstanceOfPendulum(new)
         self.assertPendulum(new, 2016, 8, 7, 14, 53, 54)
+
+    def test_add_time_to_new_transition_does_not_use_transition_rule(self):
+        pendulum.set_transition_rule(pendulum.TRANSITION_ERROR)
+        dt = pendulum.create(2013, 3, 31, 1, 59, 59, 999999, 'Europe/Paris')
+
+        self.assertPendulum(dt, 2013, 3, 31, 1, 59, 59, 999999)
+        self.assertEqual('Europe/Paris', dt.timezone_name)
+        self.assertEqual(3600, dt.offset)
+        self.assertFalse(dt.is_dst)
+
+        dt = dt.add(microseconds=1)
+
+        self.assertPendulum(dt, 2013, 3, 31, 3, 0, 0, 0)
+        self.assertEqual('Europe/Paris', dt.timezone_name)
+        self.assertEqual(7200, dt.offset)
+        self.assertTrue(dt.is_dst)
+
+    def test_create_uses_transition_rule(self):
+        pendulum.set_transition_rule(pendulum.PRE_TRANSITION)
+        dt = pendulum.create(2013, 3, 31, 2, 30, 45, 123456, 'Europe/Paris')
+
+        self.assertEqual(2013, dt.year)
+        self.assertEqual(3, dt.month)
+        self.assertEqual(31, dt.day)
+        self.assertEqual(2, dt.hour)
+        self.assertEqual(30, dt.minute)
+        self.assertEqual(45, dt.second)
+        self.assertEqual(123456, dt.microsecond)
+        self.assertEqual('Europe/Paris', dt.timezone_name)
+        self.assertEqual(3600, dt.offset)
+        self.assertFalse(dt.is_dst)