This is an automated email from the ASF dual-hosted git repository.
turaga 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 53eb65301e6 Warn when EdgeDBManager is missing from
external_db_managers in config (#62204)
53eb65301e6 is described below
commit 53eb65301e6953130e95781245e3774858dfca4b
Author: Dheeraj Turaga <[email protected]>
AuthorDate: Sat Feb 21 17:21:03 2026 -0600
Warn when EdgeDBManager is missing from external_db_managers in config
(#62204)
---
.../providers/edge3/executors/edge_executor.py | 3 +-
.../edge3/src/airflow/providers/edge3/models/db.py | 24 ++++++++++
.../edge3/plugins/edge_executor_plugin.py | 5 +++
providers/edge3/tests/unit/edge3/models/test_db.py | 51 ++++++++++++++++++++++
4 files changed, 82 insertions(+), 1 deletion(-)
diff --git
a/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
b/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
index f4a8e4b99e8..58f516ee122 100644
--- a/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
+++ b/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
@@ -29,7 +29,7 @@ from airflow.executors import workloads
from airflow.executors.base_executor import BaseExecutor
from airflow.models.taskinstance import TaskInstance
from airflow.providers.common.compat.sdk import Stats, timezone
-from airflow.providers.edge3.models.db import EdgeDBManager
+from airflow.providers.edge3.models.db import EdgeDBManager,
check_db_manager_config
from airflow.providers.edge3.models.edge_job import EdgeJobModel
from airflow.providers.edge3.models.edge_logs import EdgeLogsModel
from airflow.providers.edge3.models.edge_worker import EdgeWorkerModel,
EdgeWorkerState, reset_metrics
@@ -62,6 +62,7 @@ class EdgeExecutor(BaseExecutor):
@provide_session
def start(self, session: Session = NEW_SESSION):
"""If EdgeExecutor provider is loaded first time, ensure table
exists."""
+ check_db_manager_config()
edge_db_manager = EdgeDBManager(session)
if edge_db_manager.check_migration():
return
diff --git a/providers/edge3/src/airflow/providers/edge3/models/db.py
b/providers/edge3/src/airflow/providers/edge3/models/db.py
index ac61ed987c3..1207dcd704c 100644
--- a/providers/edge3/src/airflow/providers/edge3/models/db.py
+++ b/providers/edge3/src/airflow/providers/edge3/models/db.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+import warnings
from pathlib import Path
from sqlalchemy import inspect
@@ -66,3 +67,26 @@ class EdgeDBManager(BaseDBManager):
if inspector.has_table(version.name):
self.log.info("Dropping version table %s", version.name)
version.drop(connection)
+
+
+def check_db_manager_config() -> None:
+ """
+ Warn if EdgeDBManager is not registered in the external_db_managers config.
+
+ Should be called whenever the edge3 provider is active so operators are
alerted
+ early if the required database configuration is missing.
+ """
+ from airflow.configuration import conf
+
+ fqcn = f"{EdgeDBManager.__module__}.{EdgeDBManager.__name__}"
+ configured = conf.get("database", "external_db_managers", fallback="")
+ registered = [m.strip() for m in configured.split(",") if m.strip()]
+ if fqcn not in registered:
+ warnings.warn(
+ f"EdgeDBManager is not configured. Add '{fqcn}' to "
+ f"AIRFLOW__DATABASE__EXTERNAL_DB_MANAGERS (the
'external_db_managers' option "
+ f"in the [database] section). Without this, edge3 database tables
will not be "
+ f"managed through the standard Airflow migration process.",
+ UserWarning,
+ stacklevel=2,
+ )
diff --git
a/providers/edge3/src/airflow/providers/edge3/plugins/edge_executor_plugin.py
b/providers/edge3/src/airflow/providers/edge3/plugins/edge_executor_plugin.py
index ad22956258c..c73fa5b23c1 100644
---
a/providers/edge3/src/airflow/providers/edge3/plugins/edge_executor_plugin.py
+++
b/providers/edge3/src/airflow/providers/edge3/plugins/edge_executor_plugin.py
@@ -60,6 +60,11 @@ try:
except AirflowConfigException:
EDGE_EXECUTOR_ACTIVE = False
+if EDGE_EXECUTOR_ACTIVE:
+ from airflow.providers.edge3.models.db import check_db_manager_config
+
+ check_db_manager_config()
+
# Load the API endpoint only on api-server
# TODO(jscheffl): Remove this check when the discussion in
#
https://lists.apache.org/thread/w170czq6r7bslkqp1tk6bjjjo0789wgl
diff --git a/providers/edge3/tests/unit/edge3/models/test_db.py
b/providers/edge3/tests/unit/edge3/models/test_db.py
index 79739f2847c..2ec969dc0c3 100644
--- a/providers/edge3/tests/unit/edge3/models/test_db.py
+++ b/providers/edge3/tests/unit/edge3/models/test_db.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+import warnings
from unittest import mock
import pytest
@@ -239,3 +240,53 @@ class TestEdgeDBManager:
# The drop method should not be called on any table
# We check this by ensuring has_table was called but drop was
not
assert mock_inspector.has_table.called
+
+
+class TestCheckDbManagerConfig:
+ """Test check_db_manager_config warning helper."""
+
+ pytestmark: list = [] # no db_test needed — purely config-based
+
+ def test_warns_when_not_configured(self):
+ """Warning is emitted when EdgeDBManager is absent from
external_db_managers."""
+ from airflow.providers.edge3.models.db import check_db_manager_config
+
+ with conf_vars({("database", "external_db_managers"): ""}):
+ with pytest.warns(UserWarning, match="EdgeDBManager is not
configured"):
+ check_db_manager_config()
+
+ def test_warns_when_other_manager_configured(self):
+ """Warning is emitted when a different manager is configured but not
EdgeDBManager."""
+ from airflow.providers.edge3.models.db import check_db_manager_config
+
+ with conf_vars({("database", "external_db_managers"):
"some.other.DBManager"}):
+ with pytest.warns(UserWarning, match="EdgeDBManager is not
configured"):
+ check_db_manager_config()
+
+ def test_no_warn_when_configured(self):
+ """No warning when EdgeDBManager is properly configured."""
+ from airflow.providers.edge3.models.db import check_db_manager_config
+
+ with conf_vars(
+ {
+ ("database", "external_db_managers"):
"airflow.providers.edge3.models.db.EdgeDBManager",
+ }
+ ):
+ with warnings.catch_warnings():
+ warnings.simplefilter("error")
+ check_db_manager_config() # must not raise
+
+ def test_no_warn_when_configured_among_multiple(self):
+ """No warning when EdgeDBManager appears alongside other managers."""
+ from airflow.providers.edge3.models.db import check_db_manager_config
+
+ with conf_vars(
+ {
+ ("database", "external_db_managers"): (
+
"some.other.DBManager,airflow.providers.edge3.models.db.EdgeDBManager"
+ ),
+ }
+ ):
+ with warnings.catch_warnings():
+ warnings.simplefilter("error")
+ check_db_manager_config() # must not raise
