You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by bo...@apache.org on 2017/03/13 04:45:15 UTC
[17/45] incubator-airflow git commit: [AIRFLOW-365] Set dag.fileloc
explicitly and use for Code view
[AIRFLOW-365] Set dag.fileloc explicitly and use for Code view
Code view for subdag has not been working. I do
not think we are able
cleanly figure out where the code for the factory
method lives when we
process the dags, so we need to save the location
when the subdag is
created.
Previously for a subdag, its `fileloc` attribute
would be set to the
location of the parent dag. I think it is
appropriate to instead set
it to the actual child dag location instead. We do
not lose any
information this way (we still have the link to
the parent dag that
has its location) and now we can always read this
attribute for the
code view. This should not affect the use of this
field for refreshing
dags, because we always refresh the parent for a
subdag.
Closes #2043 from dhuang/AIRFLOW-365
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/a7abcf35
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/a7abcf35
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/a7abcf35
Branch: refs/heads/v1-8-stable
Commit: a7abcf35b0e228034f746b3d50abd0ca9bd8bede
Parents: 4db8f07
Author: Daniel Huang <dx...@gmail.com>
Authored: Thu Feb 2 13:57:20 2017 +0100
Committer: Bolke de Bruin <bo...@xs4all.nl>
Committed: Sun Mar 12 07:54:02 2017 -0700
----------------------------------------------------------------------
airflow/models.py | 7 ++++---
airflow/www/views.py | 5 ++---
tests/models.py | 18 ++++++++++++++++++
3 files changed, 24 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/a7abcf35/airflow/models.py
----------------------------------------------------------------------
diff --git a/airflow/models.py b/airflow/models.py
index 62457f0..d6ab5b8 100755
--- a/airflow/models.py
+++ b/airflow/models.py
@@ -29,6 +29,7 @@ import functools
import getpass
import imp
import importlib
+import inspect
import zipfile
import jinja2
import json
@@ -307,7 +308,6 @@ class DagBag(BaseDagBag, LoggingMixin):
if not dag.full_filepath:
dag.full_filepath = filepath
dag.is_subdag = False
- dag.module_name = m.__name__
self.bag_dag(dag, parent_dag=dag, root_dag=dag)
found_dags.append(dag)
found_dags += dag.subdags
@@ -367,7 +367,6 @@ class DagBag(BaseDagBag, LoggingMixin):
for subdag in dag.subdags:
subdag.full_filepath = dag.full_filepath
subdag.parent_dag = dag
- subdag.fileloc = root_dag.full_filepath
subdag.is_subdag = True
self.bag_dag(subdag, parent_dag=dag, root_dag=root_dag)
self.logger.debug('Loaded DAG {dag}'.format(**locals()))
@@ -2660,6 +2659,8 @@ class DAG(BaseDag, LoggingMixin):
self._pickle_id = None
self._description = description
+ # set file location to caller source path
+ self.fileloc = inspect.getsourcefile(inspect.stack()[1][0])
self.task_dict = dict()
self.start_date = start_date
self.end_date = end_date
@@ -3355,7 +3356,7 @@ class DAG(BaseDag, LoggingMixin):
orm_dag = DagModel(dag_id=dag.dag_id)
logging.info("Creating ORM DAG for %s",
dag.dag_id)
- orm_dag.fileloc = dag.full_filepath
+ orm_dag.fileloc = dag.fileloc
orm_dag.is_subdag = dag.is_subdag
orm_dag.owners = owner
orm_dag.is_active = True
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/a7abcf35/airflow/www/views.py
----------------------------------------------------------------------
diff --git a/airflow/www/views.py b/airflow/www/views.py
index b98bd74..9e68079 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -18,7 +18,6 @@ from past.builtins import basestring, unicode
import os
import pkg_resources
import socket
-import importlib
from functools import wraps
from datetime import datetime, timedelta
import dateutil.parser
@@ -577,8 +576,8 @@ class Airflow(BaseView):
dag = dagbag.get_dag(dag_id)
title = dag_id
try:
- m = importlib.import_module(dag.module_name)
- code = inspect.getsource(m)
+ with open(dag.fileloc, 'r') as f:
+ code = f.read()
html_code = highlight(
code, lexers.PythonLexer(), HtmlFormatter(linenos=True))
except IOError as e:
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/a7abcf35/tests/models.py
----------------------------------------------------------------------
diff --git a/tests/models.py b/tests/models.py
index 867e293..7ca01e7 100644
--- a/tests/models.py
+++ b/tests/models.py
@@ -200,6 +200,24 @@ class DagBagTest(unittest.TestCase):
assert dagbag.get_dag(dag_id) != None
assert dagbag.process_file_calls == 1
+ def test_get_dag_fileloc(self):
+ """
+ Test that fileloc is correctly set when we load example DAGs,
+ specifically SubDAGs.
+ """
+ dagbag = models.DagBag(include_examples=True)
+
+ expected = {
+ 'example_bash_operator': 'example_bash_operator.py',
+ 'example_subdag_operator': 'example_subdag_operator.py',
+ 'example_subdag_operator.section-1': 'subdags/subdag.py'
+ }
+
+ for dag_id, path in expected.items():
+ dag = dagbag.get_dag(dag_id)
+ self.assertTrue(
+ dag.fileloc.endswith('airflow/example_dags/' + path))
+
class TaskInstanceTest(unittest.TestCase):