[airflow] branch main updated: fix(CloudSqlProxyRunner): don't query connections from Airflow DB (#18006)

[kamilbregula]

This is an automated email from the ASF dual-hosted git repository.

kamilbregula pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 21348c1  fix(CloudSqlProxyRunner): don't query connections from 
Airflow DB (#18006)
21348c1 is described below

commit 21348c194d4149237e357e0fff9ed444d27fa71d
Author: Cedrik Neumann <7921017+m1rac...@users.noreply.github.com>
AuthorDate: Fri Sep 3 22:26:24 2021 +0200

    fix(CloudSqlProxyRunner): don't query connections from Airflow DB (#18006)
    
    Instead of directly querying connections from Airflow DB we use
    `get_connection()` which also support external secrets backends.
    
    Fixes: #18003
---
 airflow/providers/google/cloud/hooks/cloud_sql.py | 9 +++------
 1 file changed, 3 insertions(+), 6 deletions(-)

diff --git a/airflow/providers/google/cloud/hooks/cloud_sql.py 
b/airflow/providers/google/cloud/hooks/cloud_sql.py
index ee8bb1c..ce77671 100644
--- a/airflow/providers/google/cloud/hooks/cloud_sql.py
+++ b/airflow/providers/google/cloud/hooks/cloud_sql.py
@@ -39,7 +39,6 @@ from urllib.parse import quote_plus
 import httpx
 from googleapiclient.discovery import Resource, build
 from googleapiclient.errors import HttpError
-from sqlalchemy.orm import Session
 
 from airflow.exceptions import AirflowException
 
@@ -51,7 +50,6 @@ from airflow.providers.google.common.hooks.base_google import 
GoogleBaseHook
 from airflow.providers.mysql.hooks.mysql import MySqlHook
 from airflow.providers.postgres.hooks.postgres import PostgresHook
 from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.session import provide_session
 
 UNIX_PATH_MAX = 108
 
@@ -516,10 +514,9 @@ class CloudSqlProxyRunner(LoggingMixin):
         os.chmod(self.sql_proxy_path, 0o744)  # Set executable bit
         self.sql_proxy_was_downloaded = True
 
-    @provide_session
-    def _get_credential_parameters(self, session: Session) -> List[str]:
-        connection = session.query(Connection).filter(Connection.conn_id == 
self.gcp_conn_id).first()
-        session.expunge_all()
+    def _get_credential_parameters(self) -> List[str]:
+        connection = GoogleBaseHook.get_connection(conn_id=self.gcp_conn_id)
+
         if connection.extra_dejson.get(GCP_CREDENTIALS_KEY_PATH):
             credential_params = ['-credential_file', 
connection.extra_dejson[GCP_CREDENTIALS_KEY_PATH]]
         elif connection.extra_dejson.get(GCP_CREDENTIALS_KEYFILE_DICT):