You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "uranusjr (via GitHub)" <gi...@apache.org> on 2025/05/02 09:19:32 UTC
[PR] Implement slice on LazyXComSequence [airflow]
uranusjr opened a new pull request, #50117:
URL: https://github.com/apache/airflow/pull/50117
Following up https://github.com/apache/airflow/pull/50011#issuecomment-2841960775
I decided to split index and slice access to their separate endpoints, instead of reusing the GetXCom endpoint. This duplicates code a bit, but the input parameters are now a lot easier to reason with. It's unfortunate FastAPI does not natively allow unions on Query(), or this could be implemented a lot nicer.
* [ ] Fix tests on index access
* [ ] Add tests for slice access
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2120361961
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
Review Comment:
Handled in [Implement slice on LazyXComSequence](https://github.com/apache/airflow/pull/50117/commits/84b36c4a38daa9fc7e9ced598abdb973cdbf525d)
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2111078514
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -470,7 +471,27 @@ def get_sequence_item(
},
)
raise
- return XComResponse.model_validate_json(resp.read())
+ return XComSequenceIndexResponse.model_validate_json(resp.read())
+
+ def get_sequence_slice(
+ self,
+ dag_id: str,
+ run_id: str,
+ task_id: str,
+ key: str,
+ start: int | None,
+ stop: int | None,
+ step: int | None,
Review Comment:
These values are parsed from a `slice` object and passed in as variables anyway. Defaults aren’t useful, at least for what we use the API in this PR.
https://github.com/apache/airflow/pull/50117/files#diff-b16266575e3be5e029fe7cbe66e3c510e87e2a5b64db95c6c702b96898770e74R129-R143
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2108403069
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -470,7 +471,27 @@ def get_sequence_item(
},
)
raise
- return XComResponse.model_validate_json(resp.read())
+ return XComSequenceIndexResponse.model_validate_json(resp.read())
+
+ def get_sequence_slice(
+ self,
+ dag_id: str,
+ run_id: str,
+ task_id: str,
+ key: str,
+ start: int | None,
+ stop: int | None,
+ step: int | None,
Review Comment:
The problem is FastAPI does not handle None well in the query string. `{"start": None}` would just result in the backend receiving `start=None` and blowing up because it’s parsed as a literal string `"None"`. Providing defaults would not help here.
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2108395429
##########
airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_21.py:
##########
@@ -0,0 +1,37 @@
+# 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 Optional
+
+from cadwyn import VersionChange, schema
+from pydantic import Field
+
+from airflow.api_fastapi.execution_api.routes.xcoms import GetXComSliceFilterParams
+
+
+class RemoveGetXcomFilterParamsOffset(VersionChange):
+ """Remove 'offset' from GetXcomFilterParams."""
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version = (
+ schema(GetXComSliceFilterParams)
+ .field("offset")
+ .existed_as(type=Optional[int], info=Field(default=None)),
+ )
Review Comment:
I _think_ the idea of cadwyn is you should maintain backward compatibility in code, but remove the interface from spec with the migration? I’m honestly not sure.
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2113651506
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
Review Comment:
We should move the old tests under similar to airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_04_28/test_task_instances.py
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2115146147
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
Review Comment:
@kaxil since we are removing cadwyn migration here, shouldn't we be moving it to
`core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py` instead?
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2120257144
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
Review Comment:
The newly added tests are rightly present. No need to change there.
The older tests (https://github.com/apache/airflow/pull/50117/files#diff-5fb580f669f131661dde4f93ce870fc858935862892d0c751899c4e919001833L151-L156) that tested the older version of the API -- from inception till `v2025_04_28` will have to stay in `core/tests/unit/api_fastapi/execution_api/versions/v2025_04_28/test_task_instances.py`.
You do not need to create a new `v2025_06_10` folder at all.
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2108815876
##########
task-sdk/src/airflow/sdk/api/datamodels/_generated.py:
##########
@@ -25,9 +25,9 @@
from typing import Annotated, Any, Final, Literal
from uuid import UUID
-from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, JsonValue
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, JsonValue, RootModel
-API_VERSION: Final[str] = "2025-05-20"
+API_VERSION: Final[str] = "2025-05-21"
Review Comment:
10 June please. planning to cut release candidate on 3 June, with an aim to release it on 10 June
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2108286057
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -184,6 +188,132 @@ def get_xcom(
return XComResponse(key=key, value=result.value)
+@router.get(
+ "/{dag_id}/{run_id}/{task_id}/{key}/item/{offset}",
+ description="Get a single XCom value from a mapped task by sequence index",
+)
+def get_mapped_xcom_by_index(
+ dag_id: str,
+ run_id: str,
+ task_id: str,
+ key: str,
+ offset: int,
+ session: SessionDep,
+) -> XComSequenceIndexResponse:
+ xcom_query = XComModel.get_many(
+ run_id=run_id,
+ key=key,
+ task_ids=task_id,
+ dag_ids=dag_id,
+ session=session,
+ )
+ xcom_query = xcom_query.order_by(None)
+ if offset >= 0:
+ xcom_query = xcom_query.order_by(XComModel.map_index.asc()).offset(offset)
+ else:
+ xcom_query = xcom_query.order_by(XComModel.map_index.desc()).offset(-1 - offset)
+
+ if (result := xcom_query.limit(1).first()) is None:
+ message = (
+ f"XCom with {key=} {offset=} not found for task {task_id!r} in DAG run {run_id!r} of {dag_id!r}"
+ )
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail={"reason": "not_found", "message": message},
+ )
+ return XComSequenceIndexResponse(result.value)
+
+
+class GetXComSliceFilterParams(BaseModel):
+ """Class to house slice params."""
+
+ start: int | None = None
+ stop: int | None = None
+ step: int | None = None
+
+
+@router.get(
+ "/{dag_id}/{run_id}/{task_id}/{key}/slice",
+ description="Get a single XCom value from a mapped task by sequence slice",
+)
+def get_mapped_xcom_by_slice(
+ dag_id: str,
+ run_id: str,
+ task_id: str,
+ key: str,
+ params: Annotated[GetXComSliceFilterParams, Query()],
+ session: SessionDep,
+) -> XComSequenceSliceResponse:
+ orig_query = XComModel.get_many(
Review Comment:
```suggestion
xcom_query = XComModel.get_many(
```
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -184,6 +188,132 @@ def get_xcom(
return XComResponse(key=key, value=result.value)
+@router.get(
+ "/{dag_id}/{run_id}/{task_id}/{key}/item/{offset}",
+ description="Get a single XCom value from a mapped task by sequence index",
+)
+def get_mapped_xcom_by_index(
+ dag_id: str,
+ run_id: str,
+ task_id: str,
+ key: str,
+ offset: int,
+ session: SessionDep,
+) -> XComSequenceIndexResponse:
+ xcom_query = XComModel.get_many(
+ run_id=run_id,
+ key=key,
+ task_ids=task_id,
+ dag_ids=dag_id,
+ session=session,
+ )
+ xcom_query = xcom_query.order_by(None)
+ if offset >= 0:
+ xcom_query = xcom_query.order_by(XComModel.map_index.asc()).offset(offset)
+ else:
+ xcom_query = xcom_query.order_by(XComModel.map_index.desc()).offset(-1 - offset)
+
+ if (result := xcom_query.limit(1).first()) is None:
+ message = (
+ f"XCom with {key=} {offset=} not found for task {task_id!r} in DAG run {run_id!r} of {dag_id!r}"
+ )
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail={"reason": "not_found", "message": message},
+ )
+ return XComSequenceIndexResponse(result.value)
+
+
+class GetXComSliceFilterParams(BaseModel):
+ """Class to house slice params."""
+
+ start: int | None = None
+ stop: int | None = None
+ step: int | None = None
+
+
+@router.get(
+ "/{dag_id}/{run_id}/{task_id}/{key}/slice",
+ description="Get a single XCom value from a mapped task by sequence slice",
Review Comment:
```suggestion
description="Get XCom value(s) from a mapped task by sequence slice",
```
##########
airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_21.py:
##########
@@ -0,0 +1,37 @@
+# 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 Optional
+
+from cadwyn import VersionChange, schema
+from pydantic import Field
+
+from airflow.api_fastapi.execution_api.routes.xcoms import GetXComSliceFilterParams
+
+
+class RemoveGetXcomFilterParamsOffset(VersionChange):
+ """Remove 'offset' from GetXcomFilterParams."""
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version = (
+ schema(GetXComSliceFilterParams)
+ .field("offset")
+ .existed_as(type=Optional[int], info=Field(default=None)),
+ )
Review Comment:
We will also need to remove it here right? https://github.com/apache/airflow/pull/50117/files#diff-f7c9ab9ad4675e23743cffb626e7a9c4ce28ceffef5dd0b6b6a5268fe18909c5R133
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
+ pytest.param(-3, 200, "f", id="-3"),
+ pytest.param(-2, 200, "o", id="-2"),
+ pytest.param(-1, 200, "b", id="-1"),
+ pytest.param(0, 200, "f", id="0"),
+ pytest.param(1, 200, "o", id="1"),
+ pytest.param(2, 200, "b", id="2"),
Review Comment:
Why this change?
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -470,7 +471,27 @@ def get_sequence_item(
},
)
raise
- return XComResponse.model_validate_json(resp.read())
+ return XComSequenceIndexResponse.model_validate_json(resp.read())
+
+ def get_sequence_slice(
+ self,
+ dag_id: str,
+ run_id: str,
+ task_id: str,
+ key: str,
+ start: int | None,
+ stop: int | None,
+ step: int | None,
Review Comment:
Can we provide default values for these and handle it easier below by just assigning?
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2108787444
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
+ pytest.param(-3, 200, "f", id="-3"),
+ pytest.param(-2, 200, "o", id="-2"),
+ pytest.param(-1, 200, "b", id="-1"),
+ pytest.param(0, 200, "f", id="0"),
+ pytest.param(1, 200, "o", id="1"),
+ pytest.param(2, 200, "b", id="2"),
Review Comment:
Ack
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2108361158
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
+ pytest.param(-3, 200, "f", id="-3"),
+ pytest.param(-2, 200, "o", id="-2"),
+ pytest.param(-1, 200, "b", id="-1"),
+ pytest.param(0, 200, "f", id="0"),
+ pytest.param(1, 200, "o", id="1"),
+ pytest.param(2, 200, "b", id="2"),
Review Comment:
Because I changed the endpoint used in this test. This is because the runtime now uses the new endpoint.
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2111073536
##########
airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_21.py:
##########
@@ -0,0 +1,37 @@
+# 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 Optional
+
+from cadwyn import VersionChange, schema
+from pydantic import Field
+
+from airflow.api_fastapi.execution_api.routes.xcoms import GetXComSliceFilterParams
+
+
+class RemoveGetXcomFilterParamsOffset(VersionChange):
+ """Remove 'offset' from GetXcomFilterParams."""
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version = (
+ schema(GetXComSliceFilterParams)
+ .field("offset")
+ .existed_as(type=Optional[int], info=Field(default=None)),
+ )
Review Comment:
I think the problem is that older sdk clients would use the old endpoint when they do (e.g. `seq[0]`), so we need to keep the `offset` param working for them.
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr merged PR #50117:
URL: https://github.com/apache/airflow/pull/50117
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2115939278
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
Review Comment:
The new one is rightly in head, I was talking about the old ones -- i.e. the deleted code
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2119130626
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -148,12 +149,12 @@ def test_xcom_access_denied(self, client, caplog):
},
id="-4",
),
- pytest.param(-3, 200, {"key": "xcom_1", "value": "f"}, id="-3"),
- pytest.param(-2, 200, {"key": "xcom_1", "value": "o"}, id="-2"),
- pytest.param(-1, 200, {"key": "xcom_1", "value": "b"}, id="-1"),
- pytest.param(0, 200, {"key": "xcom_1", "value": "f"}, id="0"),
- pytest.param(1, 200, {"key": "xcom_1", "value": "o"}, id="1"),
- pytest.param(2, 200, {"key": "xcom_1", "value": "b"}, id="2"),
Review Comment:
I added the test under `v2025_06_10`. Not sure if it’s right.
--
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] Implement slice on LazyXComSequence [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #50117:
URL: https://github.com/apache/airflow/pull/50117#discussion_r2108806777
##########
airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_21.py:
##########
@@ -0,0 +1,37 @@
+# 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 Optional
+
+from cadwyn import VersionChange, schema
+from pydantic import Field
+
+from airflow.api_fastapi.execution_api.routes.xcoms import GetXComSliceFilterParams
+
+
+class RemoveGetXcomFilterParamsOffset(VersionChange):
+ """Remove 'offset' from GetXcomFilterParams."""
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version = (
+ schema(GetXComSliceFilterParams)
+ .field("offset")
+ .existed_as(type=Optional[int], info=Field(default=None)),
+ )
Review Comment:
If thats the case, we can probably have to handle the request structure in cadwyn for the get xcom endpoint
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -470,7 +471,27 @@ def get_sequence_item(
},
)
raise
- return XComResponse.model_validate_json(resp.read())
+ return XComSequenceIndexResponse.model_validate_json(resp.read())
+
+ def get_sequence_slice(
+ self,
+ dag_id: str,
+ run_id: str,
+ task_id: str,
+ key: str,
+ start: int | None,
+ stop: int | None,
+ step: int | None,
Review Comment:
Oh i see. Just trying to add a default so that every client caller doesn't have to manually pass in that information
##########
task-sdk/src/airflow/sdk/api/datamodels/_generated.py:
##########
@@ -25,9 +25,9 @@
from typing import Annotated, Any, Final, Literal
from uuid import UUID
-from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, JsonValue
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, JsonValue, RootModel
-API_VERSION: Final[str] = "2025-05-20"
+API_VERSION: Final[str] = "2025-05-21"
Review Comment:
This can be handled by the RM but since we know the release date for 3.0.2 (tentative), could we change this to 03 June?
##########
airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_21.py:
##########
@@ -0,0 +1,37 @@
+# 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 Optional
+
+from cadwyn import VersionChange, schema
+from pydantic import Field
+
+from airflow.api_fastapi.execution_api.routes.xcoms import GetXComSliceFilterParams
+
+
+class RemoveGetXcomFilterParamsOffset(VersionChange):
+ """Remove 'offset' from GetXcomFilterParams."""
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version = (
+ schema(GetXComSliceFilterParams)
+ .field("offset")
+ .existed_as(type=Optional[int], info=Field(default=None)),
+ )
Review Comment:
Actually now that I think of it, is offset relevant anymore for the GetXcom endpoint? We should safely be able to remove that change and make it only handle "single" map index tasks, right? -1 map index as param only.
--
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