Allow a Job to be associated with several teams

airflow-core/docs/administration-and-deployment/logging-monitoring/check-health.rst

@@ -64,7 +64,7 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc
         {
           "hostname": "triggerer-1.example.com",
           "latest_triggerer_heartbeat": "2018-12-26T17:16:12+00:00",
-          "team_name": "team-a"
+          "team_names": ["team-a"]
         }
       ]
     },
@@ -130,7 +130,7 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc
 
     * ``hostname``: host where the component is running
     * the corresponding ``latest_*_heartbeat`` for that instance
-    * ``team_name`` (triggerer only): team the triggerer is scoped to, or ``null`` when unscoped
+    * ``team_names`` (triggerer only): teams the triggerer is scoped to, empty when unscoped
     * ``bundle_names`` (Dag processor only): Dag bundles that processor is configured to parse, or ``null`` when unset
 
   * For HA deployments, prefer ``detailed_status`` and ``instances`` when you need to see every scheduler,

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,10 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``3b7a91c5df20`` (head) | ``5182d0596ee2`` | ``3.4.0``         | Fold task_map into xcom.mapped_length.                       |
+| ``c9f4b3e7a218`` (head) | ``3b7a91c5df20`` | ``3.4.0``         | Replace the ``job.team_name`` column with a ``job_team``     |
+|                         |                  |                   | association table.                                           |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``3b7a91c5df20``        | ``5182d0596ee2`` | ``3.4.0``         | Fold task_map into xcom.mapped_length.                       |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``5182d0596ee2``        | ``b6a9c2e7d410`` | ``3.4.0``         | Widen revoked_token.jti to store external-issuer token       |
 |                         |                  |                   | identifiers.                                                 |

airflow-core/src/airflow/api/common/airflow_health.py

@@ -95,7 +95,7 @@ def _legacy_status(jobs: list[Job]) -> HealthStatus:
 def _triggerer_instance_health(job: Job) -> dict[str, Any]:
     return {
         **_job_instance_health(job, "latest_triggerer_heartbeat"),
-        "team_name": job.team_name,
+        "team_names": job.team_names,
     }
 
 
@@ -168,7 +168,13 @@ def _triggerer_detailed_status(jobs: list[Job]) -> DetailedHealthStatus:
     if not expected:
         return _liveness_status(jobs)
 
-    return _coverage_status(expected, {job.team_name for job in jobs if job.is_alive()})
+    covered: set[str | None] = set()
+    for job in jobs:
+        if job.is_alive():
+            # An unscoped triggerer has no ``team_names``, it covers the team-less
+            # scope, represented by ``None`` in both ``expected`` and ``covered``.
+            covered |= set(job.team_names) or {None}
+    return _coverage_status(expected, covered)
 
 
 def get_airflow_health() -> dict[str, Any]:

airflow-core/src/airflow/api_fastapi/core_api/datamodels/job.py

@@ -37,7 +37,7 @@ class JobResponse(BaseModel):
     executor_class: str | None
     hostname: str | None
     unixname: str | None
