jason810496 commented on code in PR #76825:
URL: https://github.com/apache/airflow/pull/76825#discussion_r4524511779
##########
kubernetes-tests/lang_sdk/go_example/main.go:
##########
@@ -60,12 +62,24 @@ func goExtract(actx airflow.Context) (any, error) {
}
// goTransform reads the my_variable Airflow variable through the coordinator,
-// exercising a GetVariable round-trip over the Execution API.
+// exercising a GetVariable round-trip over the Execution API, and the result
of the
+// mapped python_fanout task, which is every instance's value in a list.
func goTransform(actx airflow.Context) error {
val, err := actx.Client().GetVariable(actx, "my_variable")
if err != nil {
return err
}
actx.Logger().InfoContext(actx, "go_transform obtained variable",
"my_variable", val)
+
+ ti := actx.TaskInstance()
+ fanout, err := actx.Client().GetXCom(
+ actx, ti.DagID, ti.RunID, "python_fanout", nil,
sdk.XComReturnValueKey, nil,
+ )
+ if err != nil {
+ return err
+ }
+ if got := fmt.Sprint(fanout); got != "[1 2 3]" {
Review Comment:
Could we don't cast it to str to show the real type that the GetXCom will
return for the users?
##########
kubernetes-tests/lang_sdk/ts_example/src/main.ts:
##########
@@ -37,6 +37,15 @@ export async function tsExtract() {
export async function tsTransform() {
const value = await getClient().getVariable("my_variable");
console.log(`ts_transform obtained variable: ${value}`);
+
+ // python_fanout is mapped, so its result is every instance's value in a
list.
+ const fanout = await getClient().getXCom<number[]>({
+ key: "return_value",
+ taskId: "python_fanout",
+ });
+ if (JSON.stringify(fanout) !== "[1,2,3]") {
Review Comment:
Ditto, would it be better to assert against the real value?
##########
ts-sdk/src/sdk/client-types.ts:
##########
@@ -24,14 +24,32 @@ export type JsonValue =
/**
* Options for pulling an XCom value.
*
- * `dagId`, `taskId`, and `runId` default to the running task's context.
- * Pass them only when pulling an XCom value from another task or run.
+ * `dagId` and `runId` default to the running task's context. Pass them only
+ * when pulling an XCom value from another Dag or run.
*/
export interface GetXComOpts {
key: string;
Review Comment:
Could we also make the key on GetXCom here as optional which use the
"return_value" by default if it's unset? Or expose the constant of the
"return_value" as public API like Go?
Since I found the inconsistency between Java and TS that we make the key
field as mandatory in TS.
It would be nice to solve in this PR that aligning the GetXCom behavior
across SDKs, thanks.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -685,13 +704,56 @@ def get(
detail=e.detail,
status_code=e.response.status_code,
)
+ if map_index is None and
_reads_minus_one_for_omitted_map_index(e.response):
+ if values := self._get_all_for_omitted_map_index(
+ dag_id,
+ run_id,
+ task_id,
+ key,
+ include_prior_dates=include_prior_dates,
+ previous_iteration=previous_iteration,
+ ):
+ return XComResponse(key=key, value=values)
# Airflow 2.x just ignores the absence of an XCom and moves on
with a return value of None
# Hence returning with key as `key` and value as `None`, so
that the message is sent back to task runner
# and the default value of None in xcom_pull is used.
return XComResponse(key=key, value=None)
raise
return XComResponse.model_validate_json(resp.read())
+ def _get_all_for_omitted_map_index(
+ self,
+ dag_id: str,
+ run_id: str,
+ task_id: str,
+ key: str,
+ *,
+ include_prior_dates: bool,
+ previous_iteration: bool,
+ ) -> list[JsonValue]:
+ """
+ Read what a server that resolves an omitted map index would return for
a mapped task.
+
+ Only a mapped task has values here when map index -1 has none. A
mapped task that has no
+ values at all cannot be told apart from an unmapped task that pushed
none, so it reads
+ as not found instead of as an empty list.
+ """
+ try:
+ response = self.get_sequence_slice(
+ dag_id,
+ run_id,
+ task_id,
+ key,
+ None,
+ None,
+ None,
+ include_prior_dates,
+ previous_iteration=previous_iteration,
+ )
+ except ServerResponseError:
+ return []
Review Comment:
How about having a small helper for the case of `if e.response.status_code
== HTTPStatus.NOT_FOUND:`? Additionally, the non-404 errors are swallowed IIUC.
--
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]
