title: Using dagrun.conf on custom operator
body: <p>I am using airflow 2 stable rest API to trigger a dag. I have created a custom DAG that takes input from MySQL (2 tables) join on key.</p>
  <p>and in the body of API, I have to send some parameters like below and these parameters will decide which 2 tables to join.</p>
  <pre><code>conf&quot;: {&quot;database_1&quot;:&quot;test&quot;,&quot;table_1&quot;:&quot;student&quot;,&quot;key_1&quot;:&quot;id&quot;,&quot;database_2&quot;: &quot;test&quot;,&quot;table_2&quot;: &quot;college&quot;,&quot;key_2&quot;: &quot;student_id&quot;},
  </code></pre>
  <p>Below is the Custom Operator implementation.</p>
  <pre><code>from typing import Dict, Iterable, Mapping, Optional, Union
  
  from airflow.models.baseoperator import BaseOperator
  from airflow.providers.mysql.hooks.mysql import MySqlHook
  from airflow.providers.mysql.operators.mysql import MySqlOperator
  from airflow.utils.decorators import apply_defaults
  
  class MySqlJoinOperator(BaseOperator):
  
      @apply_defaults
      def __init__(
              self,
              *,
              mysql_conn_id: str = 'mysql_default',
              parameters: Optional[Union[Mapping, Iterable]] = None,
              autocommit: bool = False,
              database_1: str = None,
              table_1: str = None,
              key_1: str = None,
              database_2: str = None,
              table_2: str = None,
              key_2: str = None,
              how: str = 'inner',
              **kwargs,) -&gt; None:
              super().__init__(**kwargs)
              self.mysql_conn_id = mysql_conn_id
              self.autocommit = autocommit
              self.parameters = parameters
              self.database_1 = database_1
              self.table_1 = table_1
              self.key_1 = key_1
              self.database_2 = database_2
              self.table_2 = table_2
              self.key_2= key_2
              self.how = how
  
      def execute(self, context: Dict) -&gt; None:
          self.log.info('Joining  {}.{} and {}.{}'.format(self.database_1,self.table_1,self.database_2,self.table_2))
          # hook_1 = MySqlHook(mysql_conn_id=self.mysql_conn_id, schema=self.database_1)
          # hook_2 = MySqlHook(mysql_conn_id=self.mysql_conn_id, schema=self.database_2)
          # sql_1 = &quot;select * from {}&quot;.format(self.table_1)
          # sql_2 = &quot;select * from {}&quot;.format(self.table_2)
          # dataframe_1 = hook_1.get_pandas_df(sql_1)
          # dataframe_2 = hook_2.get_pandas_df(sql_2)
          # resultant_dataframe = dataframe_1.join(dataframe_2,how=self.how)
          hook = MySqlHook(mysql_conn_id = self.mysql_conn_id)
          sql = &quot;select * from {}.{} as table_1 join {}.{} as table_2 on table_1.{} = table_2.{}&quot;.format(self.database_1, self.table_1, self.database_2, self.table_2, self.key_1, self.key_2)
          resultant_dataframe = hook.get_records(sql)
          print(resultant_dataframe)
          return resultant_dataframe
  </code></pre>
  <p>and it will be used like this</p>
  <pre><code>from airflow import DAG
  from airflow. utils.dates import days_ago
  
  from mysql_join import MySqlJoinOperator
  
  dag = DAG(
      dag_id='test_mysql_join',
      schedule_interval=None,
      start_date=days_ago(10),
      tags=['test mysql join'],
  )
  
  test_mysql_operator = MySqlJoinOperator(
      task_id='join_test',
      mysql_conn_id = &quot;root_MYSQL_connection&quot;,
      database_1= &quot;{{ dag_run.conf['database_1'] }}&quot;,
      table_1=&quot;{{ dag_run.conf['table_1'] }}&quot;,
      key_1=&quot;{{ dag_run.conf['key_1'] }}&quot;,
      database_2= &quot;{{ dag_run.conf['database_2'] }}&quot;,
      table_2 = &quot;{{ dag_run.conf['table_2'] }}&quot;,
      key_2 = &quot;{{ dag_run.conf['key_2'] }}&quot;,
      dag=dag)
  </code></pre>
  <p>but the jinja template not working here.
  Can anyone help me with how can I achieve this?</p>