You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ashb (via GitHub)" <gi...@apache.org> on 2026/10/06 15:07:35 UTC
[PR] Read current task executions by exact region and try [airflow]
ashb opened a new pull request, #74350:
URL: https://github.com/apache/airflow/pull/74350
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read current task executions by exact region and try [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb merged PR #74350:
URL: https://github.com/apache/airflow/pull/74350
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read current task executions by exact region and try [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #74350:
URL: https://github.com/apache/airflow/pull/74350#discussion_r4238418534
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/execution.py:
##########
@@ -0,0 +1,116 @@
+# 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.
+from __future__ import annotations
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import Depends, HTTPException, Query, status
+from sqlalchemy import select
+
+from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.execution import (
+ ExecutionCollectionResponse,
+ ExecutionRegionResponse,
+ ExecutionTaskResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import DagAccessEntity, requires_access_dag
+from airflow.api_fastapi.core_api.services.public.execution import get_execution_members
+from airflow.models.dagrun import DagRun
+from airflow.models.dynamic_region import SENTINEL_REGION_ID, load_region_ancestry
+
+execution_router = AirflowRouter(
+ tags=["DagRun"],
+ prefix="/dags/{dag_id}/dagRuns/{dag_run_id}",
+)
+
+
+@execution_router.get(
+ "/execution",
Review Comment:
Going to park this one for now. I think it does, but want a little bit more thinking time before committing this to the public API
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read current task executions by exact region and try [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74350:
URL: https://github.com/apache/airflow/pull/74350#discussion_r4198251299
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -1491,5 +1579,10 @@ def delete_task_instance(
)
TI.delete_attempts(
- dag_id=dag_id, run_id=dag_run_id, task_id=task_id, map_index=map_index, session=session
+ dag_id=dag_id,
+ run_id=dag_run_id,
+ task_id=task_id,
+ map_index=map_index,
Review Comment:
The existence check above uses `scope.region_index`, but this passes the raw `map_index` query param, which defaults to -1. For a loop pass addressed as `?region_id=R®ion_index=2`, the lookup finds the row and then `delete_attempts` deletes `region_index = -1` in R, which matches nothing, so the endpoint reports success and the pass stays. A mapped slot addressed by region coordinates hits the same thing. Should this be `map_index=scope.region_index`? A delete test with region coordinates would catch it.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -393,7 +427,8 @@ def get_task_instance_tries(
TI.dag_id == dag_id,
TI.run_id == dag_run_id,
TI.task_id == task_id,
- TI.map_index == map_index,
+ TI.region_index == scope.region_index,
+ TI.region_id == scope.region_id,
Review Comment:
Before this, `/tries` listed every attempt at the map index whatever its region. Now it only sees the region the live producer resolves to. For a run from 3.3, clearing a whole mapped task archives the sentinel-region rows and re-expands into a new region, so the pre-upgrade attempts and their logs drop out of `/tries` unless the client already knows to pass `region_id=00000000-0000-0000-0000-000000000000`, and nothing in the response points there. Is that intended? If so, maybe list archived attempts from the sentinel region as well, or at least document it on the route.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/log.py:
##########
@@ -196,7 +206,9 @@ def get_external_log_url(
TaskInstance.task_id == task_id,
TaskInstance.dag_id == dag_id,
TaskInstance.run_id == dag_run_id,
- TaskInstance.map_index == map_index,
+ TaskInstance.region_index == scope.region_index,
+ TaskInstance.region_id == scope.region_id,
+ TaskInstance.try_number == try_number,
Review Comment:
This now filters on `try_number`, but unlike `get_log` above it doesn't set `.execution_options(include_all_attempts=True)`, so the `working_set IS TRUE` criteria added by `_restrict_to_current_attempts` hides every archived try. After a retry, `externalLogUrl/1` returns 404 where it used to return a link (the old query ignored the try and used the live row). As far as I can trace, `test_execution_lists_live_work_and_selects_archived_tries` can't pass on this, and the `history` lookup in `test_external_link_uses_the_selected_archived_try` goes through the same filter, so it returns None after the setup's `dag.clear()` and needs the option too.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -668,14 +734,18 @@ def get_task_instances(
)
load_legacy_rendered_fields(task_instances, session=session)
return TaskInstanceCollectionResponse(
- task_instances=task_instances,
+ task_instances=[
+ task_coordinate_response(TaskInstanceResponse, ti, resolver) for ti in task_instances
+ ],
total_entries=total_entries,
total_entries_limit=total_entries_limit,
next_cursor=(
- encode_cursor(task_instances[-1], order_by) if has_next and task_instances else None
+ encode_cursor(TaskCoordinateView(task_instances[-1], resolver), order_by)
Review Comment:
The cursor is encoded through `TaskCoordinateView`, so it uses `resolver.public_map_index` (the pinned definition's `get_needs_expansion()`). The ordering and the `map_index` filter use `public_map_index_expression` (region `node_id == task_id`), and `get_execution` has a third inline copy of the region rule. They agree today, but a mapped slot cleared with run-on-latest after `.expand()` was removed keeps its region and index while being re-pinned to an unmapped definition. SQL then sorts and filters it as 3 while the response and cursor say -1, and the next page can repeat rows. Could the response side use the stored-region rule as well (`_is_mapped_region` already does that), so the two agree by construction?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/execution.py:
##########
@@ -0,0 +1,116 @@
+# 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.
+from __future__ import annotations
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import Depends, HTTPException, Query, status
+from sqlalchemy import select
+
+from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.execution import (
+ ExecutionCollectionResponse,
+ ExecutionRegionResponse,
+ ExecutionTaskResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import DagAccessEntity, requires_access_dag
+from airflow.api_fastapi.core_api.services.public.execution import get_execution_members
+from airflow.models.dagrun import DagRun
+from airflow.models.dynamic_region import SENTINEL_REGION_ID, load_region_ancestry
+
+execution_router = AirflowRouter(
+ tags=["DagRun"],
+ prefix="/dags/{dag_id}/dagRuns/{dag_run_id}",
+)
+
+
+@execution_router.get(
+ "/execution",
Review Comment:
Does this need to be a public v2 endpoint? The UI is its only consumer, and `ExecutionRegionResponse` commits `forked_from_region_id`, `resumes_from_index` and `parent_region_index` to the stable API while the loop-clear model is still moving. The TI listing already accepts and returns `region_id`/`region_index`, so the new data here is mainly the ancestry. Serving that from the private UI API would let the region model change before the release ships without a v2 deprecation.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py:
##########
@@ -111,3 +113,34 @@ def resolve_task_scope(
map_index=tasks[0].region_index,
region_id=tasks[0].region_id,
)
+
+
+def _coordinate_resolver(dag_bag: DagBagDep, session: SessionDep) -> TaskCoordinateResolver:
+ return TaskCoordinateResolver(dag_bag, session)
+
+
+CoordinateResolverDep = Annotated[TaskCoordinateResolver, Depends(_coordinate_resolver)]
+
+
+def _task_scope(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ resolver: CoordinateResolverDep,
+ map_index: int = -1,
+ region_id: Annotated[UUID | None, Query()] = None,
+ region_index: Annotated[int | None, Query(ge=-1)] = None,
+) -> TaskScope:
+ return resolve_task_scope(
+ dag_id=dag_id,
+ run_id=dag_run_id,
+ task_id=task_id,
+ session=resolver.session,
+ dag_bag=resolver.dag_bag,
+ map_index=map_index,
Review Comment:
Two things about the addressing rules now that this dependency sits on the `/{map_index}` routes. When both `region_id` and `region_index` are sent, `resolve_task_scope` returns before looking at `map_index`, so `GET .../taskInstances/m/7?region_id=M®ion_index=2` quietly returns slot 2 (the PATCH path rejects conflicting body and query coordinates with a 400). And for a loop task with no pass selected, as far as I can trace the client gets 400 without `region_id` but 409 "Multiple live producers" with `region_id` alone, on routes where 409 otherwise means a state conflict (HITL's "already been updated", for example). Maybe reject `map_index not in (-1, region_index)` with a 400, and return 400 for the missing-index case in both shapes?
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/gantt.py:
##########
@@ -26,6 +27,10 @@
class GanttTaskInstance(BaseModel):
"""Task instance data for Gantt chart."""
+ id: UUID
+ region_id: UUID
+ region_index: int
+ map_index: int
Review Comment:
These four required fields break the exact-dict assertion in `test_gantt.py::test_should_response_200`, and through the regenerated `GanttTaskInstance` type, the uncast `allTries` literals in `layouts/Details/Gantt/utils.test.ts` (tsc runs in `pnpm lint`). Both are fixed later in #74352, so this commit is red on its own. Separately, `map_index` is always -1 here since the query filters on `public_map_index_expression(...) == -1`, and the UI only reads `id` and the region fields. Could it be dropped?
##########
airflow-core/src/airflow/models/xcom.py:
##########
@@ -267,6 +267,7 @@ def set(
task_id: str,
run_id: str,
map_index: int = -1,
+ region_id: UUID | None = SENTINEL_REGION_ID,
Review Comment:
Is `None` meant to be allowed here? With `region_id=None`, `select_producers` drops the region filter and `session.scalar` takes whichever live owner comes first among regions that share the map index. The only caller passes a concrete `scope.region_id`, so `region_id: UUID = SENTINEL_REGION_ID` would close that off. The docstring's param list is also missing `region_id`.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/extra_links.py:
##########
@@ -65,19 +69,16 @@ def get_extra_links(
task_id: str,
session: SessionDep,
dag_bag: DagBagDep,
- map_index: int = -1,
+ scope: TaskScopeDep,
Review Comment:
Moving `map_index` into `TaskScopeDep` changes its position in the generated spec. FastAPI emits the route's own query params first, so `get_extra_links` is now `try_number, map_index, ...` while the released Python client has `get_extra_links(dag_id, dag_run_id, task_id, map_index, try_number)`. After a client regen, a positional call like `get_extra_links("d", "r", "t", 3, 2)` silently swaps slot and try, since both are optional ints. `get_xcom_entry`, `get_log` and `clear_task_state_store` reorder too, though those fail on type instead. Could `try_number` move into a dependency declared after `scope`, so `map_index` stays ahead of it?
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py:
##########
@@ -111,3 +113,34 @@ def resolve_task_scope(
map_index=tasks[0].region_index,
region_id=tasks[0].region_id,
)
+
+
+def _coordinate_resolver(dag_bag: DagBagDep, session: SessionDep) -> TaskCoordinateResolver:
+ return TaskCoordinateResolver(dag_bag, session)
+
+
+CoordinateResolverDep = Annotated[TaskCoordinateResolver, Depends(_coordinate_resolver)]
+
+
+def _task_scope(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ resolver: CoordinateResolverDep,
+ map_index: int = -1,
+ region_id: Annotated[UUID | None, Query()] = None,
+ region_index: Annotated[int | None, Query(ge=-1)] = None,
+) -> TaskScope:
+ return resolve_task_scope(
+ dag_id=dag_id,
+ run_id=dag_run_id,
+ task_id=task_id,
+ session=resolver.session,
Review Comment:
`_task_scope` takes the request's `TaskCoordinateResolver` only to unpack its session and dag bag, and `resolve_task_scope` then builds a second one. Routes that also inject `CoordinateResolverDep` end up with two resolvers and two sets of caches per request. Could `resolve_task_scope` take the resolver instead?
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py:
##########
@@ -154,6 +172,130 @@ def _create_xcom(self, key, value, backend=None) -> None:
class TestGetXComEntry(TestXComEndpoint):
+ @pytest.mark.parametrize("mapped", [False, True])
+ def test_regional_xcom_projects_public_map_index(self, test_client, dag_maker, session, mapped):
+ region = DynamicRegion(dag_id=TEST_DAG_ID, run_id=run_id, node_id=TEST_TASK_ID if mapped else "loop")
+ session.add(region)
+ session.flush()
+ _create_regional_ti(dag_maker, session, region_id=region.id, region_index=3)
+ XComModel.set(
+ dag_id=TEST_DAG_ID,
+ run_id=run_id,
+ task_id=TEST_TASK_ID,
+ region_id=region.id,
+ map_index=3,
+ key=TEST_XCOM_KEY,
+ value="live",
+ serialize=False,
+ session=session,
+ )
+ session.commit()
+ url = f"/dags/{TEST_DAG_ID}/dagRuns/{run_id}/taskInstances/{TEST_TASK_ID}/xcomEntries"
+ expected_index = 3 if mapped else -1
+ response = test_client.get(
+ f"{url}/{TEST_XCOM_KEY}", params={"region_id": str(region.id), "region_index": 3}
+ )
+ assert response.status_code == 200
+ assert response.json()["map_index"] == expected_index
+ assert response.json()["region_index"] == 3
+ response = test_client.get(url, params={"map_index_filter": expected_index})
+ assert response.status_code == 200
+ assert response.json()["total_entries"] == 1
+ assert response.json()["xcom_entries"][0]["map_index"] == expected_index
+
+ @pytest.mark.parametrize(
+ ("params", "expected_indexes", "total"),
+ [
+ ({"map_index": -1}, [-1], 1),
+ ({"map_index_filter": 0}, [0], 1),
+ ({"order_by": "map_index", "limit": 1, "offset": 1}, [0], 3),
+ ],
+ )
+ def test_collection_projects_loop_and_mapped_group_before_pagination(
+ self, test_client, dag_maker, session, params, expected_indexes, total
+ ):
+ @task_group
+ def body():
+ EmptyOperator(task_id="member")
+
+ @task_group
+ def mapped_group(value):
+ EmptyOperator(task_id="member")
+
+ with dag_maker("regional-xcom", serialized=True):
+ loop = create_loop(body, max_iterations=7)
+ mapped_group.expand(value=[1, 2])
+ dr = dag_maker.create_dagrun()
+ regions = {}
+ for ti in dr.task_instances:
+ if ti.task_id == loop.gate_task_id:
+ continue
+ node_id = loop.group_id if ti.task_id == "body.member" else ti.task_id
+ if node_id not in regions:
+ region = DynamicRegion(dag_id=dr.dag_id, run_id=dr.run_id, node_id=node_id)
+ session.add(region)
+ session.flush()
+ regions[node_id] = region.id
+ ti.region_id = regions[node_id]
+ if ti.task_id == "body.member":
+ ti.map_index = 5
Review Comment:
`ti.map_index = 5` writes `region_index` through the synonym, and the test then asserts `map_index == -1`. Writing `ti.region_index = 5` would say what's meant, and moving the `region_index` expectation into a parametrize column would drop the `if expected_indexes == [-1]` branch so every case checks it. The regional test in `test_task_instances.py` builds its loop pass with `map_index=2` the same way.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py:
##########
@@ -57,6 +61,61 @@
pytestmark = pytest.mark.db_test
+
+@pytest.mark.parametrize("operation", ["get", "history", "list", "respond"])
+def test_hitl_selects_loop_iteration(test_client, dag_maker, session, operation):
+ @task_group
+ def body():
+ EmptyOperator(task_id="work")
+
+ with dag_maker(serialized=True) as dag:
+ loop = create_loop(body, max_iterations=4)
+ run = dag_maker.create_dagrun()
+ region = session.scalar(select(DynamicRegion).where(DynamicRegion.node_id == loop.group_id))
+ later = TIModel(
+ task=dag.get_task("body.work"),
+ run_id=run.run_id,
+ dag_version_id=run.created_dag_version_id,
+ region_id=region.id,
Review Comment:
Only one region holds `body.work` here, so dropping `TI.region_id == scope.region_id` from the HITL readers would keep this green. The extra-links tests are in the same spot, and `get_log` content, `/{map_index}` and delete have no region test at all. A second region with the same `(task_id, region_index)`, like `test_exact_region_read_update_and_delete_preserve_sibling` in `test_xcom.py`, would bind it. Splitting the four `operation` branches into separate tests would also make each case easier to follow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read current task executions by exact region and try [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74350:
URL: https://github.com/apache/airflow/pull/74350#discussion_r4222311349
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -1496,5 +1585,10 @@ def delete_task_instance(
)
TI.delete_attempts(
- dag_id=dag_id, run_id=dag_run_id, task_id=task_id, map_index=map_index, session=session
+ dag_id=dag_id,
+ run_id=dag_run_id,
+ task_id=task_id,
+ map_index=map_index,
Review Comment:
This still passes the raw `map_index` query param (default -1), while the lookup above matched on `scope.region_index`. So `DELETE .../taskInstances/t?region_id=R®ion_index=2` finds the pass, then `delete_attempts` deletes `region_index = -1` in R, which matches nothing, and the endpoint reports success with the pass still there. `map_index=scope.region_index` fixes it, and `TestDeleteTaskInstance` has no case with region coordinates that would catch it. The route's `responses` also still lists only 404 though the dependency raises 400 and 409 (same on `/{task_id}/dependencies`).
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -398,7 +433,8 @@ def get_task_instance_tries(
TI.dag_id == dag_id,
TI.run_id == dag_run_id,
TI.task_id == task_id,
- TI.map_index == map_index,
+ TI.region_index == scope.region_index,
+ TI.region_id == scope.region_id,
Review Comment:
`/tries` still only lists attempts in `scope.region_id`. For a run from before regions, a mapped task's old attempts stay in the sentinel region (archived as superseded) once it re-expands into a region, so they and their logs drop out of `/tries` unless the client passes `region_id=00000000-0000-0000-0000-000000000000`, and the docstring doesn't mention it. Could the sentinel-region attempts for the same index be included, or at least documented on the route?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -673,14 +740,18 @@ def get_task_instances(
)
load_legacy_rendered_fields(task_instances, session=session)
return TaskInstanceCollectionResponse(
- task_instances=task_instances,
+ task_instances=[
+ task_coordinate_response(TaskInstanceResponse, ti, resolver) for ti in task_instances
+ ],
total_entries=total_entries,
total_entries_limit=total_entries_limit,
next_cursor=(
- encode_cursor(task_instances[-1], order_by) if has_next and task_instances else None
+ encode_cursor(TaskCoordinateView(task_instances[-1], resolver), order_by)
Review Comment:
The cursor still goes through `TaskCoordinateView`, so it uses `resolver.public_map_index` (the pinned definition's `get_needs_expansion()`), while the sort and the `map_index` filter use `public_map_index_expression` (stored region `node_id == task_id`), and `get_execution` has a third inline copy at `execution.py:95`. A mapped slot re-pinned to a definition without `.expand()` sorts as its index in SQL but encodes -1 in the cursor, so the next page can repeat rows. Selecting `public_map_index_expression` as a labelled column and using it for both the response and the cursor would make them agree by construction, and would also skip the pinned-Dag loads `public_map_index` does per `dag_version` on each page.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/extra_links.py:
##########
@@ -65,19 +69,16 @@ def get_extra_links(
task_id: str,
session: SessionDep,
dag_bag: DagBagDep,
- map_index: int = -1,
+ scope: TaskScopeDep,
Review Comment:
`get_extra_links` still lists `try_number` ahead of `map_index` in the generated spec, so after a client regen a positional `get_extra_links("d", "r", "t", 3, 2)` silently swaps slot and try. `get_xcom_entry` (`map_index` now after `stringify`), `get_log` (after `token`) and `clear_task_state_store` (`all_map_indices` now last) reorder too. Moving the trailing params into a dependency declared after `scope` would keep the released order.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py:
##########
@@ -113,3 +110,33 @@ def resolve_task_scope(
map_index=tasks[0].region_index,
region_id=tasks[0].region_id,
)
+
+
+def _coordinate_resolver(dag_bag: DagBagDep, session: SessionDep) -> TaskCoordinateResolver:
+ return TaskCoordinateResolver(dag_bag, session)
+
+
+CoordinateResolverDep = Annotated[TaskCoordinateResolver, Depends(_coordinate_resolver)]
+
+
+def _task_scope(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ resolver: CoordinateResolverDep,
+ map_index: int = -1,
Review Comment:
`map_index` lost its `ge=-1` bound when it moved in here, so `minimum: -1` is gone from seven operations in the spec (get/delete XCom entry and the five task state store routes), and `map_index=-5` now returns 404 instead of 422. `get_xcoms` and `_resolve_clear_scope` still declare `Query(ge=-1)`, so the siblings disagree. `Query()` can't go on this param since it also binds the `{map_index}` path segment, but an explicit `map_index < -1` check in `resolve_task_scope` would keep the bound.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -141,52 +152,52 @@
task_instances_prefix = "/dagRuns/{dag_run_id}/taskInstances"
-def _coordinate_resolver(dag_bag: DagBagDep, session: SessionDep) -> TaskCoordinateResolver:
- return TaskCoordinateResolver(dag_bag, session)
-
-
-CoordinateResolverDep = Annotated[TaskCoordinateResolver, Depends(_coordinate_resolver)]
-
-
@task_instances_router.get(
task_instances_prefix + "/{task_id}",
- responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ responses=create_openapi_http_exception_doc(
+ [status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT]
+ ),
dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.TASK_INSTANCE))],
)
def get_task_instance(
dag_id: str,
dag_run_id: str,
task_id: str,
session: SessionDep,
+ scope: TaskScopeDep,
Review Comment:
With `TaskScopeDep` here, the unmapped `GET .../taskInstances/{task_id}` now publishes a `map_index` query param (it's in the spec and in `GetTaskInstanceData.mapIndex`). For a mapped task, `?map_index=3` resolves slot 3 and then hits the "Task instance is mapped, add the map_index value to the URL" 404 below, which is confusing when the client did send one. Should this route take only `region_id`/`region_index` and leave `map_index` to `/{task_id}/{map_index}`?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/execution.py:
##########
@@ -0,0 +1,116 @@
+# 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.
+from __future__ import annotations
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import Depends, HTTPException, Query, status
+from sqlalchemy import select
+
+from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.execution import (
+ ExecutionCollectionResponse,
+ ExecutionRegionResponse,
+ ExecutionTaskResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import DagAccessEntity, requires_access_dag
+from airflow.api_fastapi.core_api.services.public.execution import get_execution_members
+from airflow.models.dagrun import DagRun
+from airflow.models.dynamic_region import SENTINEL_REGION_ID, load_region_ancestry
+
+execution_router = AirflowRouter(
+ tags=["DagRun"],
+ prefix="/dags/{dag_id}/dagRuns/{dag_run_id}",
+)
+
+
+@execution_router.get(
+ "/execution",
Review Comment:
This is still on the public v2 router, so `ExecutionRegionResponse` commits `parent_region_index`, `forked_from_region_id` and `resumes_from_index` to the stable API, and the Run execution page in the UI stays its only consumer through the rest of the stack. The new `region_id`/`region_index` params also have no description, so nothing in the API docs says what a region is or where a client gets one. Could this live on the private UI router until the region model settles?
##########
airflow-core/src/airflow/api_fastapi/common/parameters/misc.py:
##########
@@ -245,7 +246,7 @@ def _optional_boolean(value: bool | None) -> bool | None:
FilterParam[int | None],
Depends(
filter_param_factory(
- TaskInstance.region_index,
+ public_map_index_expression(TaskInstance),
Review Comment:
With the HITL `map_index` filter on `public_map_index_expression`, every loop pass reads as -1, and `get_hitl_details` has no `region_id`/`region_index` filters, so a client can't list the HITL details for one pass of a loop task anymore (`?map_index=2` used to select pass 2 through `region_index`). Could the list route take the same `region_id`/`region_index` filters the TI list routes got?
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_log.py:
##########
@@ -49,6 +55,81 @@ class TestTaskInstancesLog:
default_time = "2020-06-10T20:00:00+00:00"
+ def _place_tries_in_sibling_regions(self, session):
+ history = session.scalar(
+ select(TaskInstance)
+ .where(
+ TaskInstance.dag_id == self.DAG_ID,
+ TaskInstance.run_id == self.RUN_ID,
+ TaskInstance.task_id == self.TASK_ID,
+ TaskInstance.try_number == 1,
+ )
+ .execution_options(include_all_attempts=True)
+ )
+ current = session.scalar(
+ select(TaskInstance).where(
+ TaskInstance.dag_id == self.DAG_ID,
+ TaskInstance.run_id == self.RUN_ID,
+ TaskInstance.task_id == self.TASK_ID,
+ TaskInstance.working_set.is_(True),
+ )
+ )
+ previous_region, current_region = uuid4(), uuid4()
+ for region_id, source in ((previous_region, None), (current_region, previous_region)):
+ session.add(
+ DynamicRegion(
+ id=region_id,
+ dag_id=self.DAG_ID,
+ run_id=self.RUN_ID,
+ node_id="loop",
+ forked_from_region_id=source,
+ )
+ )
+ history.region_id, history.region_index = previous_region, 2
+ current.region_id, current.region_index, current.try_number = current_region, 2, 2
+ session.commit()
+ return history, current, previous_region, current_region
+
+ def test_external_link_uses_the_selected_archived_try(self, session):
+ history, current, previous_region, _ = self._place_tries_in_sibling_regions(session)
+ with mock.patch(
+ "airflow.api_fastapi.core_api.routes.public.log.TaskLogReader", autospec=True
+ ) as reader:
+ reader.return_value.supports_external_link = True
+ reader.return_value.log_handler = mock.create_autospec(ExternalLoggingMixin, instance=True)
+ reader.return_value.log_handler.get_external_log_url.return_value = "https://logs.example/old"
+ response = self.client.get(
+ f"/dags/{self.DAG_ID}/dagRuns/{self.RUN_ID}/taskInstances/{self.TASK_ID}/externalLogUrl/1",
+ params={"region_id": str(previous_region), "region_index": 2},
Review Comment:
Nothing here pins the `try_number` predicate on `externalLogUrl`: try 1 is the only row in `previous_region`, so this passes with `TaskInstance.try_number == try_number` deleted. Putting both tries in the same region and requesting `/1` and `/2` would bind it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org