-    team_name: str | None = None
+    team_names: list[str] = Field(default_factory=list)
     bundle_names: list[str] | None = None
     dag_display_name: str | None = Field(
         validation_alias=AliasPath("dag_model", "dag_display_name"), default=None

airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py

@@ -40,7 +40,7 @@ class TriggererInstanceInfoResponse(BaseModel):
 
     hostname: str | None
     latest_triggerer_heartbeat: str | None
-    team_name: str | None
+    team_names: list[str]
 
 
 class DagProcessorInstanceInfoResponse(BaseModel):

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -4107,11 +4107,11 @@ components:
           - type: string
           - type: 'null'
           title: Unixname
-        team_name:
-          anyOf:
-          - type: string
-          - type: 'null'
-          title: Team Name
+        team_names:
+          items:
+            type: string
+          type: array
+          title: Team Names
         bundle_names:
           anyOf:
           - items:

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -15389,11 +15389,11 @@ components:
           - type: string
           - type: 'null'
           title: Unixname
-        team_name:
-          anyOf:
-          - type: string
-          - type: 'null'
-          title: Team Name
+        team_names:
+          items:
+            type: string
+          type: array
+          title: Team Names
         bundle_names:
           anyOf:
           - items:
@@ -17211,16 +17211,16 @@ components:
           - type: string
           - type: 'null'
           title: Latest Triggerer Heartbeat
-        team_name:
-          anyOf:
-          - type: string
-          - type: 'null'
-          title: Team Name
+        team_names:
+          items:
+            type: string
+          type: array
+          title: Team Names
       type: object
       required:
       - hostname
       - latest_triggerer_heartbeat
-      - team_name
+      - team_names
       title: TriggererInstanceInfoResponse
       description: Triggerer instance info serializer for responses.
     UpdateHITLDetailPayload:

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

@@ -62,7 +62,7 @@ def triggerer_run(
     set_component_mp_start_method("triggerer")
     with _serve_logs(skip_serve_logs):
         triggerer_job_runner = TriggererJobRunner(
-            job=Job(heartrate=triggerer_heartrate, team_name=team_name),
+            job=Job(heartrate=triggerer_heartrate, team_names=[team_name] if team_name else []),
             capacity=capacity,
             queues=queues,
         )

airflow-core/src/airflow/jobs/job.py

@@ -17,14 +17,14 @@
 # under the License.
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from datetime import datetime
 from enum import Enum
 from functools import cached_property, lru_cache
 from time import sleep
 from typing import TYPE_CHECKING, NoReturn
 
-from sqlalchemy import ForeignKey, Index, Integer, String, case, select
+from sqlalchemy import Index, Integer, String, case, select
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm import Mapped, backref, foreign, mapped_column, relationship
 from sqlalchemy.orm.session import make_transient
@@ -35,6 +35,7 @@
 from airflow.exceptions import AirflowException
 from airflow.listeners.listener import get_listener_manager
 from airflow.models.base import ID_LEN, Base
+from airflow.models.team import JobTeam
 from airflow.utils.helpers import convert_camel_to_snake
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.net import get_hostname
@@ -101,9 +102,6 @@ class Job(Base, LoggingMixin):
     executor_class: Mapped[str | None] = mapped_column(String(500))
     hostname: Mapped[str | None] = mapped_column(String(500))
     unixname: Mapped[str | None] = mapped_column(String(1000))
-    team_name: Mapped[str | None] = mapped_column(
-        String(50), ForeignKey("team.name", ondelete="SET NULL"), nullable=True
-    )
     bundle_names: Mapped[list[str] | None] = mapped_column(ExtendedJSON, nullable=True)
 
     __table_args__ = (
@@ -137,6 +135,21 @@ class Job(Base, LoggingMixin):
     Only makes sense for SchedulerJob.
     """
 
+    job_teams = relationship(
+        "JobTeam",
+        cascade="all, delete-orphan",
+        lazy="selectin",
+    )
+
+    @property
+    def team_names(self) -> list[str]:
+        """Names of the teams whose workloads this job serves, empty when it is not team-scoped."""
+        return sorted(job_team.team_name for job_team in self.job_teams)
+
+    @team_names.setter
+    def team_names(self, names: Iterable[str]) -> None:
+        self.job_teams = [JobTeam(team_name=name) for name in dict.fromkeys(names)]
+
     def __init__(self, heartrate=None, **kwargs):
         # Save init parameters as DB fields
         self.heartbeat_failed = False

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -276,7 +276,10 @@ def _execute(self) -> int | None:
                 capacity=self.capacity,
                 logger=log,
                 queues=self.queues,
-                team_name=self.job.team_name,
+                # A triggerer is started with at most one ``--team-name`` for now, so while a Job can
+                # serve several teams this one never does, and the runner stays scoped to a single team.
+                # If we later let a triggerer take several ``--team-name`` values, this must fan out.
+                team_name=next(iter(self.job.team_names), None),
             )
             # Run the main DB comms loop in this process
             self.trigger_runner.run()

airflow-core/src/airflow/migrations/versions/0137_3_4_0_add_job_team_association_table.py

@@ -0,0 +1,81 @@
+#
+# 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.
+
+"""
+Replace the ``job.team_name`` column with a ``job_team`` association table.
+
+A single job can serve more than one team -- a Dag processor parsing bundles that belong to
+several teams is relevant to each of them -- so the association is many-to-many.
+
+Revision ID: c9f4b3e7a218
+Revises: 3b7a91c5df20
+Create Date: 2026-09-22 10:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c9f4b3e7a218"
+down_revision = "3b7a91c5df20"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+    """Replace ``job.team_name`` with the ``job_team`` association table."""
+    # The column goes before the table is created: on SQLite, dropping a column rebuilds the
+    # whole ``job`` table, which a ``job_team`` foreign key pointing at it would not survive.
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f("job_team_name_fkey"), type_="foreignkey")
+        batch_op.drop_column("team_name")
+
+    op.create_table(
+        "job_team",
+        sa.Column("job_id", sa.Integer(), nullable=False),
+        sa.Column("team_name", sa.String(length=50), nullable=False),
+        sa.PrimaryKeyConstraint("job_id", "team_name", name=op.f("job_team_pkey")),
+        sa.ForeignKeyConstraint(
+            columns=("job_id",),
+            refcolumns=["job.id"],
+            name="job_team_job_id_fkey",
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            columns=("team_name",),
+            refcolumns=["team.name"],
+            name="job_team_team_name_fkey",
+            ondelete="CASCADE",
+        ),
+        sa.Index("idx_job_team_team_name", "team_name", unique=False),
+    )
+
+
+def downgrade():
+    """Restore the ``job.team_name`` column."""
+    op.drop_table("job_team")
+
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("team_name", sa.String(length=50), nullable=True))
+        batch_op.create_foreign_key(
+            batch_op.f("job_team_name_fkey"), "team", ["team_name"], ["name"], ondelete="SET NULL"
+        )

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

@@ -20,7 +20,7 @@
 import re
 from typing import TYPE_CHECKING
 
-from sqlalchemy import Column, ForeignKey, Index, String, Table, select
+from sqlalchemy import Column, ForeignKey, Index, Integer, String, Table, select
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from airflow.models.base import Base, StringID
@@ -64,6 +64,28 @@ def find_invalid_team_names(names: Iterable[str]) -> list[str]:
 )
 
 
