command line: prompting with list of tasks to clear before actually clearing

airflow/bin/airflow

@@ -106,7 +106,8 @@ def clear(args):
         end_date=args.end_date,
         upstream=args.upstream,
         downstream=args.downstream,
-        only_failed=args.only_failed)
+        only_failed=args.only_failed,
+        confirm_prompt=True)
 
 
 def webserver(args):

airflow/models.py

@@ -908,7 +908,8 @@ class DAG(Base):
     def clear(
             self, start_date=None, end_date=None,
             upstream=False, downstream=False,
-            only_failed=False):
+            only_failed=False,
+            confirm_prompt=False):
         session = settings.Session()
         """
         Clears a set of task instances associated with the current dag for
@@ -926,7 +927,22 @@ class DAG(Base):
             tis = tis.filter(TI.state == State.FAILED)
 
         count = tis.count()
-        tis.delete()
+        if count == 0:
+            print("Nothing to clear.")
+            return 0
+        if confirm_prompt:
+            ti_list = "\n".join([str(t) for t in tis])
+            question = (
+                "You are about to delete these {count} tasks:\n"
+                "{ti_list}\n\n"
+                "Are you sure? (yes/no): ").format(**locals())
+            if utils.ask_yesno(question):
+                tis.delete()
+            else:
+                count = 0
+                print("Bail. Nothing was cleared.")
+        else:
+            tis.delete()
 
         session.commit()
         session.close()

airflow/utils.py

@@ -115,3 +115,18 @@ def apply_defaults(func):
         result = func(*args, **kwargs)
         return result
     return wrapper
+
+def ask_yesno(question):
+    yes = set(['yes','y',])
+    no = set(['no','n'])
+
+    done = False
+    print(question)
+    while not done:
+        choice = raw_input().lower()
+        if choice in yes:
+           return True
+        elif choice in no:
+           return False
+        else:
+            print("Please respond by yes or no.")