Don't verify team existence during cli parser loading

airflow-core/src/airflow/cli/cli_parser.py

@@ -60,7 +60,7 @@
 log = logging.getLogger(__name__)
 
 
-for executor_name in ExecutorLoader.get_executor_names():
+for executor_name in ExecutorLoader.get_executor_names(validate_teams=False):
     try:
         executor, _ = ExecutorLoader.import_executor_cls(executor_name)
         airflow_commands.extend(executor.get_cli_commands())

airflow-core/src/airflow/executors/executor_loader.py

@@ -63,16 +63,19 @@ class ExecutorLoader:
     }
 
     @classmethod
-    def _get_executor_names(cls) -> list[ExecutorName]:
+    def _get_executor_names(cls, validate_teams: bool = True) -> list[ExecutorName]:
         """
         Return the executor names from Airflow configuration.
 
+        :param validate_teams: Whether to validate that team names exist in database
         :return: List of executor names from Airflow configuration
         """
         if _executor_names:
             return _executor_names
 
-        all_executor_names: list[tuple[str | None, list[str]]] = cls._get_team_executor_configs()
+        all_executor_names: list[tuple[str | None, list[str]]] = cls._get_team_executor_configs(
+            validate_teams=validate_teams
+        )
 
         executor_names = []
         for team_name, executor_names_config in all_executor_names:
@@ -179,12 +182,14 @@ def _validate_teams_exist_in_database(cls, team_names: set[str]) -> None:
             )
 
     @classmethod
-    def _get_team_executor_configs(cls) -> list[tuple[str | None, list[str]]]:
+    def _get_team_executor_configs(cls, validate_teams: bool = True) -> list[tuple[str | None, list[str]]]:
         """
         Return a list of executor configs to be loaded.
 
         Each tuple contains the team id as the first element and the second element is the executor config
         for that team (a list of executor names/modules/aliases).
+
+        :param validate_teams: Whether to validate that team names exist in database
         """
         from airflow.configuration import conf
 
@@ -243,19 +248,20 @@ def _get_team_executor_configs(cls) -> list[tuple[str | None, list[str]]]:
 
         # Validate that all team names exist in the database (excluding None for global configs)
         team_names_to_validate = {team_name for team_name in seen_teams if team_name is not None}
-        if team_names_to_validate:
+        if team_names_to_validate and validate_teams:
             cls._validate_teams_exist_in_database(team_names_to_validate)
 
         return configs
 
     @classmethod
-    def get_executor_names(cls) -> list[ExecutorName]:
+    def get_executor_names(cls, validate_teams: bool = True) -> list[ExecutorName]:
         """
         Return the executor names from Airflow configuration.
 
+        :param validate_teams: Whether to validate that team names exist in database
         :return: List of executor names from Airflow configuration
         """
-        return cls._get_executor_names()
+        return cls._get_executor_names(validate_teams=validate_teams)
 
     @classmethod
     def get_default_executor_name(cls, team_name: str | None = None) -> ExecutorName:

airflow-core/tests/unit/cli/test_cli_parser.py

@@ -556,3 +556,13 @@ def test_airflow_config_output_does_not_contain_providers_when_excluded(self):
         )
         assert result.returncode == 0
         assert "celery_config_options" not in result.stdout
+
+    def test_cli_parser_skips_team_validation(self):
+        """Test that CLI parser calls get_executor_names with validate_teams=False to prevent database dependency during CLI loading."""
+        with patch.object(executor_loader.ExecutorLoader, "get_executor_names") as mock_get_executor_names:
+            mock_get_executor_names.return_value = []
+            # Force reload of cli_parser to trigger the executor loading
+            reload(cli_parser)
+
+            # Verify get_executor_names was called with validate_teams=False
+            mock_get_executor_names.assert_called_with(validate_teams=False)

airflow-core/tests/unit/executors/test_executor_loader.py

@@ -630,3 +630,30 @@ def test_team_validation_skips_global_teams(self):
 
                 # No team validation should occur since only global teams are configured
                 mock_get_team_names.assert_not_called()
+
+    def test_get_executor_names_skip_team_validation(self):
+        """Test that get_executor_names can skip team validation."""
+        with (
+            patch.object(executor_loader.Team, "get_all_team_names") as mock_get_team_names,
+            mock.patch.object(executor_loader.ExecutorLoader, "block_use_of_multi_team"),
+        ):
+            with conf_vars(
+                {("core", "executor"): "=CeleryExecutor;team_a=LocalExecutor", ("core", "multi_team"): "True"}
+            ):
+                # Should not call team validation when validate_teams=False
+                executor_loader.ExecutorLoader.get_executor_names(validate_teams=False)
+                mock_get_team_names.assert_not_called()
+
+    def test_get_executor_names_default_validates_teams(self):
+        """Test that get_executor_names validates teams by default."""
+        with (
+            patch.object(executor_loader.Team, "get_all_team_names") as mock_get_team_names,
+            mock.patch.object(executor_loader.ExecutorLoader, "block_use_of_multi_team"),
+        ):
+            with conf_vars(
+                {("core", "executor"): "=CeleryExecutor;team_a=LocalExecutor", ("core", "multi_team"): "True"}
+            ):
+                # Default behavior should validate teams
+                mock_get_team_names.return_value = {"team_a"}
+                executor_loader.ExecutorLoader.get_executor_names()
+                mock_get_team_names.assert_called_once()