title: Reading sql file with jinja templates in custom operator in Airflow
body: <p>I am trying to read sql file that contains query with jinja templates in the custom operator in Airflow. I have already achieved it using PythonOperator that calls function where I used </p>
  
  <pre><code>def execute_query(**kwargs)
      sql_query = open('my_sql_query.sql').read()   #(SELECT * FROM my_table WHERE date &gt; {})
      sql_query.format(kwargs['ds'])
  </code></pre>
  
  <p>but I would prefer use this syntax <strong>{{ ds }}</strong> directly in the query like 
  <code>SELECT * FROM my_table WHERE date &gt; {{ ds }}</code></p>
  
  <p>What I have done:</p>
  
  <ol>
  <li>I created CustomOperator with template_fields and template_ext</li>
  </ol>
  
  <pre><code>class SQLOperator(BaseOperator):
  
     template_fields = ['sql']
     template_ext = ('.sql',)
  
     @apply_defaults
     def __init__(
         self,
         name = None,
         sql = None,
         *args, **kwargs) -&gt; None:
         super().__init__(*args, **kwargs)
         self.name = name
         self.sql = sql
  
      def execute(self, context):
         print("Name", name) # &lt;- works
         print("Query", sql) # &lt;- doesn't work and I don't know how to get the sql file content
  
  </code></pre>
  
  <ol start="2">
  <li>Dag</li>
  </ol>
  
  <pre><code>default_args = {...}
  
  dag = DAG(
      'sql_operator_test',
       schedule_interval='0 0 * * *',
       template_searchpath=['/Users/username/airflow/dags/sql/test/'],
       default_args=default_args)
  
  sql_task = SQLOperator(
      task_id='sql_process',
      name="Aaa",
      sql="/Users/username/airflow/dags/sql/test.sql",
      dag=dag)
  </code></pre>
  
  <ol start="3">
  <li>SQL query</li>
  </ol>
  
  <pre><code>SELECT * FROM my_table WHERE date &gt; {{ ds }}
  </code></pre>
  
  <p>I am running out of ideas. Is there any option to pass the file to the operator or get its rendered content?</p>