Multi-Team: Add teams verify command to validate multi-team configuration

airflow-core/docs/core-concepts/multi-team.rst

@@ -269,6 +269,11 @@ Use the ``--team-name`` option with ``airflow pools set`` to assign a pool to a
     The ``--team-name`` option is rejected when ``core.multi_team`` is disabled.
     The specified team must exist in the database (create it first with ``airflow teams create``).
 
+    When ``core.multi_team`` is enabled, ``airflow teams create`` automatically
+    creates a default pool named ``default_pool_<team_name>``. By default, tasks
+    in Dag bundles associated with that team are automatically assigned to
+    the team's default pool unless another pool is explicitly configured.
+
 Creating Team-scoped Pools via the REST API
 """""""""""""""""""""""""""""""""""""""""""
 

airflow-core/newsfragments/69768.feature.rst

@@ -0,0 +1 @@
+Add automatic creation of team default pools in multi-team deployments and assign tasks without an explicitly configured pool to their team's default pool during Dag parsing. Existing multi-team deployments should run ``airflow teams sync`` after upgrading to provision default pools for existing teams.

airflow-core/newsfragments/70947.feature.rst

@@ -0,0 +1 @@
+Add an ``airflow teams verify`` CLI command for validating multi-team configuration. The command checks that each team has its system-managed default pool, that DAG bundles reference existing teams, and reports any configuration inconsistencies.

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

@@ -1681,6 +1681,13 @@ class GroupCommand(NamedTuple):
         func=lazy_load_command("airflow.cli.commands.team_command.team_sync"),
         args=(ARG_VERBOSE,),
     ),
