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",
  """,