This is an automated email from the ASF dual-hosted git repository.
kaxil 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 d79c0b18f35 Fix ORM models failing to configure in processes that skip
database setup (#74642)
d79c0b18f35 is described below
commit d79c0b18f35d025e14a62aab9ee288230245eb83
Author: Kaxil Naik <[email protected]>
AuthorDate: Wed Oct 14 11:14:27 2026 +0100
Fix ORM models failing to configure in processes that skip database setup
(#74642)
Models name each other in string relationships, which only resolve once
every model module is imported. The before_configured hook that imports
them all was registered in configure_orm(), so a process that skips it
(virtualenv-mode subprocesses, a task re-executed for run_as_user,
non-DB tests) or configures mappers before it runs (local settings)
only worked if it had already imported the right model modules. If it
had not, building a model failed, for example with Team.dag_bundles
unable to find DagBundleModel.
Register the hook in airflow.models.base, before any model class is
defined, and drop the test-only copy from the pytest plugin. Remove the
orm_event_handlers test: it guarded the line that moved, and now passes
whatever the code does.
---
airflow-core/src/airflow/models/base.py | 11 +++++-
.../src/airflow/utils/orm_event_handlers.py | 4 --
airflow-core/tests/unit/models/test_base.py | 43 +++++++++++++++++++++-
.../tests/unit/utils/test_orm_event_handlers.py | 37 -------------------
devel-common/src/tests_common/pytest_plugin.py | 15 --------
5 files changed, 50 insertions(+), 60 deletions(-)
diff --git a/airflow-core/src/airflow/models/base.py
b/airflow-core/src/airflow/models/base.py
index 1c7af7b275a..522753f76ba 100644
--- a/airflow-core/src/airflow/models/base.py
+++ b/airflow-core/src/airflow/models/base.py
@@ -19,10 +19,11 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any
-from sqlalchemy import Integer, MetaData, String, text
-from sqlalchemy.orm import Mapped, mapped_column, registry
+from sqlalchemy import Integer, MetaData, String, event, text
+from sqlalchemy.orm import Mapped, Mapper, mapped_column, registry
from airflow.configuration import conf
+from airflow.models import import_all_models
SQL_ALCHEMY_SCHEMA = conf.get("database", "SQL_ALCHEMY_SCHEMA")
@@ -48,6 +49,12 @@ metadata = MetaData(schema=_get_schema(),
naming_convention=naming_convention)
mapper_registry = registry(metadata=metadata)
_sentinel = object()
+# Models refer to each other by class name, such as
relationship("DagBundleModel"), and a name resolves
+# only once its module is imported. Import every model before SQLAlchemy first
configures mappers,
+# whether or not this process has a database engine. SQLAlchemy 2.0 offers
this event only on the
+# Mapper class (2.1 adds one per registry), so it fires on whichever registry
configures first.
+event.listen(Mapper, "before_configured", import_all_models, once=True)
+
if TYPE_CHECKING:
Base = Any
else:
diff --git a/airflow-core/src/airflow/utils/orm_event_handlers.py
b/airflow-core/src/airflow/utils/orm_event_handlers.py
index da42433dbb0..74d51576a5e 100644
--- a/airflow-core/src/airflow/utils/orm_event_handlers.py
+++ b/airflow-core/src/airflow/utils/orm_event_handlers.py
@@ -24,7 +24,6 @@ import traceback
from uuid import uuid4
from sqlalchemy import event, exc
-from sqlalchemy.orm import Mapper
from airflow.configuration import conf
@@ -33,9 +32,6 @@ log = logging.getLogger(__name__)
def setup_event_handlers(engine):
"""Setups event handlers."""
- from airflow.models import import_all_models
-
- event.listen(Mapper, "before_configured", import_all_models, once=True)
@event.listens_for(engine, "connect")
def connect(dbapi_connection, connection_record):
diff --git a/airflow-core/tests/unit/models/test_base.py
b/airflow-core/tests/unit/models/test_base.py
index 51fd430dbab..e5b3728081f 100644
--- a/airflow-core/tests/unit/models/test_base.py
+++ b/airflow-core/tests/unit/models/test_base.py
@@ -16,15 +16,18 @@
# under the License.
from __future__ import annotations
+import os
+import subprocess
+import sys
+
import pytest
from airflow.models.base import get_id_collation_args
from tests_common.test_utils.config import conf_vars
-pytestmark = pytest.mark.db_test
-
[email protected]_test
@pytest.mark.parametrize(
("dsn", "expected", "extra"),
[
@@ -50,3 +53,39 @@ pytestmark = pytest.mark.db_test
def test_collation(dsn, expected, extra):
with conf_vars({("database", "sql_alchemy_conn"): dsn, **extra}):
assert expected == get_id_collation_args()
+
+
+def test_models_configure_in_process_that_skips_orm_setup():
+ # Models name each other in string relationships, which resolve only once
every model is imported.
+ # This process never calls configure_orm(), so building the first model
must import the rest.
+ code = "from airflow.models.connection import Connection;
Connection(conn_id='test', conn_type='http')"
+
+ result = subprocess.run(
+ [sys.executable, "-c", code],
+ env={**os.environ, "PYTHON_OPERATORS_VIRTUAL_ENV_MODE": "1"},
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+
+ assert result.returncode == 0, result.stderr
+
+
+def test_local_settings_can_build_models_while_airflow_is_imported(tmp_path):
+ # airflow_local_settings is imported before configure_orm() runs.
+ (tmp_path / "airflow_local_settings.py").write_text(
+ "from sqlalchemy.orm import configure_mappers\n"
+ "from airflow.models.team import Team\n"
+ "configure_mappers()\n"
+ )
+ python_path = os.pathsep.join(filter(None, [str(tmp_path),
os.environ.get("PYTHONPATH")]))
+
+ result = subprocess.run(
+ [sys.executable, "-c", "import airflow"],
+ env={**os.environ, "PYTHONPATH": python_path},
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+
+ assert result.returncode == 0, result.stderr
diff --git a/airflow-core/tests/unit/utils/test_orm_event_handlers.py
b/airflow-core/tests/unit/utils/test_orm_event_handlers.py
deleted file mode 100644
index 0b670c65662..00000000000
--- a/airflow-core/tests/unit/utils/test_orm_event_handlers.py
+++ /dev/null
@@ -1,37 +0,0 @@
-# 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.
-
-"""
-These tests ensure compatibility with the deprecation of
`sqlalchemy.orm.mapper()` in SQLAlchemy 2.0.0.b1.
-See also:
https://docs.sqlalchemy.org/en/21/orm/mapping_styles.html#orm-imperative-mapping
-"""
-
-from __future__ import annotations
-
-import contextlib
-import warnings
-
-from sqlalchemy.exc import InvalidRequestError, SADeprecationWarning
-
-from airflow.utils.orm_event_handlers import setup_event_handlers
-
-
-def test_setup_event_handlers_no_sa_deprecation_warning():
- warnings.simplefilter("error", category=SADeprecationWarning)
- with contextlib.suppress(InvalidRequestError):
- # We are only interested in the warning that may be issued at the
beginning of the method.
- setup_event_handlers(engine=None)
diff --git a/devel-common/src/tests_common/pytest_plugin.py
b/devel-common/src/tests_common/pytest_plugin.py
index 3cca6dec53c..c82b2644551 100644
--- a/devel-common/src/tests_common/pytest_plugin.py
+++ b/devel-common/src/tests_common/pytest_plugin.py
@@ -139,21 +139,6 @@ if skip_db_tests:
os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = "bad_schema:///"
# Set it here to pass the flag to python-xdist spawned processes
os.environ["_AIRFLOW_SKIP_DB_TESTS"] = "true"
- # configure_orm() never succeeds against the connection string above, so
the hook it installs to
- # import every model before SQLAlchemy configures mappers is missing.
Install the same hook here,
- # or a string relationship such as Team.dag_bundles cannot resolve when a
test first builds an ORM
- # object. Imported lazily: importing airflow here would initialize its
settings before this module
- # finishes setting the environment, and not every distribution's test
environment has SQLAlchemy.
- if importlib.util.find_spec("sqlalchemy"):
- from sqlalchemy import event
- from sqlalchemy.orm import Mapper
-
- def _import_all_models() -> None:
- from airflow.models import import_all_models
-
- import_all_models()
-
- event.listen(Mapper, "before_configured", _import_all_models,
once=True)
if run_db_tests_only:
# Set it here to pass the flag to python-xdist spawned processes