+    ActionCommand(
+        name="verify",
+        help="Verify multi-team configuration",
+        description=("Verify that the multi-team configuration is internally consistent.\n"),
+        func=lazy_load_command("airflow.cli.commands.team_command.team_verify"),
+        args=(ARG_VERBOSE,),
+    ),
 )
 STATE_STORE_COMMANDS = (
     ActionCommand(

airflow-core/src/airflow/cli/commands/team_command.py

@@ -20,11 +20,13 @@
 from __future__ import annotations
 
 import re
+from typing import TYPE_CHECKING
 
 from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 
 from airflow.cli.simple_table import AirflowConsole
+from airflow.configuration import conf
 from airflow.dag_processing.bundles.manager import DagBundlesManager
 from airflow.models.connection import Connection
 from airflow.models.pool import Pool
@@ -34,6 +36,9 @@
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
 from airflow.utils.session import NEW_SESSION, provide_session
 
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
 NO_TEAMS_LIST_MSG = "No teams found."
 
 
@@ -58,6 +63,17 @@ def _extract_team_name(args):
     return team_name
 
 
+def _create_default_team_pool(team_name: str, *, session: Session) -> None:
+    Pool.create_or_update_pool(
+        name=Pool.get_default_team_pool_name(team_name),
+        slots=conf.getint("core", "default_pool_task_slot_count"),
+        description=f"Default pool for team '{team_name}'",
+        include_deferred=False,
+        team_name=team_name,
+        session=session,
+    )
+
+
 @cli_utils.action_cli
 @providers_configuration_loaded
 @provide_session
@@ -74,8 +90,14 @@ def team_create(args, *, session=NEW_SESSION):
 
     try:
         session.add(new_team)
+        session.flush()
+
+        if conf.getboolean("core", "multi_team"):
+            _create_default_team_pool(team_name=team_name, session=session)
+
         session.commit()
         print(f"Team '{team_name}' created successfully.")
+
     except IntegrityError as e:
         session.rollback()
         raise SystemExit(f"Failed to create team '{team_name}': {e}")
@@ -118,7 +140,12 @@ def team_delete(args, *, session=NEW_SESSION):
         associations.append(f"{variable_count} variable(s)")
 
     # Check pool associations
-    if pool_count := session.scalar(select(func.count(Pool.id)).where(Pool.team_name == team.name)):
+    if pool_count := session.scalar(
+        select(func.count(Pool.id)).where(
+            Pool.team_name == team.name,
+            Pool.pool != Pool.get_default_team_pool_name(team.name),
+        )
+    ):
         associations.append(f"{pool_count} pool(s)")
 
     # If there are associations, prevent deletion
@@ -139,6 +166,17 @@ def team_delete(args, *, session=NEW_SESSION):
     # Delete the team
     try:
         session.delete(team)
+
+        default_pool = session.scalar(
+            select(Pool).where(
+                Pool.pool == Pool.get_default_team_pool_name(team.name),
+                Pool.team_name == team.name,
+            )
+        )
+
+        if default_pool:
+            session.delete(default_pool)
+
         session.commit()
         print(f"Team '{team_name}' deleted successfully")
     except Exception as e:
@@ -163,6 +201,10 @@ def team_list(args, *, session=NEW_SESSION):
 @provide_session
 def team_sync(args, *, session=NEW_SESSION):
     """Sync missing teams from the dag bundle config."""
+    if not conf.getboolean("core", "multi_team"):
+        print("Warning: multi-team is not enabled; nothing to synchronize.")
+        return
+
     dag_bundle_teams = {
         bundle.team_name
         for bundle in DagBundlesManager()._bundle_config.values()
@@ -172,14 +214,70 @@ def team_sync(args, *, session=NEW_SESSION):
     teams_added = 0
 
     try:
-        for team_name in dag_bundle_teams - Team.get_all_team_names(session=session):
-            team = Team(name=team_name)
-            session.add(team)
-            teams_added += 1
+        existing_teams = Team.get_all_team_names(session=session)
+        for team_name in dag_bundle_teams:
+            if team_name not in existing_teams:
+                session.add(Team(name=team_name))
+                session.flush()
+                teams_added += 1
+
+            pool = session.scalar(
+                select(Pool).where(
+                    Pool.pool == Pool.get_default_team_pool_name(team_name),
+                    Pool.team_name == team_name,
+                )
+            )
+
+            if pool is None:
+                _create_default_team_pool(team_name=team_name, session=session)
+
         session.commit()
     except Exception as e:
         session.rollback()
         raise SystemExit(f"Failed to sync teams: {e}")
 
     if teams_added > 0:
         print(f"{teams_added} teams added.")
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+@provide_session
+def team_verify(args, *, session=NEW_SESSION):
+    """Verify that the multi-team configuration is consistent."""
+    if not conf.getboolean("core", "multi_team"):
+        print("Multi-team is not enabled.")
+        return
+
+    issues: list[str] = []
+
+    teams = session.scalars(select(Team)).all()
+
+    for team in teams:
+        default_pool_name = Pool.get_default_team_pool_name(team.name)
+
+        default_pool = session.scalar(
+            select(Pool).where(
+                Pool.pool == default_pool_name,
+                Pool.team_name == team.name,
+            )
+        )
+
+        if default_pool is None:
+            issues.append(f"Team '{team.name}' is missing default pool '{default_pool_name}'.")
+
+    existing_teams = {team.name for team in teams}
+
+    for bundle_name, bundle in DagBundlesManager()._bundle_config.items():
+        if bundle.team_name and bundle.team_name not in existing_teams:
+            issues.append(f"DAG bundle '{bundle_name}' references unknown team '{bundle.team_name}'.")
+
+    if issues:
+        print("Verification failed.\n")
+
+        for issue in issues:
+            print(f"✗ {issue}")
+
+        raise SystemExit(1)
+
+    print("Verification succeeded.")

airflow-core/src/airflow/dag_processing/dagbag.py

@@ -43,6 +43,7 @@
 )
 from airflow.executors.executor_loader import ExecutorLoader
 from airflow.listeners.listener import get_listener_manager
+from airflow.models.pool import Pool
 from airflow.serialization.definitions.notset import NOTSET, ArgNotSet, is_arg_set
 from airflow.serialization.serialized_objects import LazyDeserializedDAG
 from airflow.utils.file import correct_maybe_zipped
@@ -161,6 +162,30 @@ def _validate_executor_fields(dag: DAG, bundle_name: str | None = None) -> None:
             )
 
 
+def _assign_default_team_pools(
+    dag: DAG,
+    bundle_name: str | None = None,
+) -> None:
+    """Assign the default team pool to tasks that do not explicitly specify a pool."""
+    dag_team_name = None
+
+    if conf.getboolean("core", "multi_team"):
+        if bundle_name:
+            from airflow.dag_processing.bundles.manager import DagBundlesManager
+
+            bundle_manager = DagBundlesManager()
+            bundle_config = bundle_manager._bundle_config[bundle_name]
+
+            dag_team_name = bundle_config.team_name
+
+    if not dag_team_name:
+        return
+
+    for task in dag.tasks:
+        if task.pool == Pool.DEFAULT_POOL_NAME:
+            task.pool = Pool.get_default_team_pool_name(dag_team_name)
+
+
 class DagBag(LoggingMixin):
     """
     A dagbag is a collection of dags, parsed out of a folder tree and has high level configuration settings.
@@ -344,6 +369,7 @@ def process_file(self, filepath, only_if_updated=True, safe_mode=True):
                 # Validate before adding to bag (matches original _process_modules behavior)
                 dag.validate()
                 _validate_executor_fields(dag, self.bundle_name)
+                _assign_default_team_pools(dag, self.bundle_name)
                 self.bag_dag(dag=dag)
                 bagged_dags.append(dag)
             except AirflowClusterPolicySkipDag:

airflow-core/src/airflow/models/pool.py

@@ -121,6 +121,10 @@ def get_default_pool(*, session: Session = NEW_SESSION) -> Pool | None:
         """
         return Pool.get_pool(Pool.DEFAULT_POOL_NAME, session=session)
 
+    @staticmethod
+    def get_default_team_pool_name(team_name: str) -> str:
+        return f"default_pool_{team_name}"
+
     @staticmethod
     @provide_session
     def create_or_update_pool(