ashb commented on code in PR #46319:
URL: https://github.com/apache/airflow/pull/46319#discussion_r1937506824
##########
airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -152,14 +170,30 @@ def set_xcom(
},
)
- if not has_xcom_access(key, token):
- raise HTTPException(
- status_code=status.HTTP_403_FORBIDDEN,
- detail={
- "reason": "access_denied",
- "message": f"Task does not have access to set XCom key
'{key}'",
- },
+ if mapped_length is not None:
+ task_map = TaskMap(
+ dag_id=dag_id,
+ task_id=task_id,
+ run_id=run_id,
+ map_index=map_index,
+ length=mapped_length,
+ keys=None,
)
+ max_map_length = conf.getint("core", "max_map_length", fallback=1024)
+ if task_map.length > max_map_length:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail={
+ "reason": "unmappable_return_value_length",
+ "message": "pushed value is too large to map as a
downstream's dependency",
+ },
+ )
+ session.add(task_map)
+
+ # else:
+ # TODO: Can/should we check if a client _hasn't_ provided this for an
upstream of a mapped task? That
+ # means loading the serialized dag and that seems like a relatively costly
operation for minimal benefit
+ # (the mapped task would fail in a moment as it can't be expanded anyway.)
Review Comment:
Yeah exactly.
--
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-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
