[AIRFLOW-1804] Add time zone configuration options

Time zone defaults to UTC as is the default now in order
to maintain backwards compatibility.

airflow/config_templates/default_airflow.cfg

@@ -54,6 +54,10 @@ logging_config_class =
 log_format = [%%(asctime)s] {{%%(filename)s:%%(lineno)d}} %%(levelname)s - %%(message)s
 simple_log_format = %%(asctime)s %%(levelname)s - %%(message)s
 
+# Default timezone in case supplied date times are naive
+# can be utc (default), system, or any IANA timezone string (e.g. Europe/Amsterdam)
+default_timezone = utc
+
 # The executor class that airflow should use. Choices include
 # SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor
 executor = SequentialExecutor
@@ -364,12 +368,12 @@ authenticate = False
 
 [ldap]
 # set this to ldaps://<your.ldap.server>:<port>
-uri = 
+uri =
 user_filter = objectClass=*
 user_name_attr = uid
 group_member_attr = memberOf
-superuser_filter = 
-data_profiler_filter = 
+superuser_filter =
+data_profiler_filter =
 bind_user = cn=Manager,dc=example,dc=com
 bind_password = insecure
 basedn = dc=example,dc=com

airflow/settings.py

@@ -19,6 +19,8 @@ from __future__ import unicode_literals
 
 import logging
 import os
+import pendulum
+
 from sqlalchemy import create_engine
 from sqlalchemy.orm import scoped_session, sessionmaker
 from sqlalchemy.pool import NullPool
@@ -29,6 +31,18 @@ from airflow.logging_config import configure_logging
 log = logging.getLogger(__name__)
 
 
+TIMEZONE = pendulum.timezone('UTC')
+try:
+    tz = conf.get("core", "default_timezone")
+    if tz == "system":
+        TIMEZONE = pendulum.local_timezone()
+    else:
+        TIMEZONE = pendulum.timezone(tz)
+except:
+    pass
+log.info("Configured default timezone %s" % TIMEZONE)
+
+
 class DummyStatsLogger(object):
     @classmethod
     def incr(cls, stat, count=1, rate=1):

airflow/utils/timezone.py

@@ -0,0 +1,68 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import pendulum
+
+from airflow.settings import TIMEZONE
+
+
+# UTC time zone as a tzinfo instance.
+utc = pendulum.timezone('UTC')
+
+
+def is_localized(value):
+    """
+    Determine if a given datetime.datetime is aware.
+    The concept is defined in Python's docs:
+    http://docs.python.org/library/datetime.html#datetime.tzinfo
+    Assuming value.tzinfo is either None or a proper datetime.tzinfo,
+    value.utcoffset() implements the appropriate logic.
+    """
+    return value.utcoffset() is not None
+
+
+def is_naive(value):
+    """
+    Determine if a given datetime.datetime is naive.
+    The concept is defined in Python's docs:
+    http://docs.python.org/library/datetime.html#datetime.tzinfo
+    Assuming value.tzinfo is either None or a proper datetime.tzinfo,
+    value.utcoffset() implements the appropriate logic.
+    """
+    return value.utcoffset() is None
+
+
+def utcnow():
+    """
+    Get the current date and time in UTC
+    :return:
+    """
+
+    return pendulum.utcnow()
+
+
+def convert_to_utc(value):
+    """
+    Returns the datetime with the default timezone added if timezone
+    information was not associated
+    :param value: datetime
+    :return: datetime with tzinfo
+    """
+    if not value:
+        return value
+
+    if not is_localized(value):
+        value = pendulum.instance(value, TIMEZONE)
+
+    return value.astimezone(utc)

setup.py

@@ -226,6 +226,7 @@ def do_setup():
             'lxml>=3.6.0, <4.0',
             'markdown>=2.5.2, <3.0',
             'pandas>=0.17.1, <1.0.0',
+            'pendulum==1.3.1',
             'psutil>=4.2.0, <5.0.0',
             'pygments>=2.0.1, <3.0',
             'python-daemon>=2.1.1, <2.2',
@@ -236,6 +237,7 @@ def do_setup():
             'sqlalchemy>=0.9.8',
             'tabulate>=0.7.5, <0.8.0',
             'thrift>=0.9.2',
+            'tzlocal>=1.4',
             'zope.deprecation>=4.0, <5.0',
         ],
         setup_requires=[

tests/utils/test_timezone.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+import pendulum
+import unittest
+
+from airflow.utils import timezone
+
+CET = pendulum.timezone("Europe/Paris")
+EAT = pendulum.timezone('Africa/Nairobi')      # Africa/Nairobi
+ICT = pendulum.timezone('Asia/Bangkok')      # Asia/Bangkok
+UTC = timezone.utc
+
+
+class TimezoneTest(unittest.TestCase):
+    def test_is_aware(self):
+        self.assertTrue(timezone.is_localized(datetime.datetime(2011, 9, 1, 13, 20, 30, tzinfo=EAT)))
+        self.assertFalse(timezone.is_localized(datetime.datetime(2011, 9, 1, 13, 20, 30)))
+
+    def test_is_naive(self):
+        self.assertFalse(timezone.is_naive(datetime.datetime(2011, 9, 1, 13, 20, 30, tzinfo=EAT)))
+        self.assertTrue(timezone.is_naive(datetime.datetime(2011, 9, 1, 13, 20, 30)))
+
+    def test_utcnow(self):
+        now = timezone.utcnow()
+        self.assertTrue(timezone.is_localized(now))
+        self.assertEquals(now.replace(tzinfo=None), now.astimezone(UTC).replace(tzinfo=None))
+
+    def test_convert_to_utc(self):
+        naive = datetime.datetime(2011, 9, 1, 13, 20, 30)
+        utc = datetime.datetime(2011, 9, 1, 13, 20, 30, tzinfo=UTC)
+        self.assertEquals(utc, timezone.convert_to_utc(naive))
+
+        eat = datetime.datetime(2011, 9, 1, 13, 20, 30, tzinfo=EAT)
+        utc = datetime.datetime(2011, 9, 1, 10, 20, 30, tzinfo=UTC)
+        self.assertEquals(utc, timezone.convert_to_utc(eat))