kaxil commented on code in PR #70070:
URL: https://github.com/apache/airflow/pull/70070#discussion_r3624661238
##########
providers/common/ai/src/airflow/providers/common/ai/plugins/hitl_review.py:
##########
@@ -121,25 +121,38 @@ def _read_xcom_by_prefix(
for key, value in session.execute(query).all():
suffix = key[len(prefix) :]
if suffix.isdigit():
- # deserialize_value expects an object with a .value attribute;
- # wrap the raw column value so we can reuse the standard
deserialization path.
- row = SimpleNamespace(value=value)
- result[int(suffix)] = XComModel.deserialize_value(row)
+ result[int(suffix)] = value
return result
def _write_xcom(
session: Session, *, dag_id: str, run_id: str, task_id: str,
map_index: int = -1, key: str, value
):
"""Write data to db."""
- XComModel.set(
- key=key,
- value=value,
- dag_id=dag_id,
- task_id=task_id,
- run_id=run_id,
- map_index=map_index,
- session=session,
+ # Stores value natively to match worker-written XComs; use
XComModel.set(serialize=False) once min Airflow >= 3.2.
+ dag_run_id = session.scalar(select(DagRun.id).where(DagRun.dag_id ==
dag_id, DagRun.run_id == run_id))
+ if dag_run_id is None:
+ raise HTTPException(404, f"DAG run not found on DAG {dag_id!r}
with ID {run_id!r}")
+ session.execute(
Review Comment:
We should also wait for @gopidesupavan to review the PR too
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
