Add CI support for SQLAlchemy 2.0

.github/workflows/run-unit-tests.yml

@@ -92,6 +92,11 @@ on:  # yamllint disable-line rule:truthy
         required: false
         default: "false"
         type: string
+      upgrade-sqlalchemy:
+        description: "Whether to upgrade SQLAlchemy or not (true/false)"
+        required: false
+        default: "false"
+        type: string
       upgrade-boto:
         description: "Whether to upgrade boto or not (true/false)"
         required: false
@@ -166,6 +171,7 @@ jobs:
       PARALLEL_TEST_TYPES: ${{ matrix.test-types.test_types }}
       PYTHON_MAJOR_MINOR_VERSION: "${{ matrix.python-version }}"
       UPGRADE_BOTO: "${{ inputs.upgrade-boto }}"
+      UPGRADE_SQLALCHEMY: "${{ inputs.upgrade-sqlalchemy }}"
       AIRFLOW_MONITOR_DELAY_TIME_IN_SECONDS: "${{inputs.monitor-delay-time-in-seconds}}"
       VERBOSE: "true"
       DEFAULT_BRANCH: "${{ inputs.default-branch }}"

.github/workflows/special-tests.yml

@@ -137,6 +137,64 @@ jobs:
       use-uv: ${{ inputs.use-uv }}
       default-branch: ${{ inputs.default-branch }}
 
+  tests-latest-sqlalchemy:
+    name: "Latest SQLAlchemy test: core"
+    uses: ./.github/workflows/run-unit-tests.yml
+    permissions:
+      contents: read
+      packages: read
+    with:
+      runners: ${{ inputs.runners }}
+      platform: ${{ inputs.platform }}
+      upgrade-sqlalchemy: "true"
+      test-name: "LatestSQLAlchemy-Postgres"
+      test-scope: "DB"
+      test-group: "core"
+      backend: "postgres"
+      # The python version constraint is a TEMPORARY WORKAROUND to exclude all FAB tests. It should be
+      # removed after upgrading FAB to v5 (PR #50960). The setting below should be:
+      #   "['${{ inputs.default-python-version }}']"
+      python-versions: "['3.13']"
+      backend-versions: "['${{ inputs.default-postgres-version }}']"
+      excluded-providers-as-string: ${{ inputs.excluded-providers-as-string }}
+      excludes: "[]"
+      test-types-as-strings-in-json: ${{ inputs.core-test-types-list-as-strings-in-json }}
+      run-coverage: ${{ inputs.run-coverage }}
+      debug-resources: ${{ inputs.debug-resources }}
+      skip-providers-tests: ${{ inputs.skip-providers-tests }}
+      use-uv: ${{ inputs.use-uv }}
+      default-branch: ${{ inputs.default-branch }}
+    if: contains(fromJSON(inputs.python-versions), '3.13')  # Remove this line after upgrading FAB to v5
+
+  tests-latest-sqlalchemy-providers:
+    name: "Latest SQLAlchemy test: providers"
+    uses: ./.github/workflows/run-unit-tests.yml
+    permissions:
+      contents: read
+      packages: read
+    with:
+      runners: ${{ inputs.runners }}
+      platform: ${{ inputs.platform }}
+      upgrade-sqlalchemy: "true"
+      test-name: "LatestSQLAlchemy-Postgres"
+      test-scope: "DB"
+      test-group: "providers"
+      backend: "postgres"
+      # The python version constraint is a TEMPORARY WORKAROUND to exclude all FAB tests. It should be
+      # removed after upgrading FAB to v5 (PR #50960). The setting below should be:
+      #   "['${{ inputs.default-python-version }}']"
+      python-versions: "['3.13']"
+      backend-versions: "['${{ inputs.default-postgres-version }}']"
+      excluded-providers-as-string: ${{ inputs.excluded-providers-as-string }}
+      excludes: "[]"
+      test-types-as-strings-in-json: ${{ inputs.providers-test-types-list-as-strings-in-json }}
+      run-coverage: ${{ inputs.run-coverage }}
+      debug-resources: ${{ inputs.debug-resources }}
+      skip-providers-tests: ${{ inputs.skip-providers-tests }}
+      use-uv: ${{ inputs.use-uv }}
+      default-branch: ${{ inputs.default-branch }}
+    if: contains(fromJSON(inputs.python-versions), '3.13')  # Remove this line after upgrading FAB to v5
+
   tests-boto-core:
     name: "Latest Boto test: core"
     uses: ./.github/workflows/run-unit-tests.yml

Dockerfile.ci

