title: Connection pooling for external connections in Airflow
body: <p>I am trying to find a way for connection pool management for external connections created in Airflow.<br />
  Airflow version : 2.1.0<br />
  Python Version : 3.9.5<br />
  Airflow DB : SQLite<br />
  External connections created :  MySQL and Snowflake</p>
  <p>I know there are properties in airflow.cfg file</p>
  <pre><code>sql_alchemy_pool_enabled = True  
  sql_alchemy_pool_size = 5
  </code></pre>
  <p>But these properties are for managing the airflow internal DB which is SQLite in my case.</p>
  <p>I have few tasks which are reading or writing data in MySQL and Snowflake.</p>
  <pre><code>snowflake_insert = SnowflakeOperator(
            task_id='insert_snowflake',
            dag=dag,
            snowflake_conn_id=SNOWFLAKE_CONN_ID,
            sql=&quot;Some Insert query&quot;,
            warehouse=SNOWFLAKE_WAREHOUSE,
            database=SNOWFLAKE_DATABASE,
            schema=SNOWFLAKE_SCHEMA,
            role=SNOWFLAKE_ROLE
       )
  </code></pre>
  <p>and</p>
  <pre><code>insert_mysql_task = MySqlOperator(task_id='insert_record', mysql_conn_id='mysql_default', sql=&quot;some insert query&quot;, dag=dag)
  </code></pre>
  <p>Reading data from MySQL</p>
  <pre><code>def get_records():
      mysql_hook = MySqlHook(mysql_conn_id=&quot;mysql_default&quot;)
      records = mysql_hook.get_records(sql=r&quot;&quot;&quot;Some select query&quot;&quot;&quot;)
      print(records)
  </code></pre>
  <p>What I observed is a new session is getting created for each of the task (there are multiple tasks in the same dag) for Snowflake, haven't verified the same for MySQL.</p>
  <p>Is there a way to maintain connection pool for external connections (in my case Snowflake and MySQL) or any other way to run all the queries in same DAG in same session ?</p>
  <p>Thanks</p>