title: Passing return value from operator to following operator in Airflow
body: <p>I'm trying to give a string list to the <code>source_objects</code> field for the <code>GoogleCloudStorageToBigQueryOperator</code> but whith the following code I get an error: </p>
  
  <blockquote>
    <p>string indices must be integers, not unicode</p>
  </blockquote>
  
  <p><strong>Things I don't know :</strong> </p>
  
  <ul>
  <li>How do I get the <code>return</code> value of the <code>get_file_name</code> with an XCOM at the DAG scope ?</li>
  <li>How do I call the <code>xcom_pull</code> function at the DAG scope without having to provide a context? It seemed to me that task instances do not need to provide a context.</li>
  </ul>
  
  <p><strong>Things I thought of :</strong></p>
  
  <ul>
  <li>Rewrite the operator and take an XCOM as argument</li>
  </ul>
  
  <p><strong>The thing I want to do :</strong> </p>
  
  <ul>
  <li>I want to be able to call the operator </li>
  </ul>
  
  <p>Also, it seems that some of the operator's fields use a feature called <code>templated_field</code>, what's the mechanism behind the template fields? Isn't just for <code>PythonOperator</code> and <code>BashOperator</code>? </p>
  
  <p>And a last one, why does <code>PythonOperator</code> does not return a <code>TaskInstance</code>?</p>
  
  <pre><code>with DAG('bq_load_file_from_cloud_function', default_args=default_args) as dag:
  
      def get_file_name_from_conf(ds, **kwargs):
          fileName = kwargs['dag_run'].conf['fileName']
          return [fileName]
  
      get_file_name = PythonOperator(
          task_id='get_file_name',
          provide_context=True,
          python_callable=get_file_name_from_conf)
  
      # t1, t2 and t3 are examples of tasks created by instantiating operators
      bq_load = GoogleCloudStorageToBigQueryOperator(
          task_id='bq_load', 
          bucket='src_bucket', 
          #source_objects=['data.csv'], 
          source_objects=get_file_name.xcom_pull(context='', task_ids='get_file_name'), 
          destination_project_dataset_table='project:dataset.table', 
          write_disposition='WRITE_EMPTY')
  
      bq_load.set_upstream(get_file_name)
  </code></pre>
  
  <p>I'm kind of new to Python and Airflow. These kinds of things are supposed to be trivial I guess. I'm sure there's something I misunderstood here.</p>