You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "henry3260 (via GitHub)" <gi...@apache.org> on 2026/03/25 17:34:26 UTC
[PR] Fix XCom PATCH/POST to store native values (instead of json.dumps str… [airflow]
henry3260 opened a new pull request, #64220:
URL: https://github.com/apache/airflow/pull/64220
closes: #59032
##### 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] Fix XCom PATCH/POST to store native values (instead of json.dumps) [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #64220:
URL: https://github.com/apache/airflow/pull/64220#discussion_r2999190093
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py:
##########
@@ -270,27 +269,24 @@ def create_xcom_entry(
)
try:
- value = json.dumps(request_body.value)
- except (ValueError, TypeError):
+ XComModel.set(
+ key=request_body.key,
+ value=request_body.value,
+ dag_id=dag_id,
+ task_id=task_id,
+ run_id=dag_run_id,
+ map_index=request_body.map_index,
+ serialize=False,
+ session=session,
+ )
+ except (ValueError, TypeError) as e:
raise HTTPException(
status.HTTP_400_BAD_REQUEST, f"Couldn't serialise the XCom with key: `{request_body.key}`"
- )
-
- new = XComModel(
- dag_run_id=dag_run.id,
- key=request_body.key,
- value=value,
- run_id=dag_run_id,
- task_id=task_id,
- dag_id=dag_id,
- map_index=request_body.map_index,
- )
- session.add(new)
- session.flush()
+ ) from e
xcom = session.scalar(
select(XComModel)
- .filter(
+ .where(
Review Comment:
Why this change?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py:
##########
@@ -340,16 +337,32 @@ def update_xcom_entry(
.limit(1)
.options(joinedload(XComModel.task), joinedload(XComModel.dag_run).joinedload(DR.dag_model))
)
+ xcom_entry = session.scalar(xcom_query)
if not xcom_entry:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
f"The XCom with key: `{xcom_key}` with mentioned task instance doesn't exist.",
)
- # Update XCom entry
- xcom_entry.value = json.dumps(patch_body.value)
+ try:
+ XComModel.set(
+ key=xcom_key,
+ value=patch_body.value,
+ dag_id=dag_id,
+ task_id=task_id,
+ run_id=dag_run_id,
+ map_index=patch_body.map_index,
+ serialize=False,
+ session=session,
+ )
+ except (ValueError, TypeError) as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST, f"Couldn't serialise the XCom with key: `{xcom_key}`"
+ ) from e
+ # Re-fetch after set (delete + insert) to get fresh object for response
Review Comment:
```suggestion
# Fetch after setting, to get fresh object for response
```
--
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] Fix XCom PATCH/POST to store native values (instead of json.dumps) [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #64220:
URL: https://github.com/apache/airflow/pull/64220#discussion_r3007729181
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py:
##########
@@ -270,27 +269,24 @@ def create_xcom_entry(
)
try:
- value = json.dumps(request_body.value)
- except (ValueError, TypeError):
+ XComModel.set(
+ key=request_body.key,
+ value=request_body.value,
+ dag_id=dag_id,
+ task_id=task_id,
+ run_id=dag_run_id,
+ map_index=request_body.map_index,
+ serialize=False,
+ session=session,
+ )
+ except (ValueError, TypeError) as e:
raise HTTPException(
status.HTTP_400_BAD_REQUEST, f"Couldn't serialise the XCom with key: `{request_body.key}`"
- )
-
- new = XComModel(
- dag_run_id=dag_run.id,
- key=request_body.key,
- value=value,
- run_id=dag_run_id,
- task_id=task_id,
- dag_id=dag_id,
- map_index=request_body.map_index,
- )
- session.add(new)
- session.flush()
+ ) from e
xcom = session.scalar(
select(XComModel)
- .filter(
+ .where(
Review Comment:
OK by me.
--
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] Fix XCom PATCH/POST to store native values (instead of json.dumps) [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #64220:
URL: https://github.com/apache/airflow/pull/64220#discussion_r2998985943
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py:
##########
@@ -843,5 +843,5 @@ def test_patch_xcom_entry_with_slash_key(self, test_client, session):
)
assert response.status_code == 200
assert response.json()["key"] == slash_key
- assert response.json()["value"] == json.dumps(new_value)
+ assert response.json()["value"] == new_value
check_last_log(session, dag_id=TEST_DAG_ID, event="update_xcom_entry", logical_date=None)
Review Comment:
Wonder could we add the test scenario described in https://github.com/apache/airflow/issues/59032 ?
For example, the existing XCom value type is `int` and after we patch the XCom with different value, it should still be `int` in the API response.
To make sure the type of the XCom value is as expected after calling the patch 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] Fix XCom PATCH/POST to store native values (instead of json.dumps) [airflow]
Posted by "henry3260 (via GitHub)" <gi...@apache.org>.
henry3260 commented on code in PR #64220:
URL: https://github.com/apache/airflow/pull/64220#discussion_r2999207137
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py:
##########
@@ -270,27 +269,24 @@ def create_xcom_entry(
)
try:
- value = json.dumps(request_body.value)
- except (ValueError, TypeError):
+ XComModel.set(
+ key=request_body.key,
+ value=request_body.value,
+ dag_id=dag_id,
+ task_id=task_id,
+ run_id=dag_run_id,
+ map_index=request_body.map_index,
+ serialize=False,
+ session=session,
+ )
+ except (ValueError, TypeError) as e:
raise HTTPException(
status.HTTP_400_BAD_REQUEST, f"Couldn't serialise the XCom with key: `{request_body.key}`"
- )
-
- new = XComModel(
- dag_run_id=dag_run.id,
- key=request_body.key,
- value=value,
- run_id=dag_run_id,
- task_id=task_id,
- dag_id=dag_id,
- map_index=request_body.map_index,
- )
- session.add(new)
- session.flush()
+ ) from e
xcom = session.scalar(
select(XComModel)
- .filter(
+ .where(
Review Comment:
> Why this change?
Hi @amoghrajesh, I updated this to align with the modern SQLAlchemy 2.0 syntax. While `.filter()` still works here, `.where()` is the canonical method when using `select()` constructs.
Let me know if you'd prefer I revert this part.
--
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] Fix XCom PATCH/POST to store native values (instead of json.dumps) [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh merged PR #64220:
URL: https://github.com/apache/airflow/pull/64220
--
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] Fix XCom PATCH/POST to store native values (instead of json.dumps) [airflow]
Posted by "henry3260 (via GitHub)" <gi...@apache.org>.
henry3260 commented on code in PR #64220:
URL: https://github.com/apache/airflow/pull/64220#discussion_r3000191968
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py:
##########
@@ -843,5 +843,5 @@ def test_patch_xcom_entry_with_slash_key(self, test_client, session):
)
assert response.status_code == 200
assert response.json()["key"] == slash_key
- assert response.json()["value"] == json.dumps(new_value)
+ assert response.json()["value"] == new_value
check_last_log(session, dag_id=TEST_DAG_ID, event="update_xcom_entry", logical_date=None)
Review Comment:
> Wonder could we add the test scenario described in #59032 ? For example, the existing XCom value type is `int` and after we patch the XCom with different value, it should still be `int` in the API response.
>
> To make sure the type of the XCom value is as expected after calling the patch API.
Add it! Thanks!
--
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