[AIRFLOW-1826] Update views to use timezone aware objects
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/518a41ac Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/518a41ac Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/518a41ac Branch: refs/heads/master Commit: 518a41acf319af27d49bdc0c84bda64b6b8af0b3 Parents: f43c0e9 Author: Bolke de Bruin <[email protected]> Authored: Thu Nov 16 18:54:56 2017 +0100 Committer: Bolke de Bruin <[email protected]> Committed: Mon Nov 27 15:54:27 2017 +0100 ---------------------------------------------------------------------- airflow/utils/dates.py | 16 ++++++----- airflow/utils/timezone.py | 2 +- airflow/www/views.py | 32 +++++++++++++++++++-- scripts/ci/requirements.txt | 1 + setup.py | 2 +- tests/contrib/operators/test_druid_operator.py | 6 ++-- tests/www/test_views.py | 2 +- 7 files changed, 46 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/518a41ac/airflow/utils/dates.py ---------------------------------------------------------------------- diff --git a/airflow/utils/dates.py b/airflow/utils/dates.py index cb9c840..2ca2b2c 100644 --- a/airflow/utils/dates.py +++ b/airflow/utils/dates.py @@ -73,16 +73,17 @@ def date_range( if isinstance(delta, six.string_types): delta_iscron = True tz = start_date.tzinfo - timezone.make_naive(start_date, tz) + start_date = timezone.make_naive(start_date, tz) cron = croniter(delta, start_date) elif isinstance(delta, timedelta): delta = abs(delta) l = [] if end_date: while start_date <= end_date: - if delta_iscron: - start_date = timezone.make_aware(start_date, tz) - l.append(start_date) + if timezone.is_naive(start_date): + l.append(timezone.make_aware(start_date, tz)) + else: + l.append(start_date) if delta_iscron: start_date = cron.get_next(datetime) @@ -90,9 +91,10 @@ def date_range( start_date += delta else: for _ in range(abs(num)): - if delta_iscron: - start_date = timezone.make_aware(start_date, tz) - l.append(start_date) + if timezone.is_naive(start_date): + l.append(timezone.make_aware(start_date, tz)) + else: + l.append(start_date) if delta_iscron: if num > 0: http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/518a41ac/airflow/utils/timezone.py ---------------------------------------------------------------------- diff --git a/airflow/utils/timezone.py b/airflow/utils/timezone.py index e384a14..75c8454 100644 --- a/airflow/utils/timezone.py +++ b/airflow/utils/timezone.py @@ -100,7 +100,7 @@ def make_aware(value, timezone=None): return timezone.convert(value) else: # This may be wrong around DST changes! - return value.astimezone(tz=timezone) + return value.replace(tzinfo=timezone) def make_naive(value, timezone=None): http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/518a41ac/airflow/www/views.py ---------------------------------------------------------------------- diff --git a/airflow/www/views.py b/airflow/www/views.py index 5ecee42..6bcb66d 100644 --- a/airflow/www/views.py +++ b/airflow/www/views.py @@ -16,6 +16,7 @@ from past.builtins import basestring, unicode import ast +import datetime as dt import logging import os import pkg_resources @@ -54,7 +55,9 @@ import markdown import nvd3 from wtforms import ( - Form, SelectField, TextAreaField, PasswordField, StringField, validators) + Form, SelectField, TextAreaField, PasswordField, + StringField, validators) +from flask_admin.form.fields import DateTimeField from pygments import highlight, lexers from pygments.formatters import HtmlFormatter @@ -160,6 +163,13 @@ def state_token(state): '{state}</span>'.format(**locals())) +def parse_datetime_f(value): + if not isinstance(value, dt.datetime): + return value + + return timezone.make_aware(value) + + def state_f(v, c, m, p): return state_token(m.state) @@ -1161,7 +1171,7 @@ class Airflow(BaseView): num_runs = int(num_runs) if num_runs else 25 if base_date: - base_date = pendulum.parse(base_date) + base_date = timezone.parse(base_date) else: base_date = dag.latest_execution_date or timezone.utcnow() @@ -2217,12 +2227,18 @@ class KnownEventView(wwwutils.DataProfilingMixin, AirflowModelView): 'validators': [ validators.DataRequired(), ], + 'filters': [ + parse_datetime_f, + ], }, 'end_date': { 'validators': [ validators.DataRequired(), GreaterEqualThan(fieldname='start_date'), ], + 'filters': [ + parse_datetime_f, + ] }, 'reported_by': { 'validators': [ @@ -2240,11 +2256,14 @@ class KnownEventView(wwwutils.DataProfilingMixin, AirflowModelView): column_default_sort = ("start_date", True) column_sortable_list = ( 'label', + # todo: yes this has a spelling error ('event_type', 'event_type.know_event_type'), 'start_date', 'end_date', ('reported_by', 'reported_by.username'), ) + filter_converter = wwwutils.UtcFilterConverter() + form_overrides = dict(start_date=DateTimeField, end_date=DateTimeField) class KnownEventTypeView(wwwutils.DataProfilingMixin, AirflowModelView): @@ -2349,9 +2368,18 @@ class XComView(wwwutils.SuperUserMixin, AirflowModelView): 'value': StringField('Value'), } + form_args = { + 'execution_date': { + 'filters': [ + parse_datetime_f, + ] + } + } + column_filters = ('key', 'timestamp', 'execution_date', 'task_id', 'dag_id') column_searchable_list = ('key', 'timestamp', 'execution_date', 'task_id', 'dag_id') filter_converter = wwwutils.UtcFilterConverter() + form_overrides = dict(execution_date=DateTimeField) class JobModelView(ModelViewOnly): http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/518a41ac/scripts/ci/requirements.txt ---------------------------------------------------------------------- diff --git a/scripts/ci/requirements.txt b/scripts/ci/requirements.txt index 1ab69a1..2b5a8c9 100644 --- a/scripts/ci/requirements.txt +++ b/scripts/ci/requirements.txt @@ -64,6 +64,7 @@ pandas pandas-gbq parameterized paramiko>=2.1.1 +pendulum>=1.3.2 psutil>=4.2.0, <5.0.0 psycopg2 pygments http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/518a41ac/setup.py ---------------------------------------------------------------------- diff --git a/setup.py b/setup.py index cfe0d92..4e616d7 100644 --- a/setup.py +++ b/setup.py @@ -227,7 +227,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', + 'pendulum==1.3.2', 'psutil>=4.2.0, <5.0.0', 'pygments>=2.0.1, <3.0', 'python-daemon>=2.1.1, <2.2', http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/518a41ac/tests/contrib/operators/test_druid_operator.py ---------------------------------------------------------------------- diff --git a/tests/contrib/operators/test_druid_operator.py b/tests/contrib/operators/test_druid_operator.py index 9df6c48..c8f92f5 100644 --- a/tests/contrib/operators/test_druid_operator.py +++ b/tests/contrib/operators/test_druid_operator.py @@ -13,15 +13,15 @@ # limitations under the License. # -import datetime import mock import unittest from airflow import DAG, configuration from airflow.contrib.operators.druid_operator import DruidOperator +from airflow.utils import timezone from airflow.models import TaskInstance -DEFAULT_DATE = datetime.datetime(2017, 1, 1) +DEFAULT_DATE = timezone.datetime(2017, 1, 1) class TestDruidOperator(unittest.TestCase): @@ -29,7 +29,7 @@ class TestDruidOperator(unittest.TestCase): configuration.load_test_config() args = { 'owner': 'airflow', - 'start_date': datetime.datetime(2017, 1, 1) + 'start_date': timezone.datetime(2017, 1, 1) } self.dag = DAG('test_dag_id', default_args=args) http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/518a41ac/tests/www/test_views.py ---------------------------------------------------------------------- diff --git a/tests/www/test_views.py b/tests/www/test_views.py index a9bb28f..f5b015e 100644 --- a/tests/www/test_views.py +++ b/tests/www/test_views.py @@ -374,7 +374,7 @@ class TestLogView(unittest.TestCase): follow_redirects=True, ) self.assertEqual(response.status_code, 200) - self.assertIn('<pre id="attempt-1">*** Reading local log.\nLog for testing.\n</pre>', + self.assertIn('Log file isn', response.data.decode('utf-8'))
