feat: backwards comp get async conn

providers/common/compat/pyproject.toml

@@ -58,6 +58,7 @@ requires-python = ">=3.10"
 # After you modify the dependencies, and rebuild your Breeze CI image with ``breeze ci-image build``
 dependencies = [
     "apache-airflow>=2.10.0",
+    "asgiref>=2.3.0",
 ]
 
 # The optional dependencies should be modified in place in the generated file

providers/common/compat/src/airflow/providers/common/compat/connection/__init__.py

@@ -0,0 +1,50 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from airflow.providers.common.compat.sdk import BaseHook
+
+if TYPE_CHECKING:
+    from airflow.providers.common.compat.sdk import Connection
+
+
+log = logging.getLogger(__name__)
+
+
+async def get_async_connection(conn_id: str) -> Connection:
+    """
+    Get an asynchronous Airflow connection that is backwards compatible.
+
+    :param conn_id: The provided connection ID.
+    :returns: Connection
+    """
+    from asgiref.sync import sync_to_async
+
+    if hasattr(BaseHook, "aget_connection"):
+        log.debug("Get connection using `BaseHook.aget_connection().")
+        return await BaseHook.aget_connection(conn_id=conn_id)
+    log.debug("Get connection using `BaseHook.get_connection().")
+    return await sync_to_async(BaseHook.get_connection)(conn_id=conn_id)
+
+
+__all__ = [
+    "get_async_connection",
+]

providers/common/compat/tests/unit/common/compat/connection/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

providers/common/compat/tests/unit/common/compat/connection/test_connection.py

@@ -0,0 +1,69 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import logging
+from unittest import mock
+
+import pytest
+
+from airflow.models.connection import Connection
+from airflow.providers.common.compat.connection import get_async_connection
+
+
+class MockAgetBaseHook:
+    def __init__(*args, **kargs):
+        pass
+
+    async def aget_connection(self, conn_id: str):
+        return Connection(
+            conn_id="test_conn",
+            conn_type="http",
+            password="secret_token_aget",
+        )
+
+
+class MockBaseHook:
+    def __init__(*args, **kargs):
+        pass
+
+    def get_connection(self, conn_id: str):
+        return Connection(
+            conn_id="test_conn_sync",
+            conn_type="http",
+            password="secret_token",
+        )
+
+
+class TestGetAsyncConnection:
+    @mock.patch("airflow.providers.common.compat.connection.BaseHook", new_callable=MockAgetBaseHook)
+    @pytest.mark.asyncio
+    async def test_get_async_connection_with_aget(self, _, caplog):
+        with caplog.at_level(logging.DEBUG):
+            conn = await get_async_connection("test_conn")
+        assert conn.password == "secret_token_aget"
+        assert conn.conn_type == "http"
+        assert "Get connection using `BaseHook.aget_connection()." in caplog.text
+
+    @mock.patch("airflow.providers.common.compat.connection.BaseHook", new_callable=MockBaseHook)
+    @pytest.mark.asyncio
+    async def test_get_async_connection_with_get_connection(self, _, caplog):
+        with caplog.at_level(logging.DEBUG):
+            conn = await get_async_connection("test_conn")
+        assert conn.password == "secret_token"
+        assert conn.conn_type == "http"
+        assert "Get connection using `BaseHook.get_connection()." in caplog.text