+class JobTeam(Base):
+    """
+    Association between a Job and a team whose workloads that Job serves.
+
+    Mapped as a class rather than a bare association table so that a Job can be given its
+    teams before it is persisted: writing link rows never needs to load, and therefore can
+    never accidentally insert, a :class:`Team`.
+    """
+
+    __tablename__ = "job_team"
+
+    job_id: Mapped[int] = mapped_column(Integer, ForeignKey("job.id", ondelete="CASCADE"), primary_key=True)
+    team_name: Mapped[str] = mapped_column(
+        String(50), ForeignKey("team.name", ondelete="CASCADE"), primary_key=True
+    )
+
+    __table_args__ = (Index("idx_job_team_team_name", "team_name"),)
+
+    def __repr__(self):
+        return f"JobTeam(job_id={self.job_id}, team_name={self.team_name})"
+
+
 class Team(Base):
     """
     Contains the list of teams defined in the environment.

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -6122,16 +6122,12 @@ export const $JobResponse = {
             ],
             title: 'Unixname'
         },
-        team_name: {
-            anyOf: [
-                {
-                    type: 'string'
-                },
-                {
-                    type: 'null'
-                }
-            ],
-            title: 'Team Name'
+        team_names: {
+            items: {
+                type: 'string'
+            },
+            type: 'array',
+            title: 'Team Names'
         },
         bundle_names: {
             anyOf: [
@@ -8946,20 +8942,16 @@ export const $TriggererInstanceInfoResponse = {
             ],
             title: 'Latest Triggerer Heartbeat'
         },
-        team_name: {
-            anyOf: [
-                {
-                    type: 'string'
-                },
-                {
-                    type: 'null'
-                }
-            ],
-            title: 'Team Name'
+        team_names: {
+            items: {
+                type: 'string'
+            },
+            type: 'array',
+            title: 'Team Names'
         }
     },
     type: 'object',
-    required: ['hostname', 'latest_triggerer_heartbeat', 'team_name'],
+    required: ['hostname', 'latest_triggerer_heartbeat', 'team_names'],
     title: 'TriggererInstanceInfoResponse',
     description: 'Triggerer instance info serializer for responses.'
 } as const;

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -1639,7 +1639,7 @@ export type JobResponse = {
     executor_class: string | null;
     hostname: string | null;
     unixname: string | null;
-    team_name?: string | null;
+    team_names?: Array<(string)>;
     bundle_names?: Array<(string)> | null;
     dag_display_name?: string | null;
 };
@@ -2239,7 +2239,7 @@ export type TriggererInfoResponse = {
 export type TriggererInstanceInfoResponse = {
     hostname: string | null;
     latest_triggerer_heartbeat: string | null;
-    team_name: string | null;
+    team_names: Array<(string)>;
 };
 
 /**