title: Airflow using template files for PythonOperator
body: <p>The method of getting a <code>BashOperator</code> or <code>SqlOperator</code> to pick up an external file for its template is somewhat clearly documented, but looking at the <a href="https://github.com/apache/incubator-airflow/blob/851adc5547597ec51743be4bc47d634c77d6dc17/airflow/operators/python_operator.py" rel="noreferrer"><code>PythonOperator</code></a> my test of what I understand from the docs is not working. I am not sure how the <code>templates_exts</code> and <code>templates_dict</code> parameters would correctly interact to pick up a file.</p>
  
  <p>In my dags folder I've created: <code>pyoptemplate.sql</code> and <code>pyoptemplate.t</code> as well as <code>test_python_operator_template.py</code>:</p>
  
  <h3>pyoptemplate.sql:</h3>
  
  <pre class="lang-sql prettyprint-override"><code>SELECT * FROM {{params.table}};
  </code></pre>
  
  <h3>pyoptemplate.t:</h3>
  
  <pre class="lang-sql prettyprint-override"><code>SELECT * FROM {{params.table}};
  </code></pre>
  
  <h3>test_python_operator_template.py:</h3>
  
  <pre class="lang-python prettyprint-override"><code># coding: utf-8
  # vim:ai:si:et:sw=4 ts=4 tw=80
  """
  # A Test of Templates in PythonOperator
  """
  
  from airflow import DAG
  from airflow.operators.python_operator import PythonOperator
  from datetime import datetime
  
  import pprint
  
  pp = pprint.PrettyPrinter(indent=4)
  
  
  def templated_function(ds, **kwargs):
      """This function will try to use templates loaded from external files"""
      pp.pprint(ds)
      pp.pprint(kwargs)
  
  
  # Define the DAG
  dag = DAG(dag_id='test_python_operator_template_dag',
            default_args={"owner": "lamblin",
                          "start_date": datetime.now()},
            template_searchpath=['/Users/daniellamblin/airflow/dags'],
            schedule_interval='@once')
  
  
  # Define the single task in this controller example DAG
  op = PythonOperator(task_id='test_python_operator_template',
                      provide_context=True,
                      python_callable=templated_function,
                      templates_dict={
                          'pyoptemplate': '',
                          'pyoptemplate.sql': '',
                          'sql': 'pyoptemplate',
                          'file1':'pyoptemplate.sql',
                          'file2':'pyoptemplate.t',
                          'table': '{{params.table}}'},
                      templates_exts=['.sql','.t'],
                      params={'condition_param': True,
                              'message': 'Hello World',
                              'table': 'TEMP_TABLE'},
                      dag=dag)
  </code></pre>
  
  <p>The result from a run shows that <code>table</code> was templated correctly as a string, but the others did not pull in any files for templating.</p>
  
  <pre><code>dlamblin$ airflow test test_python_operator_template_dag test_python_operator_template 2017-01-18
  [2017-01-18 23:58:06,698] {__init__.py:36} INFO - Using executor SequentialExecutor
  [2017-01-18 23:58:07,342] {models.py:154} INFO - Filling up the DagBag from /Users/daniellamblin/airflow/dags
  [2017-01-18 23:58:07,620] {models.py:1196} INFO - 
  --------------------------------------------------------------------------------
  Starting attempt 1 of 1
  --------------------------------------------------------------------------------
  
  [2017-01-18 23:58:07,620] {models.py:1219} INFO - Executing &lt;Task(PythonOperator): test_python_operator_template&gt; on 2017-01-18 00:00:00
  '2017-01-18'
  {   u'END_DATE': '2017-01-18',
      u'conf': &lt;module 'airflow.configuration' from '/Library/Python/2.7/site-packages/airflow/configuration.pyc'&gt;,
      u'dag': &lt;DAG: test_python_operator_template_dag&gt;,
      u'dag_run': None,
      u'ds_nodash': u'20170118',
      u'end_date': '2017-01-18',
      u'execution_date': datetime.datetime(2017, 1, 18, 0, 0),
      u'latest_date': '2017-01-18',
      u'macros': &lt;module 'airflow.macros' from '/Library/Python/2.7/site-packages/airflow/macros/__init__.pyc'&gt;,
      u'params': {   'condition_param': True,
                     'message': 'Hello World',
                     'table': 'TEMP_TABLE'},
      u'run_id': None,
      u'tables': None,
      u'task': &lt;Task(PythonOperator): test_python_operator_template&gt;,
      u'task_instance': &lt;TaskInstance: test_python_operator_template_dag.test_python_operator_template 2017-01-18 00:00:00 [running]&gt;,
      u'task_instance_key_str': u'test_python_operator_template_dag__test_python_operator_template__20170118',
      'templates_dict': {   'file1': u'pyoptemplate.sql',
                            'file2': u'pyoptemplate.t',
                            'pyoptemplate': u'',
                            'pyoptemplate.sql': u'',
                            'sql': u'pyoptemplate',
                            'table': u'TEMP_TABLE'},
      u'test_mode': True,
      u'ti': &lt;TaskInstance: test_python_operator_template_dag.test_python_operator_template 2017-01-18 00:00:00 [running]&gt;,
      u'tomorrow_ds': '2017-01-19',
      u'tomorrow_ds_nodash': u'20170119',
      u'ts': '2017-01-18T00:00:00',
      u'ts_nodash': u'20170118T000000',
      u'yesterday_ds': '2017-01-17',
      u'yesterday_ds_nodash': u'20170117'}
  [2017-01-18 23:58:07,634] {python_operator.py:67} INFO - Done. Returned value was: None
  </code></pre>