incubator-airflow git commit: [AIRFLOW-413] Fix unset path bug when backfilling via pickle

[da...@apache.org]

Repository: incubator-airflow
Updated Branches:
  refs/heads/master f2b458e53 -> 6302f482a


[AIRFLOW-413] Fix unset path bug when backfilling via pickle

Please accept this PR that addresses the following issues:
- https://issues.apache.org/jira/browse/AIRFLOW-413

This fixes a bug when a pickled DAG is used in a backfill.

Testing Done:
- Existing unit tests + running a backfill command that used to fail
  before on this error

mistercrunch artwr plypaul

Closes #1723 from aoen/ddavydov/fix_undefined_path_for_backfilling


Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/6302f482
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/6302f482
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/6302f482

Branch: refs/heads/master
Commit: 6302f482a606cc578efad1d16c838a07981e36ff
Parents: f2b458e
Author: Dan Davydov <da...@airbnb.com>
Authored: Wed Aug 10 16:43:12 2016 -0700
Committer: Dan Davydov <da...@airbnb.com>
Committed: Wed Aug 10 16:43:17 2016 -0700

----------------------------------------------------------------------
 airflow/models.py | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/6302f482/airflow/models.py
----------------------------------------------------------------------
diff --git a/airflow/models.py b/airflow/models.py
index 4ccf2a1..d9f2f16 100644
--- a/airflow/models.py
+++ b/airflow/models.py
@@ -741,12 +741,13 @@ class TaskInstance(Base):
         """
         dag = self.task.dag
 
-        # Keeping existing logic, but not entirely sure why this is here.
-        if not pickle_id and dag:
-            if dag.full_filepath != dag.filepath:
-                path = "DAGS_FOLDER/{}".format(dag.filepath)
-            elif dag.full_filepath:
-                path = dag.full_filepath
+        should_pass_filepath = not pickle_id and dag
+        if should_pass_filepath and dag.full_filepath != dag.filepath:
+            path = "DAGS_FOLDER/{}".format(dag.filepath)
+        elif should_pass_filepath and dag.full_filepath:
+            path = dag.full_filepath
+        else:
+            path = None
 
         return TaskInstance.generate_command(
             self.dag_id,
@@ -820,7 +821,7 @@ class TaskInstance(Base):
         cmd += "--local " if local else ""
         cmd += "--pool {pool} " if pool else ""
         cmd += "--raw " if raw else ""
-        cmd += "-sd {file_path}"
+        cmd += "-sd {file_path}" if file_path else ""
         return cmd.format(**locals())
 
     @property