You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ashb (via GitHub)" <gi...@apache.org> on 2025/10/08 16:48:27 UTC
[PR] Simple sending of return value from Go tasks to XCom [airflow]
ashb opened a new pull request, #56481:
URL: https://github.com/apache/airflow/pull/56481
This right now is very simple and will need to evolve over time -- it doens't
support XCom backends for instance.
In order to call this HTPT API correctly we needed to update the client generator.
--
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] Simple sending of return value from Go tasks to XCom [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #56481:
URL: https://github.com/apache/airflow/pull/56481#discussion_r2414838345
##########
go-sdk/sdk/client.go:
##########
@@ -88,3 +86,57 @@ func (*client) GetConnection(ctx context.Context, connID string) (Connection, er
return connFromAPIResponse(resp)
}
+
+func (c *client) PutXComForTi(
+ ctx context.Context,
+ ti api.TaskInstance,
+ key string,
+ value any,
+) error {
+ return c.PutXCom(ctx, ti.DagId, ti.RunId, ti.TaskId, ti.MapIndex, key, value)
+}
+
+func (*client) PutXCom(
+ ctx context.Context,
+ dagId, runId, taskId string,
+ mapIndex *int,
+ key string,
+ value any,
+) error {
+ params := api.SetXcomParams{}
+
+ if mapIndex != nil && *mapIndex != -1 {
+ params.MapIndex = mapIndex
+ }
+
+ httpClient := ctx.Value(sdkcontext.ApiClientContextKey).(api.ClientInterface)
+ _, err := httpClient.Xcoms().SetResponse(ctx, dagId, runId, taskId, key, ¶ms, &value)
+ if err != nil {
+ return err
+ }
+ return nil
+}
Review Comment:
Done
--
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] Simple sending of return value from Go tasks to XCom [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #56481:
URL: https://github.com/apache/airflow/pull/56481#discussion_r2414780623
##########
go-sdk/sdk/client.go:
##########
@@ -88,3 +86,57 @@ func (*client) GetConnection(ctx context.Context, connID string) (Connection, er
return connFromAPIResponse(resp)
}
+
+func (c *client) PutXComForTi(
+ ctx context.Context,
+ ti api.TaskInstance,
+ key string,
+ value any,
+) error {
+ return c.PutXCom(ctx, ti.DagId, ti.RunId, ti.TaskId, ti.MapIndex, key, value)
+}
+
+func (*client) PutXCom(
+ ctx context.Context,
+ dagId, runId, taskId string,
+ mapIndex *int,
+ key string,
+ value any,
+) error {
+ params := api.SetXcomParams{}
+
+ if mapIndex != nil && *mapIndex != -1 {
+ params.MapIndex = mapIndex
+ }
+
+ httpClient := ctx.Value(sdkcontext.ApiClientContextKey).(api.ClientInterface)
+ _, err := httpClient.Xcoms().SetResponse(ctx, dagId, runId, taskId, key, ¶ms, &value)
+ if err != nil {
+ return err
+ }
+ return nil
+}
Review Comment:
Merge into one? Or we could make one of them private for instance
##########
go-sdk/sdk/client.go:
##########
@@ -88,3 +86,57 @@ func (*client) GetConnection(ctx context.Context, connID string) (Connection, er
return connFromAPIResponse(resp)
}
+
+func (c *client) PutXComForTi(
+ ctx context.Context,
+ ti api.TaskInstance,
+ key string,
+ value any,
+) error {
+ return c.PutXCom(ctx, ti.DagId, ti.RunId, ti.TaskId, ti.MapIndex, key, value)
+}
+
+func (*client) PutXCom(
+ ctx context.Context,
+ dagId, runId, taskId string,
+ mapIndex *int,
+ key string,
+ value any,
+) error {
+ params := api.SetXcomParams{}
+
+ if mapIndex != nil && *mapIndex != -1 {
+ params.MapIndex = mapIndex
+ }
+
+ httpClient := ctx.Value(sdkcontext.ApiClientContextKey).(api.ClientInterface)
+ _, err := httpClient.Xcoms().SetResponse(ctx, dagId, runId, taskId, key, ¶ms, &value)
+ if err != nil {
+ return err
+ }
+ return nil
+}
+
+func (*client) GetXCom(
+ ctx context.Context,
+ dagId, runId, taskId string,
+ mapIndex *int,
+ key string,
+ value any,
+) (any, error) {
+ params := api.GetXcomParams{
+ MapIndex: mapIndex,
+ }
+
+ httpClient := ctx.Value(sdkcontext.ApiClientContextKey).(api.ClientInterface)
+ res, err := httpClient.Xcoms().Get(ctx, dagId, runId, taskId, key, ¶ms)
+ if err != nil {
+ var httpError *api.GeneralHTTPError
+ if errors.As(err, &httpError) && httpError.Response.StatusCode() == 404 {
+ err = fmt.Errorf("%w: %q", XComNotFound, key)
+ }
+ return nil, err
+ }
+ // TODO: We probably want to do some level of xcom deser here
Review Comment:
Yes, can be taken care of while handling xcom backends.
##########
go-sdk/bundle/bundlev1/task.go:
##########
@@ -74,15 +77,26 @@ func (f *taskFunction) Execute(ctx context.Context, logger *slog.Logger) error {
}
}
// If there are two results, convert the first only if it's not a nil pointer
- var res any
if len(retValues) > 1 && (retValues[0].Kind() != reflect.Ptr || !retValues[0].IsNil()) {
- res = retValues[0].Interface()
+ res := retValues[0].Interface()
+ f.sendXcom(ctx, res, sdkClient, logger)
}
- // TODO: send the result to XCom
- _ = res
return err
}
+func (f *taskFunction) sendXcom(
+ ctx context.Context,
+ value any,
+ c sdk.XComClient,
+ logger *slog.Logger,
+) {
+ workload := ctx.Value(sdkcontext.WorkloadContextKey).(api.ExecuteTaskWorkload)
+ err := c.PutXComForTi(ctx, workload.TI, api.XComReturnValueKey, value)
Review Comment:
```suggestion
err := c.SetXCom(ctx, workload.TI, api.XComReturnValueKey, value)
```
nit?
--
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] Simple sending of return value from Go tasks to XCom [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb merged PR #56481:
URL: https://github.com/apache/airflow/pull/56481
--
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] Simple sending of return value from Go tasks to XCom [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #56481:
URL: https://github.com/apache/airflow/pull/56481#discussion_r2414796673
##########
go-sdk/sdk/client.go:
##########
@@ -88,3 +86,57 @@ func (*client) GetConnection(ctx context.Context, connID string) (Connection, er
return connFromAPIResponse(resp)
}
+
+func (c *client) PutXComForTi(
+ ctx context.Context,
+ ti api.TaskInstance,
+ key string,
+ value any,
+) error {
+ return c.PutXCom(ctx, ti.DagId, ti.RunId, ti.TaskId, ti.MapIndex, key, value)
+}
+
+func (*client) PutXCom(
+ ctx context.Context,
+ dagId, runId, taskId string,
+ mapIndex *int,
+ key string,
+ value any,
+) error {
+ params := api.SetXcomParams{}
+
+ if mapIndex != nil && *mapIndex != -1 {
+ params.MapIndex = mapIndex
+ }
+
+ httpClient := ctx.Value(sdkcontext.ApiClientContextKey).(api.ClientInterface)
+ _, err := httpClient.Xcoms().SetResponse(ctx, dagId, runId, taskId, key, ¶ms, &value)
+ if err != nil {
+ return err
+ }
+ return nil
+}
Review Comment:
Oh yes, I was somehow jumping ahead and confusing myself, push/set for another TI doesn't make sense, but pull for another TI we need the distinction.
--
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] Simple sending of return value from Go tasks to XCom [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #56481:
URL: https://github.com/apache/airflow/pull/56481#discussion_r2414469604
##########
go-sdk/pkg/worker/task.go:
##########
Review Comment:
This was an (unused) old copy of go-sdk/bundle/bundlev1/task.go -- I removed it here to avoid it being out of date and still confusing.
--
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] Simple sending of return value from Go tasks to XCom [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #56481:
URL: https://github.com/apache/airflow/pull/56481#discussion_r2414837612
##########
go-sdk/bundle/bundlev1/task.go:
##########
@@ -74,15 +77,26 @@ func (f *taskFunction) Execute(ctx context.Context, logger *slog.Logger) error {
}
}
// If there are two results, convert the first only if it's not a nil pointer
- var res any
if len(retValues) > 1 && (retValues[0].Kind() != reflect.Ptr || !retValues[0].IsNil()) {
- res = retValues[0].Interface()
+ res := retValues[0].Interface()
+ f.sendXcom(ctx, res, sdkClient, logger)
}
- // TODO: send the result to XCom
- _ = res
return err
}
+func (f *taskFunction) sendXcom(
+ ctx context.Context,
+ value any,
+ c sdk.XComClient,
+ logger *slog.Logger,
+) {
+ workload := ctx.Value(sdkcontext.WorkloadContextKey).(api.ExecuteTaskWorkload)
+ err := c.PutXComForTi(ctx, workload.TI, api.XComReturnValueKey, value)
Review Comment:
I've gont for `c.PushXCom` actually
##########
go-sdk/bundle/bundlev1/task.go:
##########
@@ -74,15 +77,26 @@ func (f *taskFunction) Execute(ctx context.Context, logger *slog.Logger) error {
}
}
// If there are two results, convert the first only if it's not a nil pointer
- var res any
if len(retValues) > 1 && (retValues[0].Kind() != reflect.Ptr || !retValues[0].IsNil()) {
- res = retValues[0].Interface()
+ res := retValues[0].Interface()
+ f.sendXcom(ctx, res, sdkClient, logger)
}
- // TODO: send the result to XCom
- _ = res
return err
}
+func (f *taskFunction) sendXcom(
+ ctx context.Context,
+ value any,
+ c sdk.XComClient,
+ logger *slog.Logger,
+) {
+ workload := ctx.Value(sdkcontext.WorkloadContextKey).(api.ExecuteTaskWorkload)
+ err := c.PutXComForTi(ctx, workload.TI, api.XComReturnValueKey, value)
Review Comment:
I've gone for `c.PushXCom` actually
--
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