Sync v3-1-stable with v3-1-test to release 3.1.4rc2

RELEASE_NOTES.rst

@@ -24,7 +24,7 @@
 
 .. towncrier release notes start
 
-Airflow 3.1.4 (2025-12-09)
+Airflow 3.1.4 (2025-12-10)
 --------------------------
 
 Significant Changes
@@ -35,7 +35,8 @@ No significant changes.
 Bug Fixes
 ^^^^^^^^^
 
-Fix task clearing to only find relevant upstream/downstream task instances (#58987)
+Handle invalid token in JWTRefreshMiddleware (#56904)
+Fix inconsistent Dag hashes when template fields contain unordered dicts (#59091) (#59175)
 Fix assets used only as inlets being incorrectly orphaned (#58986)
 Fix exception when logging stdout with a custom %-format string (#58963)
 Fix backfill max_active_runs race condition with concurrent schedulers (#58935)

airflow-core/src/airflow/api_fastapi/auth/middlewares/refresh_token.py

@@ -17,7 +17,8 @@
 # under the License.
 from __future__ import annotations
 
-from fastapi import Request
+from fastapi import HTTPException, Request
+from fastapi.responses import JSONResponse
 from starlette.middleware.base import BaseHTTPMiddleware
 
 from airflow.api_fastapi.app import get_auth_manager
@@ -41,28 +42,34 @@ class JWTRefreshMiddleware(BaseHTTPMiddleware):
     async def dispatch(self, request: Request, call_next):
         new_user = None
         current_token = request.cookies.get(COOKIE_NAME_JWT_TOKEN)
-        if current_token:
-            new_user = await self._refresh_user(current_token)
-            if new_user:
-                request.state.user = new_user
-
-        response = await call_next(request)
+        try:
+            if current_token:
+                new_user = await self._refresh_user(current_token)
+                if new_user:
+                    request.state.user = new_user
 
-        if new_user:
-            # If we created a new user, serialize it and set it as a cookie
-            new_token = get_auth_manager().generate_jwt(new_user)
-            secure = bool(conf.get("api", "ssl_cert", fallback=""))
-            response.set_cookie(
-                COOKIE_NAME_JWT_TOKEN,
-                new_token,
-                httponly=True,
-                secure=secure,
-                samesite="lax",
-            )
+            response = await call_next(request)
 
+            if new_user:
+                # If we created a new user, serialize it and set it as a cookie
+                new_token = get_auth_manager().generate_jwt(new_user)
+                secure = bool(conf.get("api", "ssl_cert", fallback=""))
+                response.set_cookie(
+                    COOKIE_NAME_JWT_TOKEN,
+                    new_token,
+                    httponly=True,
+                    secure=secure,
+                    samesite="lax",
+                )
+        except HTTPException as exc:
+            # If any HTTPException is raised during user resolution or refresh, return it as response
+            return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
         return response
 
     @staticmethod
     async def _refresh_user(current_token: str) -> BaseUser | None:
-        user = await resolve_user_from_token(current_token)
+        try:
+            user = await resolve_user_from_token(current_token)
+        except HTTPException:
+            return None
         return get_auth_manager().refresh_user(user=user)

airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py

@@ -729,54 +729,32 @@ def post_clear_task_instances(
     if future:
         body.end_date = None
 
-    if (task_markers_to_clear := body.task_ids) is not None:
-        mapped_tasks_tuples = {t for t in task_markers_to_clear if isinstance(t, tuple)}
+    task_ids = body.task_ids
+    if task_ids is not None:
+        tasks = set(task_ids)
+        mapped_tasks_tuples = set(t for t in tasks if isinstance(t, tuple))
         # Unmapped tasks are expressed in their task_ids (without map_indexes)
-        normal_task_ids = {t for t in task_markers_to_clear if not isinstance(t, tuple)}
-
-        def _collect_relatives(run_id: str, direction: Literal["upstream", "downstream"]) -> None:
-            from airflow.models.taskinstance import find_relevant_relatives
-
-            relevant_relatives = find_relevant_relatives(
-                normal_task_ids,
-                mapped_tasks_tuples,
-                dag=dag,
-                run_id=run_id,
-                direction=direction,
-                session=session,
+        unmapped_task_ids = set(t for t in tasks if not isinstance(t, tuple))
+
+        if upstream or downstream:
+            mapped_task_ids = set(tid for tid, _ in mapped_tasks_tuples)
+            relatives = dag.partial_subset(
+                task_ids=unmapped_task_ids | mapped_task_ids,
+                include_downstream=downstream,
+                include_upstream=upstream,
+                exclude_original=True,
             )
-            normal_task_ids.update(t for t in relevant_relatives if not isinstance(t, tuple))
-            mapped_tasks_tuples.update(t for t in relevant_relatives if isinstance(t, tuple))
-
-        # We can't easily calculate upstream/downstream map indexes when not
-        # working for a specific dag run. It's possible by looking at the runs
-        # one by one, but that is both resource-consuming and logically complex.
-        # So instead we'll just clear all the tis based on task ID and hope
-        # that's good enough for most cases.
-        if dag_run_id is None:
-            if upstream or downstream:
-                partial_dag = dag.partial_subset(
-                    task_ids=normal_task_ids.union(tid for tid, _ in mapped_tasks_tuples),
-                    include_downstream=downstream,
-                    include_upstream=upstream,
-                    exclude_original=True,
-                )
-                normal_task_ids.update(partial_dag.task_dict)
-        else:
-            if upstream:
-                _collect_relatives(dag_run_id, "upstream")
-            if downstream:
-                _collect_relatives(dag_run_id, "downstream")
-
-        task_markers_to_clear = [
-            *normal_task_ids,
-            *((t, m) for t, m in mapped_tasks_tuples if t not in normal_task_ids),
+            unmapped_task_ids = unmapped_task_ids | set(relatives.task_dict.keys())
+
+        mapped_tasks_list = [
+            (tid, map_id) for tid, map_id in mapped_tasks_tuples if tid not in unmapped_task_ids
         ]
+        task_ids = mapped_tasks_list + list(unmapped_task_ids)
 
     # Prepare common parameters
     common_params = {
         "dry_run": True,
-        "task_ids": task_markers_to_clear,
+        "task_ids": task_ids,
         "session": session,
         "run_on_latest_version": body.run_on_latest_version,
         "only_failed": body.only_failed,

airflow-core/src/airflow/example_dags/example_dynamic_task_mapping.py

@@ -22,9 +22,9 @@
 # [START example_dynamic_task_mapping]
 from datetime import datetime
 
-from airflow.sdk import DAG, task, task_group
+from airflow.sdk import DAG, task
 
-with DAG(dag_id="example_dynamic_task_mapping", schedule=None, start_date=datetime(2022, 3, 4)):
+with DAG(dag_id="example_dynamic_task_mapping", schedule=None, start_date=datetime(2022, 3, 4)) as dag:
 
     @task
     def add_one(x: int):
@@ -39,11 +39,8 @@ def sum_it(values):
     sum_it(added_values)
 
 with DAG(
-    dag_id="example_task_mapping_second_order",
-    schedule=None,
-    catchup=False,
-    start_date=datetime(2022, 3, 4),
-):
+    dag_id="example_task_mapping_second_order", schedule=None, catchup=False, start_date=datetime(2022, 3, 4)
+) as dag2:
 
     @task
     def get_nums():
@@ -61,25 +58,4 @@ def add_10(num):
     _times_2 = times_2.expand(num=_get_nums)
     add_10.expand(num=_times_2)
 
-with DAG(
-    dag_id="example_task_group_mapping",
-    schedule=None,
-    catchup=False,
-    start_date=datetime(2022, 3, 4),
-):
-
-    @task_group
-    def op(num):
-        @task
-        def add_1(num):
-            return num + 1
-
-        @task
-        def mul_2(num):
-            return num * 2
-
-        return mul_2(add_1(num))
-
-    op.expand(num=[1, 2, 3])
-
 # [END example_dynamic_task_mapping]