Carry ignore_dependencies from backfill to run commands

airflow/bin/cli.py

@@ -59,7 +59,8 @@ def backfill(args):
         end_date=args.end_date,
         mark_success=args.mark_success,
         include_adhoc=args.include_adhoc,
-        local=args.local)
+        local=args.local,
+        ignore_dependencies=args.ignore_dependencies)
 
 
 def run(args):

airflow/jobs.py

@@ -434,6 +434,7 @@ class BackfillJob(BaseJob):
             dag, start_date=None, end_date=None, mark_success=False,
             include_adhoc=False,
             donot_pickle=False,
+            ignore_dependencies=False,
             *args, **kwargs):
         self.dag = dag
         dag.override_start_date(start_date)
@@ -443,6 +444,7 @@ class BackfillJob(BaseJob):
         self.mark_success = mark_success
         self.include_adhoc = include_adhoc
         self.donot_pickle = donot_pickle
+        self.ignore_dependencies = ignore_dependencies
         super(BackfillJob, self).__init__(*args, **kwargs)
 
     def _execute(self):
@@ -508,7 +510,8 @@ class BackfillJob(BaseJob):
                         ti,
                         mark_success=self.mark_success,
                         task_start_date=self.bf_start_date,
-                        pickle_id=pickle_id)
+                        pickle_id=pickle_id,
+                        ignore_dependencies=self.ignore_dependencies)
                     ti.state = State.RUNNING
                     if key not in started:
                         started.append(key)

airflow/models.py

@@ -1701,7 +1701,7 @@ class DAG(object):
     def run(
             self, start_date=None, end_date=None, mark_success=False,
             include_adhoc=False, local=False, executor=None,
-            donot_pickle=False):
+            donot_pickle=False, ignore_dependencies=False):
         from airflow.jobs import BackfillJob
         if not executor and local:
             executor = LocalExecutor()
@@ -1714,7 +1714,8 @@ class DAG(object):
             mark_success=mark_success,
             include_adhoc=include_adhoc,
             executor=executor,
-            donot_pickle=donot_pickle)
+            donot_pickle=donot_pickle,
+            ignore_dependencies=ignore_dependencies)
         job.run()