title: airflow does not satisfy task dependencies
body: <p>I have a simple airflow workflow composed of two tasks. One does make a download of a csv file containing stock data. The other extracts the maximum stock price and write the data to another file.</p>
  
  <p>If I run the first task and then the second everything works fine, instead if execute: airflow run stocks_d get_max_share it fails to meet the dependency.</p>
  
  <pre><code>import csv
  from datetime import datetime
  from datetime import timedelta
  from airflow import DAG
  from airflow.operators.python_operator import PythonOperator
  import requests
  
  
  def get_stock_data():
      url = "https://app.quotemedia.com/quotetools/getHistoryDownload.csv?&amp;webmasterId=501&amp;startDay=02&amp;startMonth=02&amp;startYear=2002&amp;endDay=02&amp;endMonth=07&amp;endYear=2009&amp;isRanged=false&amp;symbol=APL"
      try:
          r = requests.get(url)
      except requests.RequestException as re:
          raise
      else:
          with open('/tmp/stocks/airflow_stock_data.txt', 'w') as f:
              f.write(r.text)
  
  def get_max_share():
      stock_data = []
      stock_max = {}
      with open('/tmp/stocks/airflow_stock_data.txt', 'r') as f:
          stock_reader = csv.reader(f)
          next(stock_reader, None)
          for row in stock_reader:
              stock_data.append(row)
  
      for stock in stock_data:
          stock_max[stock[2]] = stock[0]
  
      with open('/tmp/stocks/max_stock', 'w') as f:
          stock_price = max(stock_max.keys())
          stock_max_price_date = stock_max[stock_price]
          stock_entry = stock_max_price_date + ' -&gt; ' + stock_price
          f.write(stock_entry)
  
  
  default_args = {
      'owner': 'airflow',
      'depends_on_past': False,
      'start_date': datetime(2017, 5, 30),
      'email': ['mainl@domain.io'],
      'email_on_failure': False,
      'email_on_retry': False,
      'retries': 1,
      'retry_delay': timedelta(minutes=5),
      'catchup': False,
      }
  
  dag = DAG('stocks_d', default_args=default_args, schedule_interval=timedelta(minutes=5))
  
  
  task_get_stocks = PythonOperator(task_id='get_stocks', python_callable=get_stock_data, dag=dag)
  task_get_max_share = PythonOperator(task_id='get_max_share', python_callable=get_max_share, dag=dag)
  
  task_get_max_share.set_upstream(task_get_stocks)
  </code></pre>
  
  <p>Any ideas why does that happen ?</p>