Merge branch 'main' into feature/add-xcoms-keys-route

Author: dabla

.pre-commit-config.yaml

@@ -1014,6 +1014,20 @@ repos:
           ^scripts/ci/prek/generate_agent_skills\.py$
         pass_filenames: false
         additional_dependencies: ['pyyaml', 'rich>=13.6.0']
+      - id: check-documented-commands
+        name: Check documented breeze and prek commands exist
+        entry: ./scripts/ci/prek/check_documented_commands.py
+        language: python
+        files: >
+          (?x)
+          ^AGENTS\.md$|
+          ^contributing-docs/.*\.rst$|
+          ^dev/breeze/src/airflow_breeze/commands/|
+          ^(?:.*/)?\.pre-commit-config\.yaml$|
+          ^scripts/ci/prek/check_documented_commands\.py$
+        pass_filenames: false
+        require_serial: true
+        additional_dependencies: ['pyyaml', 'rich>=13.6.0']
       - id: run-skill-eval
         name: Run skill-eval against AGENTS.md and cases
         entry: ./dev/skill-evals/eval.py

airflow-core/docs/faq.rst

@@ -416,7 +416,7 @@ configuration option:
 * ``error`` — Treats detected issues as Dag import errors, preventing the Dag from loading.
 
 Additionally, you can catch these issues earlier in your development workflow by using the
-`AIR302 <https://docs.astral.sh/ruff/rules/airflow3-dag-dynamic-value/>`_ ruff rule, which detects
+`AIR304 <https://docs.astral.sh/ruff/rules/airflow3-dag-dynamic-value/>`_ ruff rule, which detects
 dynamic values in Dag and Task constructors as part of static linting. See
 :ref:`best_practices/code_quality_and_linting` for how to set up ruff with Airflow-specific rules.
 

airflow-core/src/airflow/api_fastapi/common/db/dags.py