@@ -1116,14 +1116,15 @@ function check_boto_upgrade() {
 }
 
 function check_upgrade_sqlalchemy() {
-    if [[ "${UPGRADE_SQLALCHEMY}" != "true" ]]; then
+    # The python version constraint is a TEMPORARY WORKAROUND to exclude all FAB tests. Is should be removed once we
+    # upgrade FAB to v5 (PR #50960).
+    if [[ "${UPGRADE_SQLALCHEMY}" != "true" || ${PYTHON_MAJOR_MINOR_VERSION} != "3.13" ]]; then
         return
     fi
     echo
     echo "${COLOR_BLUE}Upgrading sqlalchemy to the latest version to run tests with it${COLOR_RESET}"
     echo
-    # shellcheck disable=SC2086
-    ${PACKAGING_TOOL_CMD} install ${EXTRA_INSTALL_FLAGS} --upgrade "sqlalchemy[asyncio]<2.1" "databricks-sqlalchemy>=2"
+    uv sync --all-packages --no-install-package apache-airflow-providers-fab --resolution highest
 }
 
 function check_downgrade_sqlalchemy() {

airflow-core/pyproject.toml

@@ -125,11 +125,8 @@ dependencies = [
     "rich-argparse>=1.0.0",
     "rich>=13.6.0",
     "setproctitle>=1.3.3",
-    # We use some deprecated features of sqlalchemy 2.0 and we should replace them before we can upgrade
-    # See https://sqlalche.me/e/b8d9 for details of deprecated features
-    # you can set environment variable SQLALCHEMY_WARN_20=1 to show all deprecation warnings.
-    # The issue tracking it is https://github.com/apache/airflow/issues/28723
-    "sqlalchemy[asyncio]>=1.4.49,<2.0",
+    # The issue tracking deprecations for sqlalchemy 2 is https://github.com/apache/airflow/issues/28723
+    "sqlalchemy[asyncio]>=1.4.49",
     "sqlalchemy-jsonfield>=1.0",
     "sqlalchemy-utils>=0.41.2",
     "svcs>=25.1.0",

airflow-core/src/airflow/utils/db.py

@@ -664,11 +664,10 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
 def _create_db_from_orm(session):
     """Create database tables from ORM models and stamp alembic version."""
-    log.info("Creating Airflow database tables from the ORM")
-    from alembic import command
-
     from airflow.models.base import Base
 
+    log.info("Creating Airflow database tables from the ORM")
+
     # Debug setup if requested
     _setup_debug_logging_if_needed()
 
@@ -677,13 +676,20 @@ def _create_db_from_orm(session):
         log.info("Binding engine")
         engine = session.get_bind().engine
         log.info("Pool status: %s", engine.pool.status())
+
         log.info("Creating metadata")
         Base.metadata.create_all(engine)
 
-        # Stamp the migration head
         log.info("Getting alembic config")
         config = _get_alembic_config()
-        command.stamp(config, "head")
+
+        # Use AUTOCOMMIT for DDL to avoid metadata lock issues
+        with AutocommitEngineForMySQL():  # TODO: enable for sqlite too
+            from alembic import command
+
+            log.info("Stamping migration head")
+            command.stamp(config, "head")
+
         log.info("Airflow database tables created")
 
 

airflow-core/tests/unit/always/test_example_dags.py

@@ -163,6 +163,8 @@ def relative_path(path):
 @pytest.mark.db_test
 @pytest.mark.parametrize("example", example_not_excluded_dags())
 def test_should_be_importable(example: str):
+    pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
     dagbag = DagBag(
         dag_folder=example,
         include_examples=False,

airflow-core/tests/unit/always/test_providers_manager.py

@@ -273,11 +273,15 @@ def test_hook_values(self):
         assert [w.message for w in warning_records if "hook-class-names" in str(w.message)] == []
 
     def test_connection_form_widgets(self):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         provider_manager = ProvidersManager()
         connections_form_widgets = list(provider_manager.connection_form_widgets.keys())
         assert len(connections_form_widgets) > 29
 
     def test_field_behaviours(self):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         provider_manager = ProvidersManager()
         connections_with_field_behaviours = list(provider_manager.field_behaviours.keys())
         assert len(connections_with_field_behaviours) > 16

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py

@@ -61,6 +61,8 @@ class TestGetPlugins:
     def test_should_respond_200(
         self, test_client, session, query_params, expected_total_entries, expected_names
     ):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         response = test_client.get("/plugins", params=query_params)
         assert response.status_code == 200
 
@@ -69,6 +71,8 @@ def test_should_respond_200(
         assert [plugin["name"] for plugin in body["plugins"]] == expected_names
 
     def test_external_views_model_validator(self, test_client):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         response = test_client.get("plugins")
         body = response.json()
 

airflow-core/tests/unit/plugins/test_plugins_manager.py

@@ -79,6 +79,8 @@ def clean_plugins(self):
         plugins_manager.plugins = []
 
     def test_no_log_when_no_plugins(self, caplog):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         with mock_plugin_manager(plugins=[]):
             from airflow import plugins_manager
 
@@ -118,6 +120,8 @@ def test_loads_filesystem_plugins_exception(self, caplog, tmp_path):
             assert "testplugin.py" in received_logs
 
     def test_should_warning_about_incompatible_plugins(self, caplog):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         class AirflowAdminViewsPlugin(AirflowPlugin):
             name = "test_admin_views_plugin"
 
@@ -152,6 +156,8 @@ class AirflowAdminMenuLinksPlugin(AirflowPlugin):
         ]
 
     def test_should_warning_about_conflicting_url_route(self, caplog):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         class TestPluginA(AirflowPlugin):
             name = "test_plugin_a"
 
@@ -194,6 +200,8 @@ class TestPluginB(AirflowPlugin):
         ]
 
     def test_should_not_warning_about_fab_plugins(self, caplog):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         class AirflowAdminViewsPlugin(AirflowPlugin):
             name = "test_admin_views_plugin"
 
@@ -215,6 +223,8 @@ class AirflowAdminMenuLinksPlugin(AirflowPlugin):
         assert caplog.record_tuples == []
 
     def test_should_not_warning_about_fab_and_flask_admin_plugins(self, caplog):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         class AirflowAdminViewsPlugin(AirflowPlugin):
             name = "test_admin_views_plugin"
 
@@ -336,6 +346,8 @@ def test_should_import_plugin_from_providers(self):
 
     @skip_if_force_lowest_dependencies_marker
     def test_does_not_double_import_entrypoint_provider_plugins(self):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         from airflow import plugins_manager
 
         mock_entrypoint = mock.Mock()

airflow-core/tests/unit/serialization/test_dag_serialization.py

@@ -488,6 +488,8 @@ def setup_test_cases(self):
     @pytest.mark.db_test
     def test_serialization(self):
         """Serialization and deserialization should work for every DAG and Operator."""
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
         with warnings.catch_warnings():
             dags, import_errors = collect_dags()
         serialized_dags = {}