amoghrajesh commented on code in PR #46319:
URL: https://github.com/apache/airflow/pull/46319#discussion_r1937488045
##########
airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -104,6 +105,8 @@ def get_xcom(
return XComResponse(key=key, value=xcom_value)
+# TODO:once we have JWT tokens, then remove the dag/run/task ids from the URL
and just use the info in the
Review Comment:
```suggestion
# TODO: once we have JWT tokens, then remove the dag_id/run_id/task_id from
the URL and just use the info in the
```
##########
airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -184,13 +218,16 @@ def set_xcom(
return {"message": "XCom successfully set"}
-def has_xcom_access(xcom_key: str, token: TIToken) -> bool:
+def has_xcom_access(
+ dag_id: str, run_id: str, task_id: str, xcom_key: str, token: TIToken,
write: bool = False
Review Comment:
Can we make it `write_access: bool = False` to be in line with variables? Or
vice verse is fine too
##########
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:
Loading serdag is a route we should avoid if we can. Its super expensive and
doing it in an API is just going to slow down the response significantly
##########
airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -104,6 +105,8 @@ def get_xcom(
return XComResponse(key=key, value=xcom_value)
+# TODO:once we have JWT tokens, then remove the dag/run/task ids from the URL
and just use the info in the
Review Comment:
Nit
##########
airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -184,13 +218,16 @@ def set_xcom(
return {"message": "XCom successfully set"}
-def has_xcom_access(xcom_key: str, token: TIToken) -> bool:
+def has_xcom_access(
+ dag_id: str, run_id: str, task_id: str, xcom_key: str, token: TIToken,
write: bool = False
Review Comment:
Cool!
##########
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:
Might also affect tasks that have `execution_timeout` set
--
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]