@@ -70,9 +70,14 @@ def generate_dag_with_latest_run_query(
             break
 
     requested_order_by_set = set(order_by.value) if order_by.value is not None else set()
-    dag_run_order_by_set = set(
-        ["last_run_state", "last_run_start_date", "-last_run_state", "-last_run_start_date"],
-    )
+    dag_run_order_by_set = {
+        "last_run_state",
+        "-last_run_state",
+        "last_run_start_date",
+        "-last_run_start_date",
+        "last_run_run_after",
+        "-last_run_run_after",
+    }
 
     if has_max_run_filter or (requested_order_by_set & dag_run_order_by_set):
         query = query.join(

airflow-core/src/airflow/api_fastapi/common/parameters.py

@@ -62,6 +62,7 @@
 from airflow.models.dag import DagModel, DagTag
 from airflow.models.dag_favorite import DagFavorite
 from airflow.models.dag_version import DagVersion
+from airflow.models.dagbundle import DagBundleModel
 from airflow.models.dagrun import DagRun
 from airflow.models.errors import ParseImportError
 from airflow.models.hitl import HITLDetail
@@ -1020,6 +1021,29 @@ def depends(cls, owners: list[str] = Query(default_factory=list)) -> _OwnersFilt
         return cls().set_value(owners)
 
 
+class _TeamsFilter(BaseParam[list[str]]):
+    """Filter Dags by team name (via bundle association)."""
+
+    def to_orm(self, select: Select) -> Select:
+        if self.skip_none is False:
+            raise ValueError(f"Cannot set 'skip_none' to False on a {type(self)}")
+
+        if not self.value:
+            return select
+
+        from airflow.models.team import Team
+
+        return select.where(
+            DagModel.bundle_name.in_(
+                sql_select(DagBundleModel.name).join(DagBundleModel.teams).where(Team.name.in_(self.value))
+            )
+        )
+
+    @classmethod
+    def depends(cls, teams: list[str] = Query(default_factory=list)) -> _TeamsFilter:
+        return cls().set_value(teams)
+
+
 def _safe_parse_datetime(date_to_check: str) -> datetime:
     """
     Parse datetime and raise error for invalid dates.
@@ -1268,6 +1292,7 @@ def depends_float(
 ]
 QueryTagsFilter = Annotated[_TagsFilter, Depends(_TagsFilter.depends)]
 QueryOwnersFilter = Annotated[_OwnersFilter, Depends(_OwnersFilter.depends)]
+QueryTeamsFilter = Annotated[_TeamsFilter, Depends(_TeamsFilter.depends)]
 
 
 class _HasAssetScheduleFilter(BaseParam[bool]):

airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

@@ -32,6 +32,7 @@ class DAGWithLatestDagRunsResponse(DAGResponse):
     latest_dag_runs: list[DAGRunLightResponse]
     pending_actions: list[HITLDetail]
     is_favorite: bool
+    team_name: str | None = None
 
 
 class DAGWithLatestDagRunsCollectionResponse(BaseModel):

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

@@ -338,6 +338,14 @@ paths:
           items:
             type: string
           title: Owners
+      - name: teams
+        in: query
+        required: false
+        schema:
+          type: array
+          items:
+            type: string
+          title: Teams
       - name: dag_ids
         in: query
         required: false
@@ -510,13 +518,13 @@ paths:
             type: string
           description: 'Attributes to order by, multi criteria sort is supported.
             Prefix with `-` for descending order. Supported attributes: `dag_id, dag_display_name,
-            next_dagrun, state, start_date, last_run_state, last_run_start_date`'
+            next_dagrun, state, start_date, last_run_state, last_run_start_date, last_run_run_after`'
           default:
           - dag_id
           title: Order By
         description: 'Attributes to order by, multi criteria sort is supported. Prefix
           with `-` for descending order. Supported attributes: `dag_id, dag_display_name,
-          next_dagrun, state, start_date, last_run_state, last_run_start_date`'
+          next_dagrun, state, start_date, last_run_state, last_run_start_date, last_run_run_after`'
       - name: is_favorite
         in: query
         required: false
@@ -2629,6 +2637,11 @@ components:
         is_favorite:
           type: boolean
           title: Is Favorite
+        team_name:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Team Name
         is_backfillable:
           type: boolean
           title: Is Backfillable

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py

@@ -51,6 +51,7 @@
     QueryPausedFilter,
     QueryPendingActionsFilter,
     QueryTagsFilter,
+    QueryTeamsFilter,
     SortParam,
     filter_param_factory,
 )
@@ -96,6 +97,7 @@ def get_dags(
     offset: QueryOffset,
     tags: QueryTagsFilter,
     owners: QueryOwnersFilter,
+    teams: QueryTeamsFilter,
     dag_ids: Annotated[
         FilterParam[list[str] | None],
         Depends(filter_param_factory(DagModel.dag_id, list[str] | None, FilterOptionEnum.IN, "dag_ids")),
@@ -117,7 +119,11 @@ def get_dags(
             SortParam(
                 ["dag_id", "dag_display_name", "next_dagrun", "state", "start_date"],
                 DagModel,
-                {"last_run_state": DagRun.state, "last_run_start_date": DagRun.start_date},
+                {
+                    "last_run_state": DagRun.state,
+                    "last_run_start_date": DagRun.start_date,
+                    "last_run_run_after": DagRun.run_after,
+                },
             ).dynamic_depends()
         ),
     ],
@@ -153,6 +159,7 @@ def get_dags(
             dag_display_name_prefix_pattern,
             tags,
             owners,
+            teams,
             last_dag_run_state,
             dag_run_state,
             is_favorite,
@@ -228,6 +235,13 @@ def get_dags(
         for dag_id, hitl_detail in pending_actions:
             pending_actions_by_dag_id[dag_id].append(hitl_detail)
 
+    # Fetch team names when multi-team is enabled
+    team_names_by_dag_id: dict[str, str | None] = {}
+    if conf.getboolean("core", "multi_team") and dags:
+        team_names_by_dag_id = DagModel.get_dag_id_to_team_name_mapping(
+            [dag.dag_id for dag in dags], session=session
+        )
+
     # aggregate rows by dag_id
     # Build the dict dynamically from DAGResponse.model_fields so that new fields
     # added to DAGResponse are picked up automatically without code changes here.
@@ -245,6 +259,7 @@ def get_dags(
                 "latest_dag_runs": [],
                 "pending_actions": pending_actions_by_dag_id[dag.dag_id],
                 "is_favorite": dag.dag_id in favorite_dag_ids,
+                "team_name": team_names_by_dag_id.get(dag.dag_id),
             }
         )
         dag_runs_by_dag_id[dag.dag_id] = DAGWithLatestDagRunsResponse.model_validate(dag_data)

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

@@ -30,7 +30,7 @@
 from airflow import settings
 from airflow._shared.timezones import timezone
 from airflow.cli.simple_table import AirflowConsole
-from airflow.cli.utils import fetch_dag_run_from_run_id_or_logical_date_string
+from airflow.cli.utils import deprecated_for_airflowctl, fetch_dag_run_from_run_id_or_logical_date_string
 from airflow.exceptions import AirflowConfigException, DagRunNotFound, NotMapped, TaskInstanceNotFound
 from airflow.models import TaskInstance
 from airflow.models.dag_version import DagVersion
@@ -354,6 +354,7 @@ def _guess_debugger() -> _SupportedDebugger:
     raise exc
 
 
+@deprecated_for_airflowctl("airflowctl tasks states-for-dag-run")
 @cli_utils.action_cli(check_db=False)
 @suppress_logs_and_warning
 @providers_configuration_loaded