a task needs to have start_date; either from the dag or the task

airflow/models.py

@@ -937,6 +937,7 @@ def __init__(
         utils.validate_key(dag_id)
         self.dag_id = dag_id
         self.executor = executor
+        self.start_date = start_date
         self.end_date = end_date or datetime.now()
         self.parallelism = parallelism
         self.schedule_interval = schedule_interval
@@ -1087,6 +1088,11 @@ def get_downstream(task, level=0):
             get_downstream(t)
 
     def add_task(self, task):
+        if not self.start_date and not task.start_date:
+            raise Exception("Task is missing the start_date parameter")
+        if not task.start_date:
+            task.start_date = self.start_date
+
         if task.task_id in [t.task_id for t in self.tasks]:
             raise Exception("Task already added")
         else:

dags/examples/simple.py

@@ -11,7 +11,7 @@
                 }
 
 # Initializing a directed acyclic graph
-dag = DAG(dag_id='simple', executor=LocalExecutor())
+dag = DAG(dag_id='simple', executor=LocalExecutor(), start_date=datetime(2014, 10, 10))
 
 # MySQL Operator
 sql = "TRUNCATE TABLE tmp;"

dags/examples/test_wf_external.py

@@ -5,6 +5,7 @@
 # Setting some default operator parameters
 default_args = {
     'owner': 'max',
+    'start_date': datetime(2014, 10, 20),
     'mysql_dbid': 'local_mysql',
 }