Clean up remaining getattr connection DbApiHook

airflow/providers/apache/drill/hooks/drill.py

@@ -49,7 +49,7 @@ class DrillHook(DbApiHook):
 
     def get_conn(self) -> Connection:
         """Establish a connection to Drillbit."""
-        conn_md = self.get_connection(getattr(self, self.conn_name_attr))
+        conn_md = self.get_connection(self.get_conn_id())
         creds = f"{conn_md.login}:{conn_md.password}@" if conn_md.login else ""
         database_url = (
             f"{conn_md.extra_dejson.get('dialect_driver', 'drill+sadrill')}://{creds}"
@@ -71,7 +71,7 @@ def get_uri(self) -> str:
 
         e.g: ``drill://localhost:8047/dfs``
         """
-        conn_md = self.get_connection(getattr(self, self.conn_name_attr))
+        conn_md = self.get_connection(self.get_conn_id())
         host = conn_md.host
         if conn_md.port is not None:
             host += f":{conn_md.port}"

airflow/providers/apache/drill/provider.yaml

@@ -51,7 +51,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - sqlalchemy-drill>=1.1.0
 
 integrations:

airflow/providers/apache/druid/hooks/druid.py

@@ -196,7 +196,7 @@ def __init__(self, context: dict | None = None, *args, **kwargs) -> None:
 
     def get_conn(self) -> connect:
         """Establish a connection to druid broker."""
-        conn = self.get_connection(getattr(self, self.conn_name_attr))
+        conn = self.get_connection(self.get_conn_id())
         druid_broker_conn = connect(
             host=conn.host,
             port=conn.port,
@@ -215,7 +215,7 @@ def get_uri(self) -> str:
 
         e.g: druid://localhost:8082/druid/v2/sql/
         """
-        conn = self.get_connection(getattr(self, self.conn_name_attr))
+        conn = self.get_connection(self.get_conn_id())
         host = conn.host
         if conn.port is not None:
             host += f":{conn.port}"

airflow/providers/apache/druid/provider.yaml

@@ -57,7 +57,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - pydruid>=0.4.1
 
 integrations:

airflow/providers/apache/impala/hooks/impala.py

@@ -35,7 +35,7 @@ class ImpalaHook(DbApiHook):
     hook_name = "Impala"
 
     def get_conn(self) -> Connection:
-        conn_id: str = getattr(self, self.conn_name_attr)
+        conn_id: str = self.get_conn_id()
         connection = self.get_connection(conn_id)
         return connect(
             host=connection.host,

airflow/providers/apache/impala/provider.yaml

@@ -38,6 +38,7 @@ versions:
 
 dependencies:
   - impyla>=0.18.0,<1.0
+  - apache-airflow-providers-common-sql>=1.14.1
   - apache-airflow>=2.7.0
 
 additional-extras:

airflow/providers/apache/pinot/hooks/pinot.py

@@ -288,7 +288,7 @@ def get_uri(self) -> str:
 
         e.g: http://localhost:9000/query/sql
         """
-        conn = self.get_connection(getattr(self, self.conn_name_attr))
+        conn = self.get_connection(self.get_conn_id())
         host = conn.host
         if conn.port is not None:
             host += f":{conn.port}"

airflow/providers/apache/pinot/provider.yaml

@@ -51,7 +51,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - pinotdb>=5.1.0
 
 integrations:

airflow/providers/elasticsearch/hooks/elasticsearch.py

@@ -91,7 +91,7 @@ def __init__(self, schema: str = "http", connection: AirflowConnection | None =
 
     def get_conn(self) -> ESConnection:
         """Return an elasticsearch connection object."""
-        conn_id = getattr(self, self.conn_name_attr)
+        conn_id = self.get_conn_id()
         conn = self.connection or self.get_connection(conn_id)
 
         conn_args = {
@@ -111,7 +111,7 @@ def get_conn(self) -> ESConnection:
         return connect(**conn_args)
 
     def get_uri(self) -> str:
-        conn_id = getattr(self, self.conn_name_attr)
+        conn_id = self.get_conn_id()
         conn = self.connection or self.get_connection(conn_id)
 
         login = ""

airflow/providers/elasticsearch/provider.yaml

@@ -66,7 +66,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - elasticsearch>=8.10,<9
 
 integrations:

airflow/providers/exasol/hooks/exasol.py

@@ -55,7 +55,7 @@ def __init__(self, *args, **kwargs) -> None:
         self.schema = kwargs.pop("schema", None)
 
     def get_conn(self) -> ExaConnection:
-        conn_id = getattr(self, self.conn_name_attr)
+        conn_id = self.get_conn_id()
         conn = self.get_connection(conn_id)
         conn_args = {
             "dsn": f"{conn.host}:{conn.port}",
@@ -269,7 +269,7 @@ def set_autocommit(self, conn, autocommit: bool) -> None:
         if not self.supports_autocommit and autocommit:
             self.log.warning(
                 "%s connection doesn't support autocommit but autocommit activated.",
-                getattr(self, self.conn_name_attr),
+                self.get_conn_id(),
             )
         conn.set_autocommit(autocommit)
 

airflow/providers/exasol/provider.yaml

@@ -59,7 +59,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.10.0
+  - apache-airflow-providers-common-sql>=1.14.1
   - pyexasol>=0.5.1
   # In pandas 2.2 minimal version of the sqlalchemy is 2.0
   # https://pandas.pydata.org/docs/whatsnew/v2.2.0.html#increased-minimum-versions-for-dependencies

airflow/providers/jdbc/hooks/jdbc.py

@@ -106,7 +106,7 @@ def connection_extra_lower(self) -> dict:
 
         This is used internally for case-insensitive access of jdbc params.
         """
-        conn = self.get_connection(getattr(self, self.conn_name_attr))
+        conn = self.get_connection(self.get_conn_id())
         return {k.lower(): v for k, v in conn.extra_dejson.items()}
 
     @property
@@ -150,7 +150,7 @@ def driver_class(self) -> str | None:
         return self._driver_class
 
     def get_conn(self) -> jaydebeapi.Connection:
-        conn: Connection = self.get_connection(getattr(self, self.conn_name_attr))
+        conn: Connection = self.get_connection(self.get_conn_id())
         host: str = conn.host
         login: str = conn.login
         psw: str = conn.password

airflow/providers/jdbc/provider.yaml

@@ -51,7 +51,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - jaydebeapi>=1.1.1
 
 integrations:

airflow/providers/microsoft/mssql/hooks/mssql.py

@@ -65,7 +65,7 @@ def connection(self) -> Connection:
 
         :return: The connection object.
         """
-        return self.get_connection(getattr(self, self.conn_name_attr))
+        return self.get_connection(self.get_conn_id())
 
     @property
     def connection_extra_lower(self) -> dict:

airflow/providers/microsoft/mssql/provider.yaml

@@ -53,7 +53,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - pymssql>=2.3.0
 
 integrations:

airflow/providers/mysql/hooks/mysql.py

@@ -185,7 +185,7 @@ def get_conn(self) -> MySQLConnectionTypes:
 
         :return: a mysql connection object
         """
-        conn = self.connection or self.get_connection(getattr(self, self.conn_name_attr))
+        conn = self.connection or self.get_connection(self.get_conn_id())
 
         client_name = conn.extra_dejson.get("client", "mysqlclient")
 

airflow/providers/mysql/provider.yaml

@@ -64,7 +64,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - mysqlclient>=1.4.0
   - mysql-connector-python>=8.0.29
 

airflow/providers/odbc/hooks/odbc.py

@@ -83,7 +83,7 @@ def __init__(
     def connection(self):
         """The Connection object with ID ``odbc_conn_id``."""
         if not self._connection:
-            self._connection = self.get_connection(getattr(self, self.conn_name_attr))
+            self._connection = self.get_connection(self.get_conn_id())
         return self._connection
 
     @property

airflow/providers/odbc/provider.yaml

@@ -52,7 +52,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.10.0
+  - apache-airflow-providers-common-sql>=1.14.1
   - pyodbc>=5.0.0
 
 integrations:

airflow/providers/postgres/hooks/postgres.py

@@ -117,7 +117,7 @@ def schema(self, value):
 
     @property
     def sqlalchemy_url(self) -> URL:
-        conn = self.get_connection(getattr(self, self.conn_name_attr))
+        conn = self.get_connection(self.get_conn_id())
         return URL.create(
             drivername="postgresql",
             username=conn.login,
@@ -142,7 +142,7 @@ def _get_cursor(self, raw_cursor: str) -> CursorType:
 
     def get_conn(self) -> connection:
         """Establish a connection to a postgres database."""
-        conn_id = getattr(self, self.conn_name_attr)
+        conn_id = self.get_conn_id()
         conn = deepcopy(self.connection or self.get_connection(conn_id))
 
         # check for authentication via AWS IAM

airflow/providers/postgres/provider.yaml

@@ -63,7 +63,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - psycopg2-binary>=2.9.4
 
 additional-extras:

airflow/providers/snowflake/hooks/snowflake.py

@@ -487,7 +487,7 @@ def get_openlineage_database_specific_lineage(self, _) -> OperatorLineage | None
 
         if self.query_ids:
             self.log.debug("openlineage: getting connection to get database info")
-            connection = self.get_connection(getattr(self, self.conn_name_attr))
+            connection = self.get_connection(self.get_conn_id())
             namespace = SQLParser.create_namespace(self.get_openlineage_database_info(connection))
             return OperatorLineage(
                 run_facets={

airflow/providers/snowflake/provider.yaml

@@ -76,7 +76,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.10.0
+  - apache-airflow-providers-common-sql>=1.14.1
   # In pandas 2.2 minimal version of the sqlalchemy is 2.0
   # https://pandas.pydata.org/docs/whatsnew/v2.2.0.html#increased-minimum-versions-for-dependencies
   # However Airflow not fully supports it yet: https://github.com/apache/airflow/issues/28723

airflow/providers/sqlite/hooks/sqlite.py

@@ -46,7 +46,7 @@ def get_conn(self) -> sqlite3.dbapi2.Connection:
 
     def get_uri(self) -> str:
         """Override DbApiHook get_uri method for get_sqlalchemy_engine()."""
-        conn_id = getattr(self, self.conn_name_attr)
+        conn_id = self.get_conn_id()
         airflow_conn = self.get_connection(conn_id)
         if airflow_conn.conn_type is None:
             airflow_conn.conn_type = self.conn_type

airflow/providers/sqlite/provider.yaml

@@ -54,7 +54,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
 
 integrations:
   - integration-name: SQLite

airflow/providers/teradata/hooks/teradata.py

@@ -196,7 +196,7 @@ def bulk_insert_rows(
 
     def _get_conn_config_teradatasql(self) -> dict[str, Any]:
         """Return set of config params required for connecting to Teradata DB using teradatasql client."""
-        conn: Connection = self.get_connection(getattr(self, self.conn_name_attr))
+        conn: Connection = self.get_connection(self.get_conn_id())
         conn_config = {
             "host": conn.host or "localhost",
             "dbs_port": conn.port or "1025",
@@ -230,7 +230,7 @@ def _get_conn_config_teradatasql(self) -> dict[str, Any]:
 
     def get_sqlalchemy_engine(self, engine_kwargs=None):
         """Return a connection object using sqlalchemy."""
-        conn: Connection = self.get_connection(getattr(self, self.conn_name_attr))
+        conn: Connection = self.get_connection(self.get_conn_id())
         link = f"teradatasql://{conn.login}:{conn.password}@{conn.host}"
         connection = sqlalchemy.create_engine(link)
         return connection

airflow/providers/teradata/provider.yaml

@@ -34,7 +34,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - teradatasqlalchemy>=17.20.0.0
   - teradatasql>=17.20.0.28
 

airflow/providers/ydb/hooks/ydb.py

@@ -221,7 +221,7 @@ def get_ui_field_behaviour(cls) -> dict[str, Any]:
 
     @property
     def sqlalchemy_url(self) -> URL:
-        conn: Connection = self.get_connection(getattr(self, self.conn_name_attr))
+        conn: Connection = self.get_connection(self.get_conn_id())
         connection_extra: dict[str, Any] = conn.extra_dejson
         database: str | None = connection_extra.get("database")
         return URL.create(

airflow/providers/ydb/provider.yaml

@@ -30,7 +30,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.7.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.14.1
   - ydb>=3.12.1
 
 integrations:

dev/breeze/tests/test_packages.py

@@ -204,7 +204,7 @@ def test_get_documentation_package_path():
             "postgres",
             "beta0",
             """
-    "apache-airflow-providers-common-sql>=1.3.1b0",
+    "apache-airflow-providers-common-sql>=1.14.1b0",
     "apache-airflow>=2.7.0b0",
     "psycopg2-binary>=2.9.4",
     """,
@@ -214,7 +214,7 @@ def test_get_documentation_package_path():
             "postgres",
             "",
             """
-    "apache-airflow-providers-common-sql>=1.3.1",
+    "apache-airflow-providers-common-sql>=1.14.1",
     "apache-airflow>=2.7.0",
     "psycopg2-binary>=2.9.4",
     